Basic API development with Flask

As a data engineer, you will undoubtedly need to create API’s from time to time. An API gives programmatic access to your data to query, insert, delete etc.

Setting up an API is super simple! We can use the FLASK Python framework. Let’s walk through an example. In this example, we’ll be looking at tasks on a to-do list – we will create, query and delete those tasks as part of this example.

First, let’s get some libraries installed:

pip3 install flask
pip3 install flask_restful

Now let’s import all the things we might need and setup a flask app.

from flask import Flask, request
from flask_restful import Api, Resource, reqparse, abort
myapp = Flask(__name__)
myapi = Api(myapp)

So now we get into the detail. What do we want the user to be able to do? Well, we want them to be able to make a request to a URL to GET, PUT or DELETE data related to our tasks.

So first, let’s create an empty dictionary called ‘tasks’ to store all of our task data and create a class called task and create three methods within it.

tasks = {}

First, we have a get method. This is essentially a query – get the data related to a particular task. So, when the user passes taskid to the method, it’ll return the details of that task in json format.

class task(Resource): 
    def get(self, taskid):
        out = tasks[taskid]
        return {'found task':out}

Next, the put request. This takes the taskid and stores the data related to it – I’ll show how to send that payload with the task ID a little bit later on.

    def put(self, taskid):
        args = task_put_args.parse_args()
        tasks[taskid] = args
        return tasks[taskid]

Finally, we have the delete function. This will remove the given task ID from the dictionary of tasks.

    def delete(self, taskid):
        tasks.pop(taskid)
        return 'deleted', 204

Now we define an endpoint for the API. When the user sends /task/2, it will GET, POST or DELETE task 2 from the dictionary of tasks.

myapi.add_resource(task, "/task/<int:taskid>")

When the user PUT’s a new task in, we want to make sure they provide us with all the information we require. So, we can define some required arguments along with some help text. This ensures that every task entered into the dictionary has a name, estimate and owner defined.

task_put_args = reqparse.RequestParser()
task_put_args.add_argument("name", type=str, help="You must assign a task name", required=True)
task_put_args.add_argument("estimate", type=int, help="You must assign an estimate in hours (int)", required=True)
task_put_args.add_argument("owner", type=str, help="You must assign a task owner", required=True)

So there we are. We have our API. How how do we call it? We simply import the requests and json libraries and set our base URL for the API.

import requests
import json
base_url = "http://127.0.0.1:5000/"

To make a PUT request, we must define the payload (the three required fields) and then make a put request to our base URL along with the api endpoint information

payload = json={'name': 'This is my task', 'estimate': 30, 'owner': 'kieran'}
headers = {'accept': 'application/json'}
response = requests.put(base_url + 'task/4', json=payload)
print(response.json())

This returns: {‘name’: ‘This is my task’, ‘estimate’: 30, ‘owner’: ‘kieran’}

To make a GET request, we simply pass the base URL along with the endpoint information (/task/taskid).

response = requests.get(base_url + 'task/4')
print(response.json())

This returns: {‘data’: {‘name’: ‘This is my task’, ‘estimate’: 30, ‘owner’: ‘kieran’}}

And finally, to make a delete request, we pass the base URL along with the endpoint information (/task/taskid).

response = requests.delete(base_url + 'task/4')
print(response)

this returns: <Response [204]>

Full API code

Below is the full code. Debug should not equal true for production systems!

from flask import Flask, request
from flask_restful import Api, Resource, reqparse, abort
myapp = Flask(__name__)
myapi = Api(myapp)
#Mandatory arguments in API put request
task_put_args = reqparse.RequestParser()
task_put_args.add_argument("name", type=str, help="You must assign a task name", required=True)
task_put_args.add_argument("estimate", type=int, help="You must assign an estimate in hours (int)", required=True)
task_put_args.add_argument("owner", type=str, help="You must assign a task owner", required=True)
tasks = {}
class task(Resource): 
    def get(self, taskid):
        out = tasks[taskid]
        return {'found task':out} 
    def put(self, taskid):
        args = task_put_args.parse_args()
        tasks[taskid] = args
        return tasks[taskid]
    def delete(self, taskid):
        tasks.pop(taskid)
        return 'deleted', 204
myapi.add_resource(task, "/task/<int:taskid>") 
if __name__ == "__main__":
    myapp.run(debug=True)
Share the Post:

Related Posts