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