Asynchronous DB I/O with SQLAlchemy Core on FastAPI

asnyc_sqlalchemy_core_on_fastapi.py

import asyncio
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import sqlalchemy
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "sqlite+aiosqlite:///simulations.db", echo=True, future=True
)

metadata_obj = sqlalchemy.MetaData()
snapshots_table = sqlalchemy.Table(
    "snapshots",
    metadata_obj,
    sqlalchemy.Column("filename", sqlalchemy.String),
    sqlalchemy.Column("saved", sqlalchemy.Integer),
    sqlalchemy.Column("loop", sqlalchemy.Integer),
)


async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(metadata_obj.create_all)


class RecordSummary(BaseModel):
    filename: str
    saved: float
    max_loop: int


app = FastAPI()


@app.on_event("startup")
async def startup_event():
    await init_db()


@app.on_event("shutdown")
async def shutdown_event():
    await engine.dispose()


@app.get("/save_sim", response_model=Response)
async def save_sim():
    _values = []
    
    # ...do something to append records to _values here.
    # e.g. ('abc', 20220623, 5)
    
    stmt = sqlalchemy.insert(snapshots_table).values(_values)
    async with engine.connect() as conn:
        res = await conn.execute(stmt)
        await conn.commit()

    return {"code": 1, "message": f"Success({res})"}

  
@app.get("/get_sim", response_model=List[RecordSummary])
async def get_sim(offset: int = 0, limit: int = 10):
    stmt = sqlalchemy.select(
        snapshots_table.c.filename,
        snapshots_table.c.saved,
        sqlalchemy.func.max(snapshots_table.c.loop).label("max_loop")
    ).group_by(snapshots_table.c.filename, snapshots_table.c.saved).offset(offset).limit(limit)

    async with engine.connect() as conn:
        res = await conn.execute(stmt)

    return res.fetchall()