In this article, we will need to build an API by using the Falcon API Framework in Python. We're going to cover some hints and methods to request. This article is part of a series:
1) Bootstrapping Falcon API Framework on Docker
2) Creating an API by using individual Falcon methods. (You're here.)
You had better read the previous article or you will be confused about that. Right, let's start.
My root directory is the following. First of all, I've created the requirements.txtfile.
attrs==19.1.0
falcon==2.0.0
falcon-autocrud==1.0.36
gunicorn==19.9.0
jsonschema==3.0.1
marshmallow==2.19.5
psycopg2==2.8.3
pyrsistent==0.15.4
python-dateutil==2.8.0
six==1.12.0
SQLAlchemy==1.3.6
webargs==5.4.0
It's ready to fetch the libraries by this command.
pip install -r requirements.txt
After arranging the libraries, we can create the Python files. My main file is app.py. In spite of this, you can define another name.
import falcon
from resources import athlete
from resources import plan
from resources import exercise
from services import database_service
from middlewares import (
ContentEncodingMiddleware,
)
conn = database_service.connect()
api = falcon.API(middleware=[
ContentEncodingMiddleware(),
])
# api = falcon.API()
athlete = athlete.Athlete(conn, database_service)
plan = plan.Plan(conn, database_service)
exercise = exercise.Exercise(conn, database_service)
api.add_route('/athletes/{id}', athlete)
api.add_route('/athletes', athlete, suffix='collection')
api.add_route('/plans/{id}', plan)
api.add_route('/plans', plan, suffix='collection')
api.add_route('/exercises/{id}', exercise)
api.add_route('/exercises', exercise, suffix='collection')
This file provides Falcon app to bootstrap. I added my routes and related files into this file. Now, as you see, there's resourcedirectory. This directory ships the my objecys. I've 3 objects to ensure as the resources. First one is plan.py.
import falcon
from webargs import fields
from webargs.falconparser import use_args
class Plan(object):
post_request_args = {"name": fields.Str(required=True), "description": fields.Str(required=True),
"difficulty": fields.Int(required=True), "athlete_id": fields.Int(required=True)}
def __init__(self, conn, database_service):
self.conn, self.database_service, self.resource_name = conn, database_service, self.__class__.__name__
def on_delete(self, req, resp, id):
try:
q = " ".join(
["DELETE", "FROM", self.resource_name.lower(), "WHERE", self.resource_name.lower() + "_id = %s"])
q_resp = self.database_service.run_delete_query(self.conn, q, [id])
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
output = {"status": True, "message": self.resource_name + " was deleted successfully!", "data": None}
resp.status = falcon.HTTP_200
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
def on_get(self, req, resp, id):
try:
cur = self.conn.cursor()
q = " ".join(
["SELECT", "*", "FROM", self.resource_name.lower(), "wHERE", self.resource_name.lower() + "_id = %s"])
q_resp = self.database_service.run_get_query(cur, q, [id])
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
output = {"status": True, "message": None,
'data': self.database_service.set_columns(q_resp['data'], cur)}
resp.status = falcon.HTTP_200
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
def on_get_collection(self, req, resp):
try:
cur = self.conn.cursor()
q = " ".join(
["SELECT * FROM", self.resource_name.lower()])
q_resp = self.database_service.run_get_query(cur, q, [])
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
output = {"status": True, "message": None,
'data': self.database_service.set_columns(q_resp['data'], cur)}
resp.status = falcon.HTTP_200
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
def on_put(self, req, resp, id):
try:
cur = self.conn.cursor()
# q = "SELECT name, description, difficulty FROM " + self.resource_name.lower() + " WHERE " + self.resource_name.lower() + "_id = %s;"
get_q = " ".join(
["SELECT name,description,difficulty FROM", self.resource_name.lower(), "wHERE",
self.resource_name.lower() + "_id = %s"])
get_resp = self.database_service.run_get_query(cur, get_q, [id])
record = list(self.database_service.set_columns(get_resp['data'], cur))[0]
request = req.media
for index in record.keys():
if index in request.keys():
record[index] = request[index]
record['id'] = id
update_q = " ".join(
["UPDATE", self.resource_name.lower(), "SET name=%s, description=%s, difficulty=%s WHERE",
self.resource_name.lower() + "_id=%s RETURNING ", self.resource_name.lower() + "_id;"])
update_resp = self.database_service.run_upsert_query(self.conn, update_q, record.values())
if not update_resp['status']:
output = {"status": True, "message": update_resp['message'],
"data": None}
else:
response_data = {
"id": update_resp['data'],
"name": record['name'],
"description": record['description'],
"difficulty": record['difficulty']
}
output = {"status": True, "message": self.resource_name + " is updated successfully!",
"data": response_data}
resp.status = falcon.HTTP_201
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
@use_args(post_request_args)
def on_post_collection(self, req, resp, args):
try:
# q = "INSERT INTO " + self.resource_name.lower() + " (name, description, difficulty, athlete_id) VALUES (%s,%s,%s,%s) RETURNING " + self.resource_name.lower() + "_id;"
q = " ".join(
["INSERT INTO", self.resource_name.lower(),
"(name, description, difficulty, athlete_id) VALUES (%s,%s,%s,%s) RETURNING",
self.resource_name.lower() + "_id;"])
params = {'name': args['name'], 'description': args['description'], 'difficulty': args['difficulty'],
'athlete_id': args['athlete_id']}
q_resp = self.database_service.run_upsert_query(self.conn, q, params.values())
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
response_data = {
"id": q_resp['data'],
"name": args['name'],
"description": args['description'],
"difficulty": args['difficulty'],
"athlete_id": args['athlete_id']
}
output = {"status": True, "message": self.resource_name + " is added successfully!",
"data": response_data}
resp.status = falcon.HTTP_201
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
I'm going to share only this file as a resource. Because other files are exact same according to code structure.
You'll see the prefix in front of the methods called on_and HTTP verbs come after this prefix. We understand, each method represents the HTTP verbs.
According to this, we'll have these endpoints;
1-) http://localhost:8000/plans (GET, POST, PUT, DELETE)
Perfection! It's straightforward to bootstrap. We can pass the configuration step. Therefore, I've created a conf directory. It ships two files. config.ini and gunicorn_conf.py. The ini file is like this.
[postgresqlDB]
host = postgresql
db = test_dev
user = mertingen
pass = mertingen
That's configuration file for web service.
import multiprocessing
bind = '0.0.0.0:8000'
workers = multiprocessing.cpu_count() * 2 + 1
timeout = 30
worker_connections = 1000
I've also servicesdirectory to handle some useful processes. That's the create_schema_service.py
import psycopg2
import database_service
conn = database_service.connect()
def create_tables():
""" create tables in the PostgreSQL database"""
commands = (
"""
CREATE TABLE IF NOT EXISTS athlete (
athlete_id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
email VARCHAR(255) NOT NULL,
phone VARCHAR(255) NOT NULL,
gender VARCHAR(255) NOT NULL,
birthday DATE NOT NULL
)
""",
""" CREATE TABLE IF NOT EXISTS plan (
plan_id SERIAL PRIMARY KEY,
athlete_id INTEGER NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT NOT NULL,
difficulty VARCHAR(255) NOT NULL,
FOREIGN KEY (athlete_id)
REFERENCES athlete (athlete_id)
ON UPDATE CASCADE ON DELETE CASCADE
)
""",
"""
CREATE TABLE IF NOT EXISTS exercise (
exercise_id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NOT NULL
)
"""
)
try:
cur = conn.cursor()
for c in commands:
cur.execute(c)
print("Table was created successfully!")
cur.close()
conn.commit()
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
if __name__ == '__main__':
create_tables()
And, database_service.py
import psycopg2 as pg
import configparser as cp
import os
dir_path = os.path.dirname(os.path.realpath(__file__))
c = cp.ConfigParser()
c.read(dir_path + '/../conf/config.ini')
def connect():
try:
connection = pg.connect(user=c['postgresqlDB']['user'],
password=c['postgresqlDB']['pass'],
host=c['postgresqlDB']['host'],
port="5432",
database=c['postgresqlDB']['db'])
print("You are connected!")
return connection
except (Exception, pg.Error) as error:
print("Error while connecting to PostgreSQL", error)
# finally:
# if connection:
# connection.close()
# print("PostgreSQL connection is closed")
def set_columns(data, cur):
items = []
if data:
for x in data:
item = {}
c = 0
for col in cur.description:
item.update({col[0]: x[c]})
c = c + 1
items.append(item)
return items
else:
return []
def run_get_query(cur, query, params):
try:
if params:
cur.execute(query, tuple(params))
else:
cur.execute(query)
records = cur.fetchall()
return {"status": True, "message": "", "data": records}
except pg.InternalError as e:
return {"status": False, "message": str(e), "data": None}
def run_upsert_query(conn, q, params):
try:
cur = conn.cursor()
cur.execute(q, tuple(params))
conn.commit()
id = cur.fetchone()[0]
return {"status": True, "message": "", "data": id}
except pg.InternalError as e:
conn.rollback()
return {"status": False, "message": str(e), "data": None}
def run_delete_query(conn, q, params):
try:
cur = conn.cursor()
cur.execute(q, tuple(params))
conn.commit()
return {"status": True, "message": "", "data": None}
except pg.InternalError as e:
conn.rollback()
return {"status": False, "message": str(e), "data": None}
Lastly, I'd rather provide my resources as a JSON format. For this, I've created middleware in the root directory for Falcon.
import json
from datetime import date, datetime
class JSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, bytes):
return obj.decode("utf-8")
if isinstance(obj, (date, datetime)):
return str(obj.isoformat())
return super(JSONEncoder, self).default(obj)
class ContentEncodingMiddleware(object):
def process_response(self, req, resp, _, req_succeeded):
if not req_succeeded:
return
if req.client_accepts_json:
resp.set_header('Content-Type', 'application/json')
resp.body = json.dumps(resp.body, cls=JSONEncoder)
Let's create a plan by Postman.
That's it!
You'll able to find the codes and related repository on this link.
https://github.com/mertingen/python-falcon-framework-api
Falcon framework builds pretty-cool APIs. I hope, this article will be useful for you and see you in the next article. If you have any trouble, don't hesitate to ask.
Top comments (0)