# 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 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`](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).
# 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 | `{}` |
# `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))
```