connexion¶
connexion is a package that allows you to define Flask APIs and how endpoints are fulfilled using an OpenAPI specification. This has the advantage that a UI is automatically generated and data that is passed to endpoints is validated based on the OpenAPI specification. By combining connexion with OpenAlchemy, the OpenAPI specification not only defines the Flask application but also the SQLALchemy models.
See also
- connexion documentation
Documentation for connexion.
API specification¶
As the case for combining connexion with OpenAlchemy is compelling, an example application has been included here. The API is designed to keep track of employees of a company. The OpenAPI specification is:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | openapi: "3.0.0"
info:
title: Test Schema
description: API to illustrate OpenAlchemy.
version: "1.0"
paths:
/employee:
get:
operationId: api.search
summary: Used to retrieve all employees.
responses:
200:
description: Return all employees from the database.
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/Employee"
post:
operationId: api.post
summary: Used to save an employee to the database.
requestBody:
description: The employee to save.
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/Employee"
responses:
200:
description: Save successful.
400:
description: The Employee already exists.
/employee/{id}:
parameters:
- in: path
name: id
schema:
type: integer
required: true
get:
operationId: api.get
summary: Used to retrieve an Employee from the database.
responses:
200:
description: The Employee.
content:
application/json:
schema:
$ref: "#/components/schemas/Employee"
404:
description: The Employee was not found.
patch:
operationId: api.patch
summary: Update an Employee in the database.
requestBody:
description: The employee to save.
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/Employee"
responses:
200:
description: The Employee was updated.
404:
description: The Employee was not found.
delete:
operationId: api.delete
summary: Delete an Employee from the database.
responses:
200:
description: The Employee was deleted.
404:
description: The Employee was not found.
components:
schemas:
Employee:
description: Person that works for a company.
type: object
x-tablename: employee
properties:
id:
type: integer
description: Unique identifier for the employee.
example: 0
x-primary-key: true
x-autoincrement: true
name:
type: string
description: The name of the employee.
example: David Andersson
x-index: true
division:
type: string
description: The part of the company the employee works in.
example: Engineering
x-index: true
salary:
type: number
description: The amount of money the employee is paid.
example: 1000000.00
required:
- id
- name
- division
|
There is a POST endpoint for inserting new employees into the database, a GET endpoint for all employees and for a specific employee by id, there is a PATCH endpoint for updating an employee and a DELETE endpoint for deleting an employee.
Database Models¶
The OpenAPI specification above includes a schemas section which defines the Employee schema. This schema is used to construct the Employee model. The database setup is defined in the following file:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | """Setup for the database."""
import os
from flask_sqlalchemy import SQLAlchemy
from open_alchemy import init_yaml
# Construct models
db = SQLAlchemy()
SPEC_DIR = os.path.dirname(__file__)
SPEC_FILE = os.path.join(SPEC_DIR, "api.yaml")
MODELS_FILENAME = os.path.join(SPEC_DIR, "models_autogenerated.py")
init_yaml(SPEC_FILE, base=db.Model, models_filename=MODELS_FILENAME)
|
Endpoint Fulfillment¶
The API endpoints are all fulfilled in the api.py file which has a function for each combination of path and method:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | """Functions handling API endpoints."""
import database
import models_autogenerated as models
def search():
"""Get all employees from the database."""
employees = models.Employee.query.all()
employee_dicts = map(lambda employee: employee.to_dict(), employees)
return list(employee_dicts)
def post(body):
"""Save an employee to the database."""
if models.Employee.query.filter_by(id=body["id"]).first() is not None:
return ("Employee already exists.", 400)
employee = models.Employee.from_dict(**body)
database.db.session.add(employee)
database.db.session.commit()
def get(id):
"""Get an employee from the database."""
employee = models.Employee.query.filter_by(id=id).first()
if employee is None:
return ("Employee not found.", 404)
return employee.to_dict()
def patch(body, id):
"""Update an employee in the dayabase."""
employee = models.Employee.query.filter_by(id=id).first()
if employee is None:
return ("Employee not found.", 404)
employee.name = body["name"]
employee.division = body["division"]
employee.salary = body["salary"]
database.db.session.commit()
return 200
def delete(id):
"""Delete an employee from the database."""
result = models.Employee.query.filter_by(id=id).delete()
if not result:
return ("Employee not found.", 404)
database.db.session.commit()
return 200
|
App Construction¶
All the steps needed to make the Flask app work are defined in the app.py file:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | """Application code."""
import connexion
import database
# Creating Flask app
app = connexion.FlaskApp(__name__, specification_dir=".")
# Initializing database
app.app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
database.db.init_app(app.app)
with app.app.app_context():
database.db.create_all()
# Running app
app.add_api("api.yaml")
app.run(port=8080)
|
Conclusion¶
The duplication of the data schema has been reduced by defining the SQLAlchemy models based on the OpenAPI specification. This means that, to change the database schema, the OpenAPI specification has to be updated and vice-versa. This ensures that the two are always in synch and up to date.
See also