10 Tips om SQLAlchemy met FastAPI te Integreren

Introductie

De populariteit van FastAPI is de afgelopen jaren gestaag gegroeid. Het stijgt langzaam naar de top 25 van webframeworks en technologieën die worden gebruikt in de Stack Overflow Developer Survey 2022,, net achter Flask en Django (maar verslaat beide in Loved vs. Dreaded). In een tijdperk waarin data en AI een groter deel van elk bedrijf opeisen, is het geen verrassing dat er veel interesse is in het lichtgewicht, prestatiegerichte API-framework voor Python. En je hebt slechts een paar regels code nodig om aan de slag te gaan:

1
2
3
4
5
6
7

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return "hello"

Het gebruik van FastAPI onder datawetenschappers en ML-engineers groeit op dit moment. Het framework is zo minimalistisch dat je misschien zou denken dat er weinig ruimte is voor fouten. Maar complexiteit is incrementeel. Uiteindelijk moeten deze API’s worden bijgewerkt, uitgebreid en onderhouden. Je kunt beter The Art of Computer Programming hebben gelezen dan The Art of Statistics. Alleen al denken aan de halve ETL-pijplijn die ik jaren geleden in elkaar heb gehackt met Pandas geeft met tot de dag van vandaag de rillingen.

Daarom willen we enkele van onze inzichten delen over het volwassen worden van FastAPI-apps, in het bijzonder: het toevoegen van een relationele database. Of je nu FastAPI gebruikt in een grote webapplicatie of een kleine ML-inferentieservice, de kans is groot dat je op een gegeven moment een relationele database introduceert. Misschien wil je metingen bijhouden of wat gebruikers toevoegen. Het populairste Python-pakket om een relationele database toe te voegen is SQLAlchemy. Het biedt een kernframework voor communicatie met een database, evenals een ORM. Iedereen die met een relationele database werkt in Python heeft gewerkt met - of op zijn minst overwogen - SQLAlchemy. In deze blogpost delen we 10 tips om SQLAlchemy in je FastAPI applicatie te integreren. De voorbeeldcode is te zien in actie in ons FastAPI Template Project.

FastAPI en DDD

Wanneer je besluit een relationele database toe te voegen, is het belangrijk om dit op de juiste manier te integreren. In dit gedeelte introduceren we een paar praktijken en abstracties die we nuttig hebben gevonden voor zowel het onderhouden als het testen van de code.

1. Stop geen BaseModels in Endpoints

SQLAlchemy ORM gebruikt wat het datamapperpattern wordt genoemd. Het datamapperpattern is handig vanwege zijn flexibiliteit en probeert de Object–relational impedance mismatch tussen objecten en databases te overbruggen. Maar er zijn nadelen verbonden aan het gebruik van deze ORM-modellen in je hele applicatie, met name in je API. Wat als je de endpoints eenvoudig wilt houden voor gebruikers, maar objecten wilt opslaan in meerdere tabellen of bepaalde velden voor gebruikers wilt verbergen? De code kan gemakkelijk in bochten wringen om onhandige logica toe te voegen. Dit kan je API moeilijk te gebruiken maken. Je moet overwegen om je API en databaseschema afzonderlijk te laten evolueren.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class TodoInDB(SQL_BASE):  # type: ignore
    __tablename__ = "todo"

    id = Column(Integer, primary_key=True, autoincrement=True)
    key = Column(String(length=128), nullable=False, unique=True)
    value = Column(String(length=128), nullable=False)
    done = Column(Boolean, default=False)


class Todo(BaseModel):
    key: str
    value: str
    done: bool = False

In het bijzonder moet je niet proberen een klasse te maken die zowel een Pydantic BaseModel als een declaratieve basis van SQLAlchemy is. Pydantic is voornamelijk een bibliotheek voor invoervalidatie. Hierboven zien we een voorbeeld van een model waarvoor zowel een Pydantic BaseModel is gemaakt voor de API als een SQLAlchemy-model. Je hoeft deze objecten alleen maar naar elkaar te vertalen. (De volgende tip vertelt je waar je dat moet doen.)

2. Gebruik de Repository Pattern

Het vermogen om componenten in je software te vervangen, geeft aan dat het modulair is. En hoewel je waarschijnlijk niet binnenkort van database zult wisselen, stelt het kunnen vervangen van daadwerkelijke persistentie met in-process geheugen je in staat om de beslissing om nieuwe entiteiten (of zelfs een hele database) in te voeren uit te stellen, evenals het gemakkelijker maken om unit tests te schrijven. Hiervoor is het nodig om een interface te creëren rond de opslaglaag en de bijbehorende querylogica. Hier komt de Repository Pattern om de hoek kijken.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class TodoRepository:
    def save(self, todo: Todo):
        raise NotImplementedError()

    def get_by_key(self, key: str) -> Optional[Todo]:
        raise NotImplementedError()

class SQLTodoRepository(TodoRepository):
    def __init__(self, session: Session):
        self._session: Session = session

    def save(self, todo: Todo):
        self._session.add(TodoInDB(key=todo.key, value=todo.value))

    def get_by_key(self, key: str) -> Optional[Todo]:
        instance = self._session.query(TodoInDB).filter(TodoInDB.key == key).first()

        if instance:
            return Todo(key=instance.key, value=instance.value, done=instance.done)

De Repository Pattern is te vinden in boeken zoals Domain Driven Design. Voor onze doeleinden zijn repositories bemiddelaars die verantwoordelijk zijn voor het opslaan van entiteiten in je domeinmodel. Een repository heeft doorgaans methods zoals save(), get_by_id() or get_all_enabled(). De signatures van deze methods bestaan uit domeinmodellen en queryparameters zoals filters. Ze onthullen niet hoe ze entiteiten opslaan, maar alleen welke entiteiten moeten worden opgeslagen en ontvangen met behulp van domeinmodellen.

De interface van een repository kan worden geïmplementeerd door een SQLAlchemy-implementatie en een eenvoudige in-memory-implementatie. Zoals eerder vermeld, kan de laatstgenoemde worden gebruikt voor unit tests.

3. Gebruik Query Objecten

Een pattern dat ik persoonlijk nuttig heb gevonden, is het gebruik van Query Objects. Omdat je vaak je database wilt bevragen met verschillende filters, stelt het verzamelen ervan in een Query Object je in staat om filterparameters toe te voegen of te verwijderen, standaard filterwaarden te wijzigen en zelfs filters opnieuw te gebruiken zonder de interface van een Repository te wijzigen. Een ander voordeel is dat het de hoeveelheid argumenten vermindert die je moet bijhouden bij het ophalen van entiteiten.

 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

class TodoFilter(BaseModel):
    limit: int = 1
    key_contains: Optional[str] = None
    value_contains: Optional[str] = None
    done: Optional[bool] = None

class SQLTodoRepository:
    ...

    def get(self, todo_filter: TodoFilter) -> List[Todo]:
        query = self._session.query(TodoInDB)

        if todo_filter.key_contains is not None:
            query = query.filter(TodoInDB.key.contains(todo_filter.key_contains))

        if todo_filter.value_contains is not None:
            query = query.filter(TodoInDB.value.contains(todo_filter.value_contains))

        if todo_filter.done is not None:
            query = query.filter(TodoInDB.done == todo_filter.done)

        if todo_filter.done is not None:
            query = query.limit(todo_filter.limit)

        return [Todo(key=todo.key, value=todo.value, done=todo.done) for todo in query]

We hebben allemaal op een gegeven moment functies geschreven met te veel argumenten; een codegeur die aangeeft dat je mogelijk een nieuw object wilt introduceren. In het geval van repositories is dit vaak een Query Object dat je nodig hebt.

Het Managen van SQLAlchemy Scopes

Zowel FastAPI als SQLAlchemy hebben hun eigen documentatie over hoe je een database kunt toevoegen met behulp van depends (FastAPI) en het beheren van verbindingen, sessies en transacties, enz. (SQLAlchemy). Maar de eerste werkt met een vrij beknopte setup, terwijl de laatste een zeer gedetailleerde uitleg is over hoe je met SQLAlchemy in al zijn glorie kunt werken. Hier laten we een complete maar eenvoudige setup zien die voor de meeste API’s voldoende zou moeten zijn. Dit maakt gebruik van de verschillende scopes van SQLAlchemy zonder implementatiedetails aan je domein te koppelen.

4. Herbruik database connecties

Zoals elke documentatiepagina je zal vertellen, is het niet hergebruiken van databaseverbindingen een groot anti-pattern. Het opzetten van verbindingen is duur, dus telkens opnieuw verbinding maken voor elke transactie brengt aanzienlijke overhead met zich mee. De documentatie van FastAPI laat zien hoe je hiermee kunt omgaan. Een manier om hiermee om te gaan, is het cachen van de functie die de engine initialiseert (het object dat het verbindingspool beheert).

1
2
3

@lru_cache(maxsize=None)
def get_engine():
    return create_engine(os.getenv("DB_STRING"), pool_pre_ping=True)

Op deze manier hergebruik je de engine tussen requests en betaal je alleen voor de verbindingen bij de eerste.

5. Een Sessie per Request

De tweede laag van administratie is het beheren van sessies. Sessies zijn niet duur om te maken. Ze vormen een laag tussen de database die transacties uitvoert, objecten met identiteit toewijst, enzovoort. SQLAlchemy heeft zijn eigen documentatie over sessies, waarin ze de transactie-scope en de sessie-scope definiëren. Kort gezegd is het in het geval van API’s een goede praktijk om één sessie per request te maken.

Om de levenscyclus van een sessie te beheren, kunnen we gebruikmaken van FastAPI’s Depends door een iterator te maken.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def create_todo_repository() -> Iterator[TodoRepository]:
    session = sessionmaker(bind=get_engine())()
    todo_repository = SQLTodoRepository(session)

    try:
        yield todo_repository
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

De FastAPI-endpoint om een Todo te maken zou er dan eenvoudigweg zo uitzien:

1
2
3

@app.post("/{key}")
def create(key: str, value: str, todo_repository: TodoRepository = Depends(create_todo_repository)):
    ...

Telkens wanneer het endpoint wordt aangeroepen, wordt er een repository geïnstantieerd met een nieuwe sessie die automatisch wordt afgesloten nadat de request is voltooid. Zelfs als er een uitzondering optreedt, rollen we de sessie terug.

6. Transacties met Context Managers

Het laatste wat we moeten beheren zijn individuele transacties. Soms biedt één transactie per request voldoende controle. In dat geval volstaat het om de sessie te committen vlak voor het sluiten in de create_todo_repository-functie. Wanneer je transacties wilt committen in de service-laag, hebben repositories een manier nodig om transacties te construeren zonder een ‘leaky abstraction’ te introduceren (door de sessie bloot te stellen). Gelukkig zijn contextmanagers hiervoor een goede oplossing.

Om van een repository een contextmanager te maken, moet je een __enter__() en een __exit__() methode implementeren. We kunnen de sessie committen bij het verlaten van de context. De __exit__()-methode stelt je zelfs in staat om eventuele uitzonderingen die zich voordoen in de huidige context af te handelen, zodat je veilig eventuele lopende transacties kunt terugdraaien voordat je de context verlaat.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class SQLTodoRepository:
    def __init__(self, session):
        self._session: Session = session

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value: str, exc_traceback: str) -> None:
        if exc_type is not None:
            self._session.rollback()
            return

        try:
            self._session.commit()
        except DatabaseError:
            self._session.rollback()
            raise

    ...

7. Gebruik Alembic

Op een gegeven moment zul je waarschijnlijk je databaseschema wijzigen. Alembic is de standaardoplossing voor het beheer van databaseschema’s voor SQLAlchemy. (Ze zijn ook ontwikkeld door dezelfde auteur!)

8. Test de Repositories

Het mooie aan repositories is dat ze gemakkelijk te testen zijn. Je moet controleren of ze de juiste entiteiten opslaan en retourneren, de juiste filtering uitvoeren en geen ongeldige waarden accepteren. Als je je repositories goed test, geeft dat je veel vertrouwen bij het schrijven van complexere bedrijfslogica, omdat je dan in ieder geval weet dat je de juiste entiteiten gebruikt. Als je in-memory-implementaties hebt (en misschien ook ContractTests toevoegt voor je repositories), kun je zelfs de SQL-implementaties vervangen door deze lichtgewicht testdoubles om de bedrijfslogica unitair te testen.

Terwijl we hiermee bezig zijn, kunnen we net zo goed onze migraties testen door Alembic te gebruiken in onze tests. Dit zorgt ervoor dat je niet vergeet om je migraties te genereren en te committen, aangezien je tests anders zullen falen.

 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

@pytest.fixture
def todo_repository():
    time.sleep(1)
    alembicArgs = ["--raiseerr", "upgrade", "head"]
    alembic.config.main(argv=alembicArgs)

    engine = get_engine(os.getenv("DB_STRING"))
    session = sessionmaker(bind=engine)()

    yield SQLTodoRepository(session)

    session.close()

    sessionmaker(bind=engine, autocommit=True)().execute(
        ";".join([f"TRUNCATE TABLE {t} CASCADE" for t in SQL_BASE.metadata.tables.keys()])
    )

@pytest.mark.integration
def test_repository(todo_repository: SQLTodoRepository):
    with todo_repository as r:
        r.save(Todo(key="testkey", value="testvalue"))

    todo = r.get_by_key("testkey")
    assert todo.value == "testvalue"

    with pytest.raises(IntegrityError):
        with todo_repository as r:
            r.save(Todo(key="testkey", value="not allowed: unique todo keys!"))

    with pytest.raises(DataError):
        with todo_repository as r:
            r.save(Todo(key="too long", value=129 * "x"))

@pytest.mark.integration
def test_repository_filter(todo_repository: SQLTodoRepository):
    with todo_repository as r:
        r.save(Todo(key="testkey", value="testvalue"))
        r.save(Todo(key="abcde", value="v"))

    todos = r.get(TodoFilter(key_contains="test"))
    assert len(todos) == 1
    assert todos[0].value == "testvalue"

    todos = r.get(TodoFilter(key_contains="abcde"))
    assert len(todos) == 1
    assert todos[0].value == "v"

    assert len(r.get(TodoFilter(key_contains="e"))) == 2
    assert len(r.get(TodoFilter(key_contains="e", limit=1))) == 1
    assert len(r.get(TodoFilter(value_contains="v"))) == 2
    assert len(r.get(TodoFilter(done=True))) == 0

Wanneer moet je een Database aan FastAPI Toevoegen

Voordat we dieper ingaan op SQLAlchemy, zijn hier twee tips die je helpen om de juiste overwegingen te maken als je nog niet de stap hebt gezet om SQLAlchemy te introduceren.

9. Zorg ervoor dat de App echt een Relationele Database nodig heeft

Alles is een afweging. En hoewel een groot deel van het web draait op relationele databases, is PostgreSQL niet het antwoord op alles. Ondanks hun veelzijdigheid kunnen relationele databases niet altijd goed passen bij API’s, vooral in een vroeg stadium. Heeft jouw inferentie-eindpunt alleen een getraind model nodig dat zo nu en dan wordt bijgewerkt? Dan is het waarschijnlijk voldoende om het versienummer toe te voegen aan de bestandsnaam van de modellen. (Python wheels is een van de vele voorbeelden van hoe data grotendeels beheerd kan worden door middel van een naamgevingsconventie.) Wil je bepaalde berekeningen cachen om de responstijden te versnellen? Bekijk dan eerst de caching decorators van functools voordat je Redis toevoegt. Zelfs bij het toevoegen van authenticatie aan je API, als het alleen voor je eerste klant is, kun je hun hashed referenties rechtstreeks verifiëren in je omgeving.

Aan de andere kant kan jouw API prestatievereisten hebben die niet kunnen worden voldaan door een relationele database. Relationele databases zijn ontworpen en geoptimaliseerd voor OLTP-doeleinden. Voor OLAP-gebruiksscenario’s waarbij jouw API complexe analytische dashboards levert op grote hoeveelheden data, kunnen er meer geschikte databasetechnologieën zijn (bekijk Google BigQuery, Amazon Redshift, HBase… of zelfs DuckDB!). De verwachte hoeveelheid lees- en schrijfoperaties, het meest voorkomende querypattern en de ACID-vereisten zijn allemaal factoren die in overweging moeten worden genomen. Het domein van data is enorm geëvolueerd en als we iets hebben geleerd, is het wel dat er geen one-size-fits-all-oplossing is.

10. Besteed Genoeg Aandacht aan je Datamodel

Als je het toevoegen van een relationele database lang genoeg hebt uitgesteld, moet je nog eens kritisch kijken naar je datamodel. Natuurlijk, als je de data zelf beheert en het niet van cruciaal belang is, kun je bijna alles laten passeren. Maar wanneer je jouw app in een productieomgeving plaatst en de gegevens van anderen begint op te slaan, kunnen migraties van een snel evoluerend databaseschema een echte uitdaging worden. Klassieke fouten zijn onder andere het aanmaken van een nullable kolom die verplicht had moeten zijn of het verkorten van VARCHAR-lengtes. En hoe genormaliseerd moet jouw schema zijn? Maak je die extra tabel en offer je wat queryprestaties op vanwege extra joins die je moet doen, of accepteer je duplicatie?

Architectuur is moeilijk te veranderen, en jouw databaseschema maakt daar deel van uit. Het kan ook nuttig zijn om extern advies te vragen!

Conclusie

SQLAlchemy en FastAPI zijn beide geweldige Python-pakketten die je snel op weg helpen met minimale configuratie. In deze post hebben we verschillende aspecten genoemd waar je rekening mee moet houden bij het toevoegen van een relationele database aan je API en naarmate je project volwassener wordt. Maar zoals hierboven vermeld, is alles een afweging. Misschien vind je het veel gemakkelijker werken om de ORM-modus van Pydantic in te schakelen en heb je nooit behoefte aan meer flexibiliteit op zowel je API als de mapping met de database. Misschien wil je de contextmanager weglaten en gewoon autocommit=True instellen bij het maken van een nieuwe sessie. Uiteindelijk ken je de vereisten van jouw applicatie het beste.

Dat gezegd hebbende, hopelijk helpen deze tips je bij het bouwen van een goede data-applicatie of geven ze op zijn minst stof tot nadenken over het ontwerp en de configuratie ervan. Als je alle code in deze post in actie wilt zien, kun je onze FastAPI Template Project op Github bekijken en onze post erover lezen. Als je denkt dat we belangrijke tips hebben gemist of het oneens bent met de genoemde tips, voel je vrij om contact met ons op te nemen! We staan altijd open voor discussie en verbetering van ideeën over Python API’s en data.