# FastSQLA
Async SQLAlchemy 2.0+ extension for FastAPI with SQLModel support, built-in pagination & more.
# Sections
# FastSQLA
*Async SQLAlchemy 2.0+ for FastAPI — boilerplate, pagination, and seamless session management.*
**Documentation**:
**Github Repo:**
______________________________________________________________________
`FastSQLA` is an async [`SQLAlchemy 2.0+`](https://docs.sqlalchemy.org/en/20/) extension for [`FastAPI`](https://fastapi.tiangolo.com/) with built-in pagination, [`SQLModel`](http://sqlmodel.tiangolo.com/) 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](https://docs.sqlalchemy.org/en/20/orm/session_basics.html#when-do-i-construct-a-session-when-do-i-commit-it-and-when-do-i-close-it).
## Features
- Easy setup at app startup using [`FastAPI` Lifespan](https://fastapi.tiangolo.com/advanced/events/#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 typing import Annotated
from fastapi import Depends
from fastsqla import Page, PaginateType, new_pagination
CustomPaginate = Annotated[
PaginateType[HeroModel],
Depends(new_pagination(default_page_size=5, max_page_size=500)),
]
@app.get("/heroes", response_model=Page[HeroModel])
async def get_heroes(paginate: CustomPaginate):
return await paginate(select(Hero))
```
- Session lifecycle management: session is commited on request success or rollback on failure.
- [`SQLModel`](http://sqlmodel.tiangolo.com/) 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](https://docs.astral.sh/uv/):
```
uv add fastsqla
```
Using [pip](https://pip.pypa.io/):
```
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 <.
## License
This project is licensed under the terms of the [MIT license](https://github.com/hadrien/FastSQLA/blob/main/LICENSE).
## For coding agents and LLMs
FastSQLA publishes agent-readable documentation alongside the website:
- [`llms.txt`](https://hadrien.github.io/FastSQLA/llms.txt) is the concise documentation index.
- [`llms-full.txt`](https://hadrien.github.io/FastSQLA/llms-full.txt) contains the complete documentation in one file.
- Every indexed page has a Markdown twin, such as [`setup/index.md`](https://hadrien.github.io/FastSQLA/setup/index.md).
- [Context7](https://context7.com/hadrien/fastsqla) serves the current documentation and FastSQLA-specific usage rules.
The repository also bundles Agent Skills for setup, session management, and pagination. Install all three as one plugin:
### Claude Code
```
claude plugin marketplace add hadrien/FastSQLA
claude plugin install fastsqla@fastsqla
```
### Codex
```
codex plugin marketplace add hadrien/FastSQLA
codex plugin add fastsqla@fastsqla
```
# Setup
FastSQLA provides two ways to configure your SQLAlchemy database connection:
- **Environment variables** (lifespan): Simple configuration following [12-factor app](https://12factor.net/config) principles, ideal for most use cases.
- **Programmatic** (new_lifespan): Direct SQLAlchemy engine configuration for advanced customization needs
## `fastsqla.lifespan`
Use `fastsqla.lifespan` to set up SQLAlchemy directly from environment variables.
In an ASGI application, [lifespan events](https://asgi.readthedocs.io/en/latest/specs/lifespan.html) are used to communicate startup & shutdown events.
The [`lifespan`](https://fastapi.tiangolo.com/advanced/events/#lifespan) parameter of the `FastAPI` app can be assigned to a context manager, which is opened when the app starts and closed when the app stops.
In order for `FastSQLA` to setup `SQLAlchemy` before the app is started, set `lifespan` parameter to `fastsqla.lifespan`:
```
from fastapi import FastAPI
from fastsqla import lifespan
app = FastAPI(lifespan=lifespan)
```
If multiple lifespan contexts are required, create an async context manager function to handle them and set it as the app's lifespan:
```
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastsqla import lifespan as fastsqla_lifespan
from this_other_library import another_lifespan
@asynccontextmanager
async def lifespan(app:FastAPI) -> AsyncGenerator[dict, None]:
async with AsyncExitStack() as stack:
yield {
**stack.enter_async_context(lifespan(app)),
**stack.enter_async_context(another_lifespan(app)),
}
app = FastAPI(lifespan=lifespan)
```
To learn more about lifespan protocol:
- [Lifespan Protocol](https://asgi.readthedocs.io/en/latest/specs/lifespan.html)
- [Use Lifespan State instead of `app.state`](https://github.com/Kludex/fastapi-tips?tab=readme-ov-file#6-use-lifespan-state-instead-of-appstate)
- [FastAPI lifespan documentation](https://fastapi.tiangolo.com/advanced/events/)
### Lifespan configuration
Configuration is done exclusively via environment variables, adhering to the [**Twelve-Factor App methodology**](https://12factor.net/config).
The only required key is **`SQLALCHEMY_URL`**, which defines the database URL. It specifies the database driver in the URL's scheme and allows embedding driver parameters in the query string. Example:
```
sqlite+aiosqlite:////tmp/test.db
```
All parameters of sqlalchemy.create_engine can be configured by setting environment variables, with each parameter name prefixed by **`SQLALCHEMY_`**.
Note
FastSQLA is **case-insensitive** when reading environment variables, so parameter names prefixed with **`SQLALCHEMY_`** can be provided in any letter case.
#### Examples
1. PostgreSQL url using asyncpg driver with a pool_recycle of 30 minutes:
```
export SQLALCHEMY_URL=postgresql+asyncpg://postgres@localhost/postgres
export SQLALCHEMY_POOL_RECYCLE=1800
```
1. SQLite db file using aiosqlite driver with a pool_size of 50:
```
export sqlalchemy_url=sqlite+aiosqlite:///tmp/test.db
export sqlalchemy_pool_size=50
```
1. MariaDB url using aiomysql driver with echo parameter set to `True`
```
export sqlalchemy_url=mysql+aiomysql://bob:password!@db.example.com/app
export sqlalchemy_echo=true
```
## `fastsqla.new_lifespan`
Create a new lifespan async context manager.
It expects the exact same parameters as sqlalchemy.ext.asyncio.create_async_engine
Example:
```
from fastapi import FastAPI
from fastsqla import new_lifespan
lifespan = new_lifespan(
"sqlite+aiosqlite:///app/db.sqlite", connect_args={"autocommit": False}
)
app = FastAPI(lifespan=lifespan)
```
Parameters:
| Name | Type | Description | Default |
| ----- | ------ | ---------------------------------------------------------------------------------- | ------- |
| `url` | `str` | Database url. | `None` |
| `kw` | `dict` | Configuration parameters as expected by sqlalchemy.ext.asyncio.create_async_engine | `{}` |
Raises:
| Type | Description |
| --------------------------- | -------------------------------------------- |
| `MissingConfigurationError` | If a required SQLAlchemy setting is missing. |
# `SQLAlchemy` Session
## Lifecycle
[`SQLAlchemy` documentation](https://docs.sqlalchemy.org/en/20/orm/session_basics.html#when-do-i-construct-a-session-when-do-i-commit-it-and-when-do-i-close-it) recommends the following:
- Keep the lifecycle of the session **separate and external** from functions and objects that access and/or manipulate database data.
- Make sure you have a clear notion of where transactions begin and end, and keep transactions **short**, meaning, they end at the series of a sequence of operations, instead of being held open indefinitely.
`FastSQLA` automatically manages the session lifecycle:
- If the request is successful, the session is committed.
- If the request fails, the session is rolled back.
- In all cases, at the end of the request, the session is closed and the associated connection is returned to the connection pool.
To learn more about `SQLAlchemy` sessions:
- [Session Basics](https://docs.sqlalchemy.org/en/20/orm/session_basics.html#)
## `fastsqla.Session` dependency
Dependency used exclusively in endpoints to get an `SQLAlchemy` or `SQLModel` session.
`Session` is a [`FastAPI` dependency](https://fastapi.tiangolo.com/tutorial/dependencies/) that provides an asynchronous `SQLAlchemy` session or `SQLModel` one if it's installed. By defining an argument with type `Session` in an endpoint, `FastAPI` will automatically inject an async session into the endpoint.
At the end of request handling:
- If no exceptions are raised, the session is automatically committed.
- If an exception is raised, the session is automatically rolled back.
- In all cases, the session is closed and the associated connection is returned to the connection pool.
Example:
example.py
```
from fastsqla import Item, Session
...
@app.get("/heros/{hero_id}", response_model=Item[HeroItem])
async def get_items(
session: Session, # (1)!
item_id: int,
):
hero = await session.get(Hero, hero_id)
return {"data": hero}
```
1. Just define an argument with type `Session` to get an async session injected in your endpoint.
______________________________________________________________________
**Recommendation**: Unless there is a good reason to do so, avoid committing the session manually, as `FastSQLA` handles it automatically.
If you need data generated by the database server, such as auto-incremented IDs, flush the session instead:
```
from fastsqla import Item, Session
...
@app.post("/heros", response_model=Item[HeroItem])
async def create_item(session: Session, new_hero: HeroBase):
hero = Hero(**new_hero.model_dump())
session.add(hero)
await session.flush()
return {"data": hero}
```
Or use the session context manager instead.
## `fastsqla.open_session` context manager
Async context manager that opens a new `SQLAlchemy` or `SQLModel` async session.
To the contrary of the Session dependency which can only be used in endpoints, `open_session` can be used anywhere such as in background tasks.
On exit, it automatically commits the session if no errors occur inside the context, or rolls back when an exception is raised. In all cases, it closes the session and returns the associated connection to the connection pool.
Returns:
| Type | Description |
| ------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------- |
| `AsyncGenerator[AsyncSession, None]` | When SQLModel is not installed, an async generator that yields an SQLAlchemy AsyncSession. |
| `AsyncGenerator[AsyncSession, None]` | When SQLModel is installed, an async generator that yields an SQLModel AsyncSession which inherits from SQLAlchemy AsyncSession. |
```
from fastsqla import open_session
async def example():
async with open_session() as session:
await session.execute(...)
```
# Object-relational mapping
## `fastsqla.Base`
Bases: `DeclarativeBase`, `DeferredReflection`
Inherit from `Base` to declare an `SQLAlchemy` model.
Example:
```
from fastsqla import Base
from sqlalchemy.orm import Mapped, mapped_column
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]
```
To learn more on `SQLAlchemy` ORM & Declarative mapping:
- [ORM Quick Start](https://docs.sqlalchemy.org/en/20/orm/quickstart.html)
- [Declarative Mapping](https://docs.sqlalchemy.org/en/20/orm/mapping_styles.html#declarative-mapping)
Note
You don't need this if you use [`SQLModel`](http://sqlmodel.tiangolo.com/).
Source code in `src/fastsqla.py`
````
class Base(DeclarativeBase, DeferredReflection):
"""Inherit from `Base` to declare an `SQLAlchemy` model.
Example:
```py
from fastsqla import Base
from sqlalchemy.orm import Mapped, mapped_column
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]
```
To learn more on `SQLAlchemy` ORM & Declarative mapping:
* [ORM Quick Start](https://docs.sqlalchemy.org/en/20/orm/quickstart.html)
* [Declarative Mapping](https://docs.sqlalchemy.org/en/20/orm/mapping_styles.html#declarative-mapping)
!!! note
You don't need this if you use [`SQLModel`](http://sqlmodel.tiangolo.com/).
"""
__abstract__ = True
````
# Pagination
## `fastapi.Page[T]`
Bases: `Collection[T]`
Generic container that contains collection data and page metadata.
The `Page` model is used to return paginated data in paginated endpoints:
```
{
"data": list[T],
"meta": {
"offset": int,
"total_items": int,
"total_pages": int,
"page_number": int,
}
}
```
## `fastsqla.Paginate`
A dependency used in endpoints to paginate `SQLAlchemy` select queries.
It adds **`offset`** and **`limit`** query parameters to the endpoint, which are used to paginate. The model returned by the endpoint is a Page model.
## `SQLAlchemy` example
example.py
```
from fastapi import FastAPI
from fastsqla import Base, Paginate, Page, lifespan
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
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 HeroModel(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
secret_identity: str
age: int
@app.get("/heros", response_model=Page[HeroModel]) # (1)!
async def list_heros(paginate: Paginate): # (2)!
return await paginate(select(Hero)) # (3)!
```
1. The endpoint returns a `Page` model of `HeroModel`.
1. Just define an argument with type `Paginate` to get an async `paginate` function injected in your endpoint function.
1. Await the `paginate` function with the `SQLAlchemy` select statement to get the paginated result.
To add filtering, just add whatever query parameters you need to the endpoint:
```
@app.get("/heros", response_model=Page[HeroModel])
async def list_heros(paginate: Paginate, age:int | None = None):
stmt = select(Hero)
if age:
stmt = stmt.where(Hero.age == age)
return await paginate(stmt)
```
## `SQLModel` example
```
from fastapi import FastAPI
from fastsqla import Page, Paginate, Session
from sqlmodel import Field, SQLModel
from sqlalchemy import select
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))
```
# Troubleshooting
Search for the exact error text below, then apply the correction for that failure mode.
## `Missing sqlalchemy_url in environ.`
The default fastsqla.lifespan reads the database URL from `SQLALCHEMY_URL`. The variable was absent when FastAPI started.
Set an async SQLAlchemy URL before starting the application:
```
export SQLALCHEMY_URL=sqlite+aiosqlite:///db.sqlite
uvicorn example:app
```
For configuration in code, create the lifespan explicitly:
```
from fastapi import FastAPI
from fastsqla import new_lifespan
app = FastAPI(lifespan=new_lifespan("sqlite+aiosqlite:///db.sqlite"))
```
## `The asyncio extension requires an async driver to be used`
The URL selects a synchronous database driver. For example, `sqlite:///db.sqlite` loads Python's synchronous `pysqlite` driver and produces:
```
The asyncio extension requires an async driver to be used.
The loaded 'pysqlite' is not async.
```
Install the asynchronous driver for the database and include it in the URL:
| Database | Install | URL prefix |
| ---------- | ----------------------- | ----------------------- |
| PostgreSQL | `pip install asyncpg` | `postgresql+asyncpg://` |
| SQLite | `pip install aiosqlite` | `sqlite+aiosqlite:///` |
| MySQL | `pip install aiomysql` | `mysql+aiomysql://` |
## `Could not locate a bind configured on SQL expression or this Session.`
A Session or open_session() operation ran outside the FastSQLA lifespan. The lifespan binds the shared session factory at startup and clears it at shutdown.
Attach the lifespan to FastAPI:
```
from fastapi import FastAPI
from fastsqla import lifespan
app = FastAPI(lifespan=lifespan)
```
Use `Session` only in endpoint parameters. Use `open_session()` for background work that runs after application startup and finishes before shutdown:
```
from fastsqla import open_session
from sqlalchemy import select
async def refresh_cache() -> None:
async with open_session() as session:
heroes = (await session.scalars(select(Hero))).all()
```
Do not import or configure `SessionFactory`; it is an internal lifecycle detail.
## `MissingGreenlet: greenlet_spawn has not been called`
SQLAlchemy attempted implicit database I/O while ordinary Python code accessed an unloaded ORM attribute. This often happens when response serialization touches a lazy-loaded relationship.
Load relationships explicitly inside the awaited query. `selectinload()` is a good default for collections:
```
from fastsqla import Session
from sqlalchemy import select
from sqlalchemy.orm import selectinload
async def get_team(team_id: int, session: Session) -> Team:
stmt = (
select(Team)
.where(Team.id == team_id)
.options(selectinload(Team.heroes))
)
return (await session.scalars(stmt)).one()
```
See SQLAlchemy's guidance on [preventing implicit I/O with `AsyncSession`](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession).
## `Input should be greater than or equal to 0`
The built-in Paginate dependency rejects a negative `offset`. Start at zero:
```
GET /heroes?offset=0&limit=10
```
## `Input should be less than or equal to 100`
The built-in `Paginate` dependency accepts `limit` values from 1 through 100. Use a smaller value or define an intentional maximum:
```
from typing import Annotated
from fastapi import Depends
from fastsqla import PaginateType, new_pagination
LargePage = Annotated[
PaginateType[HeroModel],
Depends(new_pagination(default_page_size=10, max_page_size=250)),
]
```
## `This Session's transaction has been rolled back due to a previous exception`
A database operation such as `flush()` raised an exception, but application code caught it and then tried to keep using the invalid transaction.
Translate the original database error into an exception and let FastSQLA roll back:
```
from fastapi import HTTPException
from fastsqla import Session
from sqlalchemy.exc import IntegrityError
async def create_hero(new_hero: HeroInput, session: Session) -> Hero:
hero = Hero(**new_hero.model_dump())
session.add(hero)
try:
await session.flush()
except IntegrityError as exc:
raise HTTPException(status_code=409, detail="Hero already exists") from exc
return hero
```
Do not catch and ignore `IntegrityError`, and do not call `commit()` inside an endpoint.