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

Getting Started