Flask 4 - REST APIs met Connexion

Auteur

Fabrice Devaux

Publicatiedatum

16 december 2025

Inleiding

We bouwden een eenvoudige Flask applicatie voor een blog website. De browser stuurt HTTP GET of POST requests naar de applicatie en deze voert de nodige acties uit en stuurt een HTML pagina terug.

Maar wat als we…

Dezelfde dienst willen aanbieden als een mobile app?
Willen integreren met andere diensten (bvb. om automatisch een blog post aan te maken)?
Met een dynamisch en interactief frontend framework zoals React willen werken?

Voor die situaties is het handiger dat de blog “server” gestructureerde data, zoals JSON stuurt en ontvangt. De verschillende clients kunnen dan de data elk op hun eigen manier verwerken en presenteren.

REST API

De server wordt dan een API of Application Programming Interface, een verzameling afspraken (regels en functies) waarmee softwareonderdelen met elkaar kunnen praten. Die onderdelen kunnen lokaal binnen eenzelfde systeem of programma zijn – bvb. de verzameling standard functies en classes in Python. Maar die onderdelen kunnen ook aan een typisch client-server model voldoen, zoals een web-API waarmee blog posts kunnen worden opgehaald, aangemaakt of gewijzigd.

Er is steeds een client die communiceert met een API server.
De resources (in dit voorbeeld blogpost) hebben elk een unieke identificatie (1, 2, …).
Elke request van een client wordt apart behandeld door de API server, zonder verband met vorige requests van dezelfde client. We zeggen dat de server stateless is.

Deze eigenschappen vormen de basis voor wat gekend is als REST, of REpresentational State Transer.

Doorgaans wordt als onderliggend protocol steeds HTTP gebruikt en is JSON het formaat gebruikt om data (de resources) uit te wisselen.

JSON

JSON, of JavaScript Object Notation is een manier om data op een gestructureerde en leesbare manier voor te stellen.

Onze blogpost uit onze Flask applicatie heeft bijvoorbeeld altijd een unieke ID, een titel en een inhoud.

{
  "id": 1,
  "title": "Hello World!",
  "body": "Today I used Flask, and it was quite nice.\nI like it a lot."
}

Python ondersteunt JSON rechtstreeks!

import json

post = {
    "id": 1,
    "title": "Hello World!",
    "body": "Today I used Flask, and it was quite nice.\nI like it a lot.",
}

with open("post.json", mode="w", encoding="utf-8") as file:
1    json.dump(post, file)

with open("post.json", mode="r", encoding="utf-8") as file:
2    loaded_post = json.load(file)

print(type(loaded_post))
print(loaded_post)

# Output:
# <class 'dict'>
# {'id': 1, 'title': 'Hello World!', 'body': 'Today I used Flask, and it was quite nice.\nI like it a lot.'}

1: Met json.dump kunnen we een basis Python type (dict, list, …) omzetten naar JSON en opslaan in een bestand.
2: Omgekeerd kunnen we met json.load JSON data uit een bestand inlezen.

REST API met de Flask App

Wat als we onze bestaande Flask applicatie willen aanpassen naar een REST API?

Maak een nieuwe /posts route, op basis van de bestaande index view. Bij REST APIs wordt veelal gesproken over een endpoint i.p.v. een route.
We gebruiken geen Jinja2 (dus geen render_template).
Als een view een list of dict returnt zet Flask dit automatisch om in JSON.

Oplossing

flaskr/blog.py

@bp.route("/posts")
def posts():
    db = get_db()
    posts = db.execute(
        "SELECT p.id, title, body, created, author_id, username "
        "FROM post p JOIN user u ON p.author_id = u.id "
        "ORDER BY created DESC"
    ).fetchall()

    return {"posts": [dict(post) for post in posts]}

We bekijken wat we precies teruggestuurd krijgen, met de browser developer tools, of bvb. met curl:

curl -v http://127.0.0.1:5000/posts
*   Trying 127.0.0.1:5000...
* Connected to 127.0.0.1 (127.0.0.1) port 5000
> GET /posts HTTP/1.1
> Host: 127.0.0.1:5000
> User-Agent: curl/8.7.1
> Accept: */*
>
* Request completely sent off
< HTTP/1.1 200 OK
< Server: Werkzeug/3.1.3 Python/3.13.5
< Date: Sun, 28 Sep 2025 09:00:39 GMT
< Content-Type: application/json
< Content-Length: 191
< Vary: Cookie
< Connection: close
<
{
  "posts": [
    {
      "author_id": 1,
      "body": "world",
      "created": "Tue, 23 Sep 2025 06:12:15 GMT",
      "id": 1,
      "title": "hello",
      "username": "fab"
    }
  ]
}

OpenAPI

OAS

De OpenAPI Specification (OAS) is een standaard formaat om REST API’s te beschrijven. Het is eigenlijk een soort contract dat vastlegt:

welke endpoints er zijn
welke parameters en request bodies verwacht worden
welke responses terugkomen

Met zo een contract weten gebruikers (clients) van de API precies hoe ze de deze moeten gebruiken en wat ze kunnen verwachten.

Dit gebeurt meestal in een YAML of JSON bestand. Zo’n beschrijving is zowel leesbaar voor mensen (documentatie) als bruikbaar voor machines (bijv. codegeneratie, validatie, testen).

Hieronder krijg je als voorbeeld een voorproefje van hoe de Flaskr blog app uit de vorige les er zou uitzien als een API, beschreven als een OpenAPI specification. We komen hier later in meer detail op terug.

openapi: 3.0.3
info:
  title: Flaskr Blog API
  version: 1.0.0
  description: API for the Flaskr blog (create, read, update, delete posts)

servers:
  - url: http://localhost:5000
    description: Local development server

paths:
  /posts:
    get:
      summary: List all posts
      responses:
        '200':
          description: A list of posts
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: '#/components/schemas/Post'
    post:
      summary: Create a new post
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/PostInput'
      responses:
        '201':
          description: Post created
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Post'

  /posts/{postId}:
    parameters:
      - name: postId
        in: path
        required: true
        schema:
          type: integer
    get:
      summary: Retrieve a specific post
      responses:
        '200':
          description: The requested post
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Post'
        '404':
          description: Post not found
    put:
      summary: Update an existing post
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/PostInput'
      responses:
        '200':
          description: Updated post
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Post'
        '404':
          description: Post not found
    delete:
      summary: Delete a post
      responses:
        '204':
          description: Post deleted (no content)
        '404':
          description: Post not found

components:
  schemas:
    Post:
      type: object
      properties:
        id:
          type: integer
        title:
          type: string
        body:
          type: string
      required:
        - id
        - title
        - body

    PostInput:
      type: object
      properties:
        title:
          type: string
        body:
          type: string
      required:
        - title
        - body

Swagger

De OpenAPI specification stond vroeger bekend als Swagger specification en is oorspronkelijk ontwikkeld door het bedrijf Smartbear. De specificatie is in tussen 2015 en 2017 afgesplitst als een afzonderlijk open initiatief. De term Swagger spec wordt nog veel gebruikt maar eigenlijk is het dus OAS. Meer informatie vind je op https://swagger.io/blog/api-strategy/difference-between-swagger-and-openapi/.

Smartbear ontwikkelt wel nog een reeks tools onder de noemer Swagger waarvan we er twee aan bod laten komen.

Swagger UI laat via een interactieve web pagina toe OpenAPI specs te visualiseren en te testen. Een demo vind je op https://petstore.swagger.io/.

Swagger Editor is een online editor waar je een OAS in kan plakken en bewerken. Je krijgt live de bijhorende Swagger UI te zien. Probeer dit zeker uit met de Flask Blog spec die we hiervoor hebben bekeken.

Connexion Introductie

Wat is Connexion

Connexion is een web framework ontwikkeld rond 2015 door Zalando (ja, die Zalando), met als doel het ondersteunen van een API-first of Spec-first methodologie. Hierbij wordt een toekomstige API eerst volledige gedefinieerd met een OpenAPI spec. (Meer over API-first in deze blogpost van Zalando zelf, of deze van Atlassian)

Connexion bouwt boven op Flask en biedt als belangrijkste features:

Het linken van route functies aan operations (path + method) in de OAS
Automatische validatie van requests parameters en body (op basis van de OAS)
Parameter parsing en injectie (geen flask.request object nodig)
Automatisch correct omzetten van respons objecten
Swagger UI integratie
Authenticatie

Zoals we in dit hoofdstuk verder zullen zien wordt het met Connexion een stuk gemakkelijker een REST API te maken dan wanneer we alles zelf zouden moeten doen met enkel Flask.

Mocking

Connexion komt met een CLI waarmee je onder andere zonder extra code een OpenAPI spec beschikbaar kan maken via een test server. Dit kan bijzonder handig zijn om toekomstige API clients alvast de overeengekomen OAS te laten valideren.

Voorbeeld:

connexion run https://raw.githubusercontent.com/spec-first/connexion/main/examples/helloworld_async/spec/openapi.yaml --mock=all

Quotes API

Intro

Als eerste kennismaking met Connexion bouwen we samen een eenvoudige API om bekende quotes te beheren. We starten met een enkel endpoint:

GET /quotes/random - stuurt een random quote terug

De OpenAPI spec ziet er als volgt uit:

quotes/openapi.yaml

openapi: 3.0.0
info:
  title: Quotes API
  version: 1.0.0

1paths:
  /quotes/random:
2    get:
3      operationId: quotes.app.get_random_quote
4      summary: Get a random quote
5      responses:
        "200":
          description: A single random quote
          content:
6            application/json:
              schema:
                $ref: "#/components/schemas/Quote"

7components:
8  schemas:
9    Quote:
      type: object
      properties:
        id:
          type: integer
        text:
          type: string
      required: [id, text]

1: Onder paths komen alle unieke paden. We starten met enkel /quotes/random.
2: Onder elk pad komen één of meerdere HTTP methods die beschikbaar zijn voor dit pad. We starten met enkel get.
3: operationId bepaald de view functie voor deze operation (= combinatie van path + method). In dit geval de function get_random_quote in the quotes/app.py module.
4: summary geeft een korte beschrijving van wat deze operation doet.
5: responses bevat één of meerdere HTTP status codes. Voor elke code geeft de description een korte beschrijving van wat je mag verwachten in dit geval, en de content de exacte beschrijving van de structuur van de respons.
6: In dit geval bevat de respons (body) JSON data. Het schema van die data kan ofwel rechtstreeks beschreven worden, ofwel verwijzen naar een latere schema definitie. In dit geval kiezen we voor de tweede optie omdat we het schema van “een quote” ook in andere operations zullen gebruiken.
7: components is net zoals paths een top-level element. Het is optioneel en kan verschillende componenten bevatten die gebruikt worden in de operations.
8: In dit geval bevat components enkel schemas. Elk schema beschrijft een element waar elders naar verwezen kan worden.
9: We starten met een enkel schema met de (vrij te kiezen) naam Quote. Een Quote is een object (kan je zien als het JSON equivalent van een dict) en bevat een id en een text. Voor elk element (“property”) in het object geven we het type mee (hier int en string). Optioneel kunnen we ook definiëren welke elementen verplicht aanwezig moeten zijn, met required.

We starten een nieuw project en installeren Connexion:

uv
conda

mkdir quotes
cd quotes

uv init
uv add "connexion[flask,swagger-ui,uvicorn]~=3.0"

mkdir quotes
cd quotes

conda create -n quotes
conda activate quotes
conda install "connexion[flask,swagger-ui,uvicorn]~=3.0"

conda activate quotes

De OpenAPI spec plaatsen we in een quotes folder, waar zo meteen ook de Python code komt te staan.

./quotes
├── quotes/
    └── openapi.yaml

quotes/app.py

import random

import connexion

QUOTES = [
    {"id": 1, "text": "Premature optimization is the root of all evil. - Donald Knuth"},
    {"id": 2, "text": "Talk is cheap. Show me the code. - Linus Torvalds"},
    {"id": 3, "text": "Good code is its own best documentation. - Steve McConnell"},
    {"id": 4, "text": "Make it work, make it right, make it fast. - Kent Beck"},
]


1def get_random_quote():
    return random.choice(QUOTES)


2app = connexion.FlaskApp(__name__)
3app.add_api("openapi.yaml")

if __name__ == "__main__":
4    app.run("quotes.app:app", port=5000)

1: get_random_quote is the view functie en komt overeen met de operationId in the OpenAPI spec. De functie returnt een random gekozen dict. Net als Flask zet Connexion dit automatisch om naar JSON in de HTTP respons, en gebruikt default status code 200.
2: Een Connexion app object maken werkt op dezelfde manier als bij Flask. Dit object stelt de volledige applicatie voor.
3: Met add_api kan een OpenAPI spec ingeladen worden. Dit kan herhaald worden als er verschillende specs zijn.
4: app.run start een eenvoudige test server om te applicatie via HTTP toegankelijk te maken.

Testen en Swagger UI

Om de applicatie te starten:

$ python -m quotes.app

curl is een bijzonder populaire all-around CLI HTTP client. Hiermee kunnen we eenvoudige requests naar onze API sturen.

$ curl http://127.0.0.1:5000/quotes/random
{
  "id": 4,
  "text": "Make it work, make it right, make it fast. - Kent Beck"
}

We krijgen, zoals verwacht, een JSON respons met een van de voorgeprogrammeerde quotes.

Zoals eerder aangehaald bestaat er ook een zogenaamde Swagger UI, een webpagina waarmee de API bekeken en gebruikt kan worden. Omdat we Connexion geïnstalleerd hebben met de swagger-ui is deze UI automatisch beschikbaar via http://127.0.0.1:5000/ui.

Oefeningen

De volgende oefeningen voegen extra operaties toe.

Alle quotes opvragen

Met de GET /quotes operatie willen we de lijst met alle quotes terugkrijgen.

Om te testen gebruik je het volgende curl commando:

$ curl http://127.0.0.1:5000/quotes

Tips

In de OAS krijgen we een nieuw path: /quotes, met daarin een enkele method: get
In response komt opnieuw een "200" maar het schema zal nu een lijst van Quote objecten zijn.

Dit schema kan je als volgt voorstellen in de OAS:

              schema:
                type: array
                items:
                  $ref: "#/components/schemas/Quote"

Oplossing

quotes/openapi.yaml

paths:
  /quotes:
    get:
      operationId: quotes.app.get_all_quotes
      summary: List all quotes
      responses:
        "200":
          description: A list of quotes
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: "#/components/schemas/Quote"

flaskr/app.py

def get_all_quotes():
    return QUOTES

Een specifieke quote opvragen

Met GET /quotes/{id} willen we enkel de quote met de gegeven id terugkrijgen. Als de id niet bestaat willen we een HTTP 404 respons met een foutmelding.

Om te testen gebruik je het volgende curl commando:

$ curl http://127.0.0.1:5000/quotes/1

$ curl http://127.0.0.1:5000/quotes/9

Tips

Een path met een variabel deel wordt voorgesteld met /quotes/{id}
Die id is dan een parameter. Parameters worden in een parameters lijst (onder de method, dus op hetzelfde niveau als summary en responses) gedefinieerd.

Documentatie: - https://spec.openapis.org/oas/latest.html#path-templating - https://spec.openapis.org/oas/latest.html#parameter-object

Deze id parameter definieer je als volgt:

      parameters:
        - in: path
          name: id
          schema:
            type: integer
          required: true
          description: The ID of the quote

Oplossing

quotes/openapi.yaml

paths:

  /quotes/{id}:
    get:
      operationId: quotes.app.get_quote_by_id
      summary: Get a quote by ID
      parameters:
        - in: path
          name: id
          schema:
            type: integer
          required: true
          description: The ID of the quote
      responses:
        "200":
          description: A single quote
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/Quote"
        "404":
          description: Quote not found

flaskr/app.py

def get_quote_by_id(id: int):
    for quote in QUOTES:
        if quote["id"] == id:
            return quote

    return {"title": "Not Found", "detail": f"Quote with id {id} not found"}, 404

Een quote toevoegen

Met de POST /quotes operatie moet een nieuwe quote worden toegevoegd. Hiervoor voegen we gewoon een item toe aan de QUOTES lijst (die toegevoegde quotes gaan dus verloren als de server herstart). In de request verwachten we een JSON object met enkel het text veld. In de respons willen we de volledige quote die is toegevoegd, inclusief de id.

Om te testen gebruik je het volgende curl commando:

$ curl -X 'POST' 'http://127.0.0.1:5000/quotes'-H 'Content-Type: application/json' -d '{"text": "Python is an experiment in how much freedom programmers need. - Guido van Rossum"}'

Tips

Een JSON object in de request body definieer je requestBody, onder de method, dus op hetzelfde niveau als summary en responses.

Documentatie: https://spec.openapis.org/oas/latest.html#request-body-object

      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/NewQuote"

NewQuote is dus een extra schema dat we ook moeten definiëren. Het is eigenlijk hetzelfde als de bestaande Quote, maar zonder de id.

Oplossing

quotes/openapi.yaml

paths:

    post:
      operationId: quotes.app.add_quote
      summary: Add a new quote
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/NewQuote"
      responses:
        "201":
          description: Quote created
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/Quote"

  schemas:

    NewQuote:
      type: object
      properties:
        text:
          type: string
      required:
        - text

flaskr/app.py

def add_quote(body: str):
    new_id = max(q["id"] for q in QUOTES) + 1 if QUOTES else 1
    new_quote = {"id": new_id, "text": body["text"]}
    QUOTES.append(new_quote)
    return new_quote, 201

We kunnen nu ook demonstreren hoe Connexion automatisch de request data voor ons valideert:

$ curl -X 'POST' 'http://127.0.0.1:5000/quotes' -H 'Content-Type: application/json' -d '{"text": 123}'
{"type": "about:blank", "title": "Bad Request", "detail": "123 is not of type 'string' - 'text'", "status": 400}

$ curl -X 'POST' 'http://127.0.0.1:5000/quotes' -H 'Content-Type: application/json' -d '{"txt": "Test"}'
{"type": "about:blank", "title": "Bad Request", "detail": "'text' is a required property", "status": 400

$ curl -X 'POST' 'http://127.0.0.1:5000/quotes' -H 'Content-Type: text/plain' -d '{"text": "test"}'
{"type": "about:blank", "title": "Unsupported Media Type", "detail": "Invalid Content-type (text/plain), expected ['application/json']", "status": 415}

Authenticatie

Connexion voorziet verschillende opties om een API request de authentificeren. We bekijken hier als voorbeeld een eenvoudige HTTP Basic Authentication. In de Connextion documentatie vind je alle andere mogelijkheden.

In de OAS zijn 2 toevoegingen nodig:

quotes/openapi.yaml

paths:
  /quotes:
    post:
      operationId: quotes.app.add_quote
      summary: Add a new quote
      requestBody: ...
      responses: ...
1      security:
        - basic: []

components:
2  securitySchemes:
    basic:
      type: http
      scheme: basic
      x-basicInfoFunc: quotes.app.basic_auth

1: In elke operation waarvoor we authenticatie willen gebruiken voegen we een security block toe dat verwijst naar een securityScheme. In dit geval willen we enkel de posten van een nieuwe quote beveiligen.
2: Onder components voegen we securitySchemes toe. We definiëren hier maar een schema met de naam basic (moet overeenstemmen met de naam gebruikt in stap 1). De x-basicInfoFunc parameter verwijst naar de Python functie die de authenticatie zal uitvoeren.

quotes/app.py

PASSWD = {"admin": "secret", "foo": "bar"}


def basic_auth(username: str, password: str):
    if PASSWD.get(username) == password:
        return {"sub": username}
    return None

Dit is natuurlijk een simpel en onveilig voorbeeld. In de praktijk zou bijvoorbeeld een password-database gebruikt worden, met enkel een hash van de wachtwoorden.

Een test, zonder en dan met authenticatie:

$ curl -X 'POST' -H 'Content-Type: application/json' 'http://127.0.0.1:5000/quotes' -d '{"text": "test"}'
{"type": "about:blank", "title": "Unauthorized", "detail": "No authorization token provided", "status": 401}

$ curl -X 'POST' -u admin:secret -H 'Content-Type: application/json' 'http://127.0.0.1:5000/quotes' -d '{"text": "test"}'
# Of:
$ curl -X 'POST' -H 'Authorization: Basic YWRtaW46c2VjcmV0' -H 'Content-Type: application/json' 'http://127.0.0.1:5000/quotes' -d '{"text": "test"}'
{
  "id": 5,
  "text": "test"
}

YWRtaW46c2VjcmV0 is de Base64 encoding van de string admin:secret.

Flaskr-API

In dit hoofdstuk bouwen we onze Flaskr blog website om naar een REST API. We bouwen hier verder op de implementatie met SQLAlchemy, uit les 3. Heb je geen werkende code, dan kan je gebruik maken van https://github.com/dvx76/flask-tutorial/tree/sqlalchemy.

Tip: maak een nieuw Git branch in je flaskr project.

Voor we kunnen starten moeten we natuurlijk het Connexion package installeren.

uv add "connexion[flask,swagger-ui,uvicorn]~=3.0"

Blog-post Representatie

Eerste moeten we definiëren hoe we een blog-post gaan voorstellen in JSON. Één blog-post bestaat uit een aantal velden, waardoor het in JSON de type object zal krijgen. Voor elke blog-post hebben we deze velden:

id is een integer en bevat de unieke ID van de post
author is een string en bevat de naam van de auteur
created is een timestamp (date) maar zal in JSON voorgesteld worden als een string
title is een string
body is een string

Op basis hiervan kunnen we onze OpenAPI Spec starten met een schema voor een post.

Belangrijk

Het is zeker niet noodzakelijk om object schemas in de OpenAPI Spec te laten overeenstemmen met het database schema van de applicatie.

Bij het opmaken van de OAS bepaal je op welke manier je informatie will representeren voor de gebruikers van de API. Zo kiezen we er hier voor om de username van de post auteur terug te sturen, en niet de user_id. We combineren en selecteren dus data uit 2 verschillende database tabellen.

Het is zelfs aangeraden om de API schemas volledig los te koppelen van interne representaties (zoals een database schema). Eenmaal in gebruik is een API immers moeilijk aan te passen. Maar we willen mogelijks wel de interne representatie aanpassen.

flaskr/openapi.yaml

openapi: 3.0.3
info:
  title: Flaskr Blog API
  version: 1.0.0
  description: API for the Flaskr blog

components:
  schemas:
    Post:
      type: object
      properties:
        id:
          type: integer
        author:
          type: string
1        created:
          type: string
          format: date-time
        title:
          type: string
        body:
          type: string

1: De created timestamp is dus een string type maar met format: date-time maken we duidelijk dat deze string een timestamp voorstelt.

Eerste operatie: GET all posts

Om te starten implementeren we GET /posts. We willen een lijst terugkrijgen met alle bestaande posts.

Een eerste path in de OAS:

flaskr/openapi.yaml

paths:
  /posts:
    get:
      summary: List all posts
      operationId: flaskr.app.get_all_posts
      responses:
        "200":
          description: A list of posts
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: "#/components/schemas/Post"

Voor de implementatie in flaskr/app.py kunnen we ons grotendeels baseren op het vorig quotes project en index view uit onze Flask code.

We starten met de volgende code en werken samen als oefening de get_all_posts functie af:

flaskr/app.py

import connexion
from sqlalchemy import select
from sqlalchemy.orm import joinedload

from .db import create_db_session
from .models import Post

app = connexion.FlaskApp(__name__)
app.add_api("openapi.yaml")

1db_session, remove_session = create_db_session(None)
2app.app.teardown_appcontext(remove_session)

def get_all_posts():
    # Verder af te werken

if __name__ == "__main__":
    app.run("flaskr.app:app", port=5000)

1: Om een SQLAlchemy Session te verkrijgen kunnen we de bestaande functie uit de db module gebruiken. We laten de functie de default database_url gebruiken (= SQLite). Voor de eenvoud en omdat we hier uit een enkel bestand werken gebruiken we db_session als globale variabele en hoeven we in de route functies geen gebruik de maken van get_db_session.
2: app is nu een connexion.FlaskApp object, en geen flask.Flask object. Via .app hebben we wel nog toegang tot het onderliggende Flask object. Op die manier kunnen we alsnog de teardown_appcontext instellen.

Oplossing

flaskr/app.py

import connexion
from sqlalchemy import select
from sqlalchemy.orm import joinedload

from .db import create_db_session
from .models import Post

app = connexion.FlaskApp(__name__)
app.add_api("openapi.yaml")

db_session, remove_session = create_db_session(None)
app.app.teardown_appcontext(remove_session)

def get_all_posts():
    posts = db_session.scalars(
1        select(Post).options(joinedload(Post.author)).order_by(Post.created.desc())
    )
    return [
2        {
            "id": post.id,
            "author": post.author.username,
            "created": post.created,
            "title": post.title,
            "body": post.body,
        }
        for post in posts
    ]

if __name__ == "__main__":
    app.run("flaskr.app:app", port=5000)

1: We gebruiken joinedload omdat we voor elke Post ook de author.username willen gebruiken.
2: We zetten “handmatig” het resultaat van de ORM query om naar de gewenste API respons. Connexion zorgt voor het serializeren naar JSON.

Een blog-post toevoegen

Dit wordt de POST /posts operation. Om een nieuwe blog-post te maken hebben we enkel title en body nodig. We voegen hiervoor een extra schema toe in de OAS.

flaskr/openapi.yaml

paths:
  /posts:
    post:
      summary: Create a new post
      operationId: flaskr.app.create_post
      requestBody:
1        required: true
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/NewPost"
      responses:
2        "201":
          description: Post created
          content:
            application/json:
              schema:
3                $ref: "#/components/schemas/Post"

components:
  schemas:
    NewPost:
      type: object
      properties:
        title:
          type: string
        body:
          type: string
      required:
        - title
        - body

1: In de POST request verwachten we een body in JSON formaat volgens het NewPost schema.
2: HTTP 201 - Created, we bevestigen aan de client dat de post resource is aangemaakt.
3: Bij REST APIs is het gebruikelijk om bij een POST alle details van de aangemaakte resource terug te sturen (m.a.w. hetzelfde antwoord als bij een GET voor die specifieke resource)

Voor de bijhorende functie starten we opnieuw vanuit de overeenkomstige Flask view.

flaskr/app.py

1def create_post(body: dict):
2    post = Post(title=body["title"], body=body["body"], author_id=1)
    db_session.add(post)
3    db_session.commit()
    return {
        "id": post.id,
        "author": post.author.username,
        "created": post.created,
        "title": post.title,
        "body": post.body,
    }

1: Connexion valideert automatisch de JSON body van de request, en zet deze om naar een passend Python type - in dit geval een dict.
2: We hebben een author_id (of author) nodig maar hebben (nog) geen authenticatie ingebouwd. Als tijdelijke oplossing hardcoden we de id van een bestaande user.
3: Eenmaal het Post object gecommit is naar de database zal SQLAlchemy ORM automatisch het object in de session updaten, bvb met de id.

Om de nieuwe operation te testen gebruiken we het volgende curl commando:

curl -X 'POST' 'http://127.0.0.1:5000/posts' -H 'Content-Type: application/json' -d '{"title": "Post from curl", "body": "Somebody"}

Oefeningen

De volgende oefeningen voegen extra operaties toe.

Specifieke blog-post opvragen

Dit wordt de GET /posts/{id} operatie. Bedenkt wat er moet gebeuren als de gevraagde post niet bestaat.

Oplossing

flaskr/openapi.yaml

paths:
  /posts/{id}:
    get:
      summary: Get a single post
      operationId: flaskr.app.get_post
      parameters:
        - in: path
          name: id
          required: true
          schema:
            type: integer
      responses:
        "200":
          description: A single post
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/Post"
        "404":
          description: Post not found

flaskr/app.py

def get_post(id: int):
    post = db_session.scalar(
        select(Post).options(joinedload(Post.author)).where(Post.id == id)
    )

    if not post:
        return {"message": f"Post {id} not found"}, 404

    return {
        "id": post.id,
        "author": post.author.username,
        "created": post.created,
        "title": post.title,
        "body": post.body,
    }

Blog-post aanpassen

Met PUT /posts/{id}.

Oplossing

flaskr/openapi.yaml

  /posts/{id}:
    put:
      summary: Update a post
      operationId: flaskr.app.update_post
      parameters:
        - in: path
          name: id
          required: true
          schema: {type: integer}
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/NewPost"
      responses:
        "200":
          description: Updated post
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/Post"
        "400":
          description: Invalid input
        "404":
          description: Post not found

flaskr/app.py

def update_post(id: int, body: dict):
    post = db_session.scalar(
        select(Post).options(joinedload(Post.author)).where(Post.id == id)
    )

    if not post:
        return {"message": f"Post {id} not found"}, 404

    post.title = body["title"]
    post.body = body["body"]
    db_session.commit()

    return {
        "id": post.id,
        "author": post.author.username,
        "created": post.created,
        "title": post.title,
        "body": post.body,
    }

Blog-post verwijderen

Met DELETE /posts/{id}.

Oplossing

flaskr/openapi.yaml

  /posts/{id}:
    delete:
      summary: Delete a post
      operationId: flaskr.app.delete_post
      parameters:
        - in: path
          name: id
          required: true
          schema: {type: integer}
      responses:
        "204":
          description: Post deleted
        "404":
          description: Post not found

flaskr/app.py

def delete_post(id: int):
    post = db_session.scalar(
        select(Post).options(joinedload(Post.author)).where(Post.id == id)
    )

    if not post:
        return {"message": f"Post {id} not found"}, 404

    db_session.delete(post)
    db_session.commit()

    return None, 204

Refactoring

Het is je waarschijnlijk opgevallen dat in de meeste operation functies dezelfde code wordt herhaald om een Post ORM object om te zetten naar een dict.

We zouden een eenvoudige helper functie kunnen toevoegen, bvb.

flaskr/app.py

def _post_to_dict(post: Post) -> dict:
    return {
          "id": post.id,
          "author": post.author.username,
          "created": post.created,
          "title": post.title,
          "body": post.body,
      }

Authenticatie

Dit kan opnieuw met HTTP Basic Auth, zoals in de Quotes API. Deze keer hebben we wel een echte user database waar we het password mee kunnen valideren. Dat kan op dezelfde manier gebeuren als in de Flask implementatie.

Op deze manier willen we de POST operatie aanpassen om de juiste author_id te gebruiken, en de PUT en DELETE operaties beveiligen zodat een gebruiken enkel zijn eigen posts kan aanpassen.

Bedenkt ook eens wat er moet gebeuren als de author_id niet overeenkomt.

Voor de OAS volgen we precies dezelfde aanpak als bij de Quotes API.

flaskr/openapi.yaml

paths:
  /posts:
    post:
      summary: Create a new post
      ...
      security:
        - basic: []

  /posts/{id}:
    put:
      summary: Update a post
      ...
      security:
        - basic: []

    delete:
      summary: Delete a post
      ...
      security:
        - basic: []

components:
  securitySchemes:
    basic:
      type: http
      scheme: basic
      x-basicInfoFunc: flaskr.app.auth

flaskr/app.py

def auth(username: str, password: str):
    user = db_session.scalar(select(User).where(User.username == username))

    if user and check_password_hash(user.password, password):
1        return {"sub": user.id, "username": username}
    return

1: De return value (dict) van deze functie is automatisch beschikbaar in alle view functies via een extra token_info argument.

PUT en DELETE

Voor de PUT en DELETE operations kunnen we simpelweg de post.author_id vergelijken met de userid uit de authenticatie. Die laatste is beschikbaar via token_info["sub"].

Als deze niet overeenkomt krijgt de client een 404/Not Found respons. Dit is in de praktijk niet ongebruikelijk omdat op deze manier het al dan niet bestaan van de gevraagde resource ID niet wordt vrijgegeven (ook al is dat hier minder relevant omdat je zonder authenticatie alle posts kan ophalen).

flaskr/app.py

def update_post(id: int, body: dict, token_info: dict):
    post = db_session.scalar(
        select(Post).options(joinedload(Post.author)).where(Post.id == id)
    )

    if not post or post.author_id != token_info["sub"]:
        return {"message": f"Post {id} not found"}, 404

    post.title = body["title"]
    post.body = body["body"]
    db_session.commit()

    return _post_to_dict(post)

def delete_post(id: int, token_info: dict):
    post = db_session.scalar(
        select(Post).options(joinedload(Post.author)).where(Post.id == id)
    )

    if not post or post.author_id != token_info["sub"]:
        return {"message": f"Post {id} not found"}, 404

    db_session.delete(post)
    db_session.commit()

    return None, 204

POST

Voor het maken van een nieuwe post gebruiken we dezelfde aanpak. token_info["sub"] vervangt de tot nu toe hardcoded waarde 1.

flaskr/app.py

def create_post(body: dict, token_info: dict):
    post = Post(title=body["title"], body=body["body"], author_id=token_info["sub"])
    db_session.add(post)
    db_session.commit()
    return _post_to_dict(post)

Extra

We halen kort nog een aantal onderwerpen aan voor wie OpenAPI en Connexion nog verder wil verkennen

Petstore

Wil je een uitgebreider voorbeeld van een OpenAPI spec, bekijk dan zeker eens Petstore en de bijhorende specificatie. Dit is een beetje het defacto voorbeeld dat door de jaren en versies heen is mee geëvolueerd met de specificatie.

Testen

Net als Flask voorziet Connexion een test client. Een eenvoudig voorbeeld:

tests/test_api_app.py

from flaskr.app import app

def test_get_post():
    with app.test_client() as client:
        response = client.get("/posts/1")

        assert response.status_code == 200
        assert response.json() == {
            "author_id": 1,
            "body": "world",
            "created": "2025-09-23T06:12:15Z",
            "id": 1,
            "title": "hello",
        }

Context

Ook zoals Flask heeft Connexion verschillende context objecten, zoals bvb connexion.request, waarmee je rechtstreeks toegang krijgt tot de request details (headers, path, query params, …)

Exception Handlers

In de huidige implementatie proberen we fouten, zoals wanneer een gevraagde post niet bestaat, rechtstreeks af te handelen door een aangepaste status code and body terug te sturen als respons.

Voor dit eenvoudig voorbeeld is dat prima maar in complexere applicaties wordt de code al snel opgesplitst in verschillende modules en wordt er vaak met (eigen) exceptions gewerkt.

Op https://connexion.readthedocs.io/en/stable/exceptions.html vinden we niet alleen een overzicht van de verschillende exceptions eigen aan Connexion, en hoe deze worden afgehandeld, maar ook een manier om andere exceptions af te handelen.

We laten hier een mogelijke aanpak zien waarbij we een eigen exception maken die we zullen raisen wanneer een post niet gevonden is.

flaskr/app.py

class ResourceNotFound(connexion.ProblemException):
    def __init__(self, resource_id: int | None = None):
        super().__init__(
            status=404,
            title="NotFound",
            detail=f"Resource {resource_id if resource_id else ''} does not exist",
        )

def get_post(id: int):
    post = db_session.scalar(
        select(Post).options(joinedload(Post.author)).where(Post.id == id)
    )

    if not post:
        raise ResourceNotFound(id)

    return _post_to_dict(post)

Referentie Implementatie

Een volledig werkende implementatie gebaseerd op deze les vind je op https://github.com/dvx76/flask-tutorial/tree/flaskr-api-sqla.