FastSQLA

Async SQLAlchemy 2.0+ for FastAPI — boilerplate, pagination, and seamless session management.

Documentation: https://hadrien.github.io/FastSQLA/

Github Repo: https://github.com/hadrien/fastsqla

---

FastSQLA is an async SQLAlchemy 2.0+ extension for FastAPI with built-in pagination, SQLModel support and more.

It streamlines the configuration and asynchronous connection to relational databases by providing boilerplate and intuitive helpers. Additionally, it offers built-in customizable pagination and automatically manages the SQLAlchemy session lifecycle following SQLAlchemy's best practices.

Features

• Easy setup at app startup using FastAPI Lifespan:

  from fastapi import FastAPI
  from fastsqla import lifespan
  
  app = FastAPI(lifespan=lifespan)

• SQLAlchemy async session dependency:

  ...
  from fastsqla import Session
  from sqlalchemy import select
  ...
  
  @app.get("/heros")
  async def get_heros(session:Session):
      stmt = select(...)
      result = await session.execute(stmt)
      ...

• SQLAlchemy async session with an async context manager:

  from fastsqla import open_session
  
  async def background_job():
      async with open_session() as session:
          stmt = select(...)
          result = await session.execute(stmt)
          ...

• Built-in pagination:

  ...
  from fastsqla import Page, Paginate
  from sqlalchemy import select
  ...
  
  @app.get("/heros", response_model=Page[HeroModel])
  async def get_heros(paginate:Paginate):
      return await paginate(select(Hero))

  👇 /heros?offset=10&limit=10 👇

  {
    "data": [
      {
        "name": "The Flash",
        "secret_identity": "Barry Allen",
        "id": 11
      },
      {
        "name": "Green Lantern",
        "secret_identity": "Hal Jordan",
        "id": 12
      }
    ],
    "meta": {
      "offset": 10,
      "total_items": 12,
      "total_pages": 2,
      "page_number": 2
    }
  }

• Pagination customization:

  ...
  from fastapi import Page, new_pagination
  ...
  
  Paginate = new_pagination(min_page_size=5, max_page_size=500)
  
  @app.get("/heros", response_model=Page[HeroModel])
  async def get_heros(paginate:Paginate):
      return paginate(select(Hero))

• Session lifecycle management: session is commited on request success or rollback on failure.

• SQLModel support:

  ...
  from fastsqla import Item, Page, Paginate, Session
  from sqlmodel import Field, SQLModel
  ...
  
  class Hero(SQLModel, table=True):
      id: int | None = Field(default=None, primary_key=True)
      name: str
      secret_identity: str
      age: int
  
  
  @app.get("/heroes", response_model=Page[Hero])
  async def get_heroes(paginate: Paginate):
      return await paginate(select(Hero))
  
  
  @app.get("/heroes/{hero_id}", response_model=Item[Hero])
  async def get_hero(session: Session, hero_id: int):
      hero = await session.get(Hero, hero_id)
      if hero is None:
          raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
      return {"data": hero}

Installing

Using uv:

uv add fastsqla

Using pip:

pip install fastsqla

Quick Example

example.py

Let's write some tiny app in example.py:

# example.py
from http import HTTPStatus

from fastapi import FastAPI, HTTPException
from fastsqla import Base, Item, Page, Paginate, Session, lifespan
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Mapped, mapped_column


app = FastAPI(lifespan=lifespan)


class Hero(Base):
    __tablename__ = "hero"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)
    secret_identity: Mapped[str]
    age: Mapped[int]


class HeroBase(BaseModel):
    name: str
    secret_identity: str
    age: int


class HeroModel(HeroBase):
    model_config = ConfigDict(from_attributes=True)
    id: int


@app.get("/heros", response_model=Page[HeroModel])
async def list_heros(paginate: Paginate):
    stmt = select(Hero)
    return await paginate(stmt)


@app.get("/heros/{hero_id}", response_model=Item[HeroModel])
async def get_hero(hero_id: int, session: Session):
    hero = await session.get(Hero, hero_id)
    if hero is None:
        raise HTTPException(HTTPStatus.NOT_FOUND, "Hero not found")
    return {"data": hero}


@app.post("/heros", response_model=Item[HeroModel])
async def create_hero(new_hero: HeroBase, session: Session):
    hero = Hero(**new_hero.model_dump())
    session.add(hero)
    try:
        await session.flush()
    except IntegrityError:
        raise HTTPException(HTTPStatus.CONFLICT, "Duplicate hero name")
    return {"data": hero}

Database

💡 This example uses an SQLite database for simplicity: FastSQLA is compatible with all asynchronous db drivers that SQLAlchemy is compatible with.

Let's create an SQLite database using sqlite3 and insert 12 rows in the hero table:

sqlite3 db.sqlite <<EOF
-- Create Table hero
CREATE TABLE hero (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    name            TEXT NOT NULL UNIQUE, -- Unique hero name (e.g., Superman)
    secret_identity TEXT NOT NULL,        -- Secret identity (e.g., Clark Kent)
    age             INTEGER NOT NULL      -- Age of the hero (e.g., 30)
);

-- Insert heroes with their name, secret identity, and age
INSERT INTO hero (name, secret_identity, age) VALUES ('Superman',        'Clark Kent',       30);
INSERT INTO hero (name, secret_identity, age) VALUES ('Batman',          'Bruce Wayne',      35);
INSERT INTO hero (name, secret_identity, age) VALUES ('Wonder Woman',    'Diana Prince',     30);
INSERT INTO hero (name, secret_identity, age) VALUES ('Iron Man',        'Tony Stark',       45);
INSERT INTO hero (name, secret_identity, age) VALUES ('Spider-Man',      'Peter Parker',     25);
INSERT INTO hero (name, secret_identity, age) VALUES ('Captain America', 'Steve Rogers',     100);
INSERT INTO hero (name, secret_identity, age) VALUES ('Black Widow',     'Natasha Romanoff', 35);
INSERT INTO hero (name, secret_identity, age) VALUES ('Thor',            'Thor Odinson',     1500);
INSERT INTO hero (name, secret_identity, age) VALUES ('Scarlet Witch',   'Wanda Maximoff',   30);
INSERT INTO hero (name, secret_identity, age) VALUES ('Doctor Strange',  'Stephen Strange',  40);
INSERT INTO hero (name, secret_identity, age) VALUES ('The Flash',       'Barry Allen',      28);
INSERT INTO hero (name, secret_identity, age) VALUES ('Green Lantern',   'Hal Jordan',       35);
EOF

Run the app

Let's install required dependencies:

pip install uvicorn aiosqlite fastsqla

Let's run the app:

sqlalchemy_url=sqlite+aiosqlite:///db.sqlite?check_same_thread=false \
  uvicorn example:app

Check the result

Execute GET /heros?offset=10&limit=10 using curl:

curl -X 'GET' -H 'accept: application/json' 'http://127.0.0.1:8000/heros?offset=10&limit=10'

Returns:

{
  "data": [
    {
      "name": "The Flash",
      "secret_identity": "Barry Allen",
      "id": 11
    },
    {
      "name": "Green Lantern",
      "secret_identity": "Hal Jordan",
      "id": 12
    }
  ],
  "meta": {
    "offset": 10,
    "total_items": 12,
    "total_pages": 2,
    "page_number": 2
  }
}

You can also check the generated openapi doc by opening your browser to http://127.0.0.1:8000/docs.

License

This project is licensed under the terms of the MIT license.