Simple REST API using Flask and Peewee

Lately I started experimenting with REST APIs using Python’s flask and Peewee.
Note :- I am writing a tutorial for the first time, so if you find any issues please let me know.

What is REST API?
The REST APIs provide programmatic access to read and write data.

The characteristics of a REST system are defined by six design rules:

  • Client-Server: There should be a separation between the server that offers a service, and the client that consumes it.
  • Stateless: Each request from a client must contain all the information required by the server to carry out the request. In other words, the server cannot store information provided by the client in one request and use it in another request.
  • Cacheable: The server must indicate to the client if requests can be cached or not.
  • Layered System: Communication between a client and a server should be standardized in such a way that allows intermediaries to respond to requests instead of the end server, without the client having to do anything different.
  • Uniform Interface: The method of communication between a client and a server must be uniform.
  • Code on demand: Servers can provide executable code or scripts for clients to execute in their context. This constraint is the only one that is optional.

Before jumping into the next section i would recommend going through intro to rest.

What is covered in this post

  1. peewee — ORM tool
  2. flask — python based microservice

Structure of REST API
we will be using cities data to build our rest api. The data is stored in postgres database.
Three api endpoints we will be building:

  1. http://127.0.0.1:5000/api/v1/cities
  2. http://127.0.0.1:5000/api/v1/cities/<country_code>
  3. http://127.0.0.1:5000/api/v1/cities/<country_code>/<city_name>

Peewee

peewee is a simple and small ORM. It is easy to use and gets the job done. 
Why do we need ORM?
ORMs helps us to communicate with the underlying database in a clean and efficient way. It circumvent the need to write sql queries and makes use of inbuilt functions.
Whats the tradeoff?
ORMs are slow when compared to native database drivers as native drivers use sql queries, where as ORMs need to convert given input into sql queries.

First we need to create a class representing particular table in the database.

Configure database engine.

psql_db = PostgresqlDatabase(
'db_name',
user='username',
password='password',
host='localhost',
)

Note:- Create Database before hand, table will be created using peewee.

Now create a base model of the table which contains Meta configuration, it is passed on to subclasses, so our project’s models will all subclass BaseModel.

class BaseModel(Model):

class Meta:
database = psql_db

class
City(BaseModel):
id = PrimaryKeyField(null=False)
name = CharField(max_length=35)
countrycode = CharField(max_length=3)
district = CharField(max_length=20)
population = BigIntegerField()

Since we will be transmitting data over network and resurrected on the same or a different system, we need to serialize the data in order to preserve the state of it. This is achieved by adding serialize property to the City class.

@property
def serialize(self):
data = {
'id': self.id,
'name': str(self.name).strip(),
'countrycode': str(self.countrycode).strip(),
'district': str(self.district).strip(),
'population': self.population,
}

return data

def __repr__(self):
return "{}, {}, {}, {}, {}".format(
self.id,
self.name,
self.countrycode,
self.district,
self.population
)

As we defined the structure for the table, now we will create it in the database.

# create table
psql_db.create_table(City, safe=True)

with psql_db.atomic():
# insert data
for city in cities:
track = dict()
track["name"]= city[1]
track["countrycode"] = city[2]
track["district"] = city[3]
track["population"] = city[4]

City.create(**track)

create_table function takes in the class defined to create the table and the second argument(optional) safe=True checks if the table is present or not, if not then it creates the table. The next part is the suggested a way to insert bulk records into database. This will create the table and insert data into it.

REST API

As described previously we will be implementing three endpoints to it.

City Endpoint — http://127.0.0.1:5000/api/v1/cities

City endpoint supports both GET and POST methods on it. GET method has two routs to access it. Since the records are paginated you can call the GET request with specific page number or without specifying page number. Number of items for a page are set to 10.

@app.route('/api/v1/cities', methods=['GET', 'POST'])
@app.route('/api/v1/cities/<int:page>', methods=['GET'])
def city_endpoint(page=1):

# get request
if request.method == 'GET':
per_page = 10
query = model.City.select().paginate(page, per_page)
data = [i.serialize for i in query]

if data:
res = jsonify({
'cities': data,
'meta': {
'page': page,
'per_page': per_page,
'page_url': request.url}
})
res.status_code = 200
else:
# if no results are found.
output = {
"error": "No results found. Check url again",
"url": request.url,
}
res = jsonify(output)
res.status_code = 404
return res

peewee comes with an option to paginated the records requested. Once the records are retuned then you can serialize the records. Then check if any records were returned or not, if not then return appropriate error message. If data is present then use flask jsonify to convert the data into Response object with application/json mimetype and add respective status code before returning it.

elif request.method == 'POST':  # post request

row = model.City.create(**request.json)
query = model.City.select().where(
model.City.name == row.name,
model.City.district == row.district
)
data = [i.serialize for i in query]
res = jsonify({
'city': data,
'meta': {'page_url': request.url}
})
res.status_code = 201
return res

check if the request is POST, then insert the data into database by using create function which take a dictionary as input key — column name and value is the data to be inserted into database.

In order to send POST, PUT and DELETE requests you can make use of curl or postman. Send POST request using postman.

{
"countrycode": "CN",
"district": "BELLWOOD",
"name": "BEN10",
"population": 237500
}

id need not be specified, peewee takes care of it.

City/Country Endpoint — http://127.0.0.1:5000/api/v1/cities/<country_code>

This endpoint supports only GET request and take accepts country_code as input parameter. Then filters all records based on the country code.

@app.route('/api/v1/cities/<string:country_code>', methods=['GET'])
@app.route('/api/v1/cities/<string:country_code>/<int:page>', methods=['GET'])
def city_country_endpoint(country_code, page=1):

# get request
if request.method == 'GET':
per_page = 10
query = model.City.select().where(model.City.countrycode == country_code).paginate(page, per_page)
data = [i.serialize for i in query]

if data:
res = jsonify({
'cities': data,
'meta': {'page': page, 'per_page': per_page, 'page_url': request.url}
})
res.status_code = 200
else:
output = {
"error": "No results found. Check url again",
"url": request.url,
}
res = jsonify(output)
res.status_code = 404
return res

City/Country/Name Endpoint — http://127.0.0.1:5000/api/v1/cities/<country_code>/<city_name>

This end point supports GET, PUT and DELETE. The third endpoint accepts country code and city name as input parameters.

Incase of PUT request.

elif request.method == "PUT":  # put endpoint
c = model.City.get(
model.City.countrycode == country_code,
model.City.name == city_name
)

if not c:
abort(404)
if not request.json:
abort(400)

if 'district' in request.json and type(request.json['district']) != str:
abort(400)
else:
c.district = request.json['district']
if 'population' in request.json and type(request.json['population']) is not int:
abort(400)
else:
c.population = request.json['population']

c.save()

query = model.City.select().where(
model.City.name == c.name,
model.City.countrycode == c.countrycode
)
data = [i.serialize for i in query]
res = jsonify({
'city': data,
'meta': {'page_url': request.url}
})
res.status_code = 200
return res

get specific record from database and then validated the incoming request. If the incoming request satisfies all the requirements then update population and/or district name of the city. Once the values are updated calling save on the city object c will update specific record in database. The PUT request to the end point is as below.

{
"district": "bellwood",
"population": 341831456
}
or 
{
"district": "bellwood",
}
or
{
"population": 341831456
}

Incase of DELETE request.

elif request.method == "DELETE":  # delete endpoint

try:
city = model.City.get(
model.City.countrycode == country_code,
model.City.name == city_name
)
except:
city = None

if
city:
city.delete_instance()
res = jsonify({})
res.status_code = 204
return res
else:
res = jsonify({
"Error": "The requested resource is no longer available at the "
"server and no forwarding address is known."
,
"Status Code": 410,
"URL": request.url
})
res.status_code = 410
return res

Get the city based on the input parameters passed to the endpoint, if city not found return error with status code 410. If city found then delete it by calling delete_instance on the city object and return empty response with status code of 210.

This concludes the tutorial which performs GET, PUT, POST and DELETE operations using the data in postgres. Complete code can be found here. First run data.py to create table and insert data, then run app.py.

Suggestions:-

I would suggest to use postman to send requests to REST API.
Go through all the reading to get better understanding of REST API.

Next
Implement RESTful authentication using flask.

One clap, two clap, three clap, forty?

By clapping more or less, you can signal to us which stories really stand out.