# Fastapi-SQLA
[](https://codecov.io/gh/dialoguemd/fastapi-sqla)
[](https://dl.circleci.com/status-badge/redirect/gh/dialoguemd/fastapi-sqla/tree/master)
[](https://conventionalcommits.org)
[](https://github.com/psf/black)
Fastapi-SQLA is an [SQLAlchemy] extension for [FastAPI] easy to setup with support for
pagination, asyncio, [SQLModel] and [pytest].
It supports SQLAlchemy>=1.3 and is fully compliant with [SQLAlchemy 2.0].
It is developped, maintained and used on production by the team at [@dialoguemd] with
love from Montreal 🇨🇦.
# Installing
Using [pip](https://pip.pypa.io/):
```
pip install fastapi-sqla
```
# Quick Example
Assuming it runs against a DB with a table `user` with 3 columns, `id`, `name` and
unique `email`:
```python
# main.py
from fastapi import FastAPI, HTTPException
from fastapi_sqla import Base, Item, Page, Paginate, Session, setup
from pydantic import BaseModel, EmailStr
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
app = FastAPI()
setup(app)
class User(Base):
__tablename__ = "user"
class UserIn(BaseModel):
name: str
email: EmailStr
class UserModel(UserIn):
id: int
class Config:
orm_mode = True
@app.get("/users", response_model=Page[UserModel])
def list_users(paginate: Paginate):
return paginate(select(User))
@app.get("/users/{user_id}", response_model=Item[UserModel])
def get_user(user_id: int, session: Session):
user = session.get(User, user_id)
if user is None:
raise HTTPException(404)
return {"data": user}
@app.post("/users", response_model=Item[UserModel])
def create_user(new_user: UserIn, session: Session):
user = User(**new_user.model_dump())
session.add(user)
try:
session.flush()
except IntegrityError:
raise HTTPException(409, "Email is already taken.")
return {"data": user}
```
Creating a db using `sqlite3`:
```bash
sqlite3 db.sqlite <<EOF
CREATE TABLE user (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL,
name TEXT NOT NULL
);
CREATE UNIQUE INDEX user_email_idx ON user (email);
EOF
```
Running the app:
```bash
sqlalchemy_url=sqlite:///db.sqlite?check_same_thread=false uvicorn main:app
```
# Configuration
## Environment variables:
The keys of interest in `os.environ` are prefixed with `sqlalchemy_`.
Each matching key (after the prefix is stripped) is treated as though it were the
corresponding keyword argument to [`sqlalchemy.create_engine`]
call.
The only required key is `sqlalchemy_url`, which provides the database URL, example:
```bash
export sqlalchemy_url=postgresql://postgres@localhost
```
### Multi-session support
In order to configure multiple sessions for the application,
set the environment variables with this prefix format: `fastapi_sqla__MY_KEY__`.
As with the default session, each matching key (after the prefix is stripped)
is treated as though it were the corresponding keyword argument to [`sqlalchemy.create_engine`]
call.
For example, to configure a session with the `read_only` key:
```bash
export fastapi_sqla__read_only__sqlalchemy_url=postgresql://postgres@localhost
```
### `asyncio` support using [`asyncpg`]
SQLAlchemy `>= 1.4` supports `asyncio`.
To enable `asyncio` support against a Postgres DB, install `asyncpg`:
```bash
pip install asyncpg
```
And define the environment variable `sqlalchemy_url` with `postgres+asyncpg` scheme:
```bash
export sqlalchemy_url=postgresql+asyncpg://postgres@localhost
```
## Setup the app:
```python
import fastapi_sqla
from fastapi import FastAPI
app = FastAPI()
fastapi_sqla.setup(app)
```
# SQLAlchemy
## Adding a new entity class:
```python
from fastapi_sqla import Base
class Entity(Base):
__tablename__ = "table-name-in-db"
```
## Getting an sqla session
### Using dependency injection
Use [FastAPI dependency injection] to get a session as a parameter of a path operation
function.
The SQLAlchemy session is committed before the response is returned or rollbacked if any
exception occurred:
```python
from fastapi import APIRouter
from fastapi_sqla import AsyncSession, Session
router = APIRouter()
@router.get("/example")
def example(session: Session):
return session.execute("SELECT now()").scalar()
@router.get("/async_example")
async def async_example(session: AsyncSession):
return await session.scalar("SELECT now()")
```
In order to get a session configured with a custom key:
```python
from typing import Annotated
from fastapi import APIRouter, Depends
from fastapi_sqla import (
AsyncSessionDependency,
SessionDependency,
SqlaAsyncSession,
SqlaSession,
)
router = APIRouter()
# Preferred
ReadOnlySession = Annotated[SqlaSession, Depends(SessionDependency(key="read_only"))]
AsyncReadOnlySession = Annotated[
SqlaAsyncSession, Depends(AsyncSessionDependency(key="read_only"))
]
@router.get("/example")
def example(session: ReadOnlySession):
return session.execute("SELECT now()").scalar()
@router.get("/async_example")
async def async_example(session: AsyncReadOnlySession):
return await session.scalar("SELECT now()")
# Alternative
@router.get("/example/alt")
def example_alt(session: SqlaSession = Depends(SessionDependency(key="read_only"))):
return session.execute("SELECT now()").scalar()
@router.get("/async_example/alt")
async def async_example_alt(
session: SqlaAsyncSession = Depends(AsyncSessionDependency(key="read_only")),
):
return await session.scalar("SELECT now()")
```
### Using a context manager
When needing a session outside of a path operation, like when using
[FastAPI background tasks], use `fastapi_sqla.open_session` context manager.
The SQLAlchemy session is committed when exiting context or rollbacked if any exception
occurred:
```python
from fastapi import APIRouter, BackgroundTasks
from fastapi_sqla import open_async_session, open_session
router = APIRouter()
@router.get("/example")
def example(bg: BackgroundTasks):
bg.add_task(run_bg)
bg.add_task(run_async_bg)
def run_bg():
with open_session() as session:
session.execute("SELECT now()").scalar()
def run_bg_with_key():
with open_session(key="read_only") as session:
session.execute("SELECT now()").scalar()
async def run_async_bg():
async with open_async_session() as session:
await session.scalar("SELECT now()")
async def run_async_bg_with_key():
async with open_async_session(key="read_only") as session:
await session.scalar("SELECT now()")
```
## Pagination
```python
from fastapi import APIRouter
from fastapi_sqla import Base, Page, Paginate
from pydantic import BaseModel
from sqlalchemy import select
router = APIRouter()
class User(Base):
__tablename__ = "user"
class UserModel(BaseModel):
id: int
name: str
class Config:
orm_mode = True
@router.get("/users", response_model=Page[UserModel])
def all_users(paginate: Paginate):
return paginate(select(User))
```
By default:
* It returns pages of 10 items, up to 100 items;
* Total number of items in the collection is queried using [`Query.count`].
* Response example for `/users?offset=40&limit=10`:
```json
{
"data": [
{
"id": 41,
"name": "Pat Thomas"
},
{
"id": 42,
"name": "Mulatu Astatke"
}
],
"meta": {
"offset": 40,
"total_items": 42,
"total_pages": 5,
"page_number": 5
}
}
```
### Paginating non-scalar results
To paginate a query which doesn't return [scalars], specify `scalars=False` when invoking
`paginate`:
```python
from fastapi import APIRouter
from fastapi_sqla import Base, Page, Paginate
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.orm import relationship
router = APIRouter()
class User(Base):
__tablename__ = "user"
notes = relationship("Note")
class Note(Base):
__tablename__ = "note"
class UserModel(BaseModel):
id: int
name: str
notes_count: int
@router.get("/users", response_model=Page[UserModel])
def all_users(paginate: Paginate):
query = (
select(User.id, User.name, func.count(Note.id).label("notes_count"))
.join(Note)
.group_by(User)
)
return paginate(query, scalars=False)
```
### Customize pagination
You can customize:
- Minimum and maximum number of items per pages;
- How the total number of items in the collection is queried;
To customize pagination, create a dependency using `fastapi_sqla.Pagination`:
```python
from fastapi import APIRouter, Depends
from fastapi_sqla import Base, Page, Pagination, Session
from pydantic import BaseModel
from sqlalchemy import func, select
router = APIRouter()
class User(Base):
__tablename__ = "user"
class UserModel(BaseModel):
id: int
name: str
def query_count(session: Session) -> int:
return session.execute(select(func.count()).select_from(User)).scalar()
CustomPaginate = Pagination(min_page_size=5, max_page_size=500, query_count=query_count)
@router.get("/users", response_model=Page[UserModel])
def all_users(paginate: CustomPaginate = Depends()):
return paginate(select(User))
```
### Async pagination
When using the asyncio support, use the `AsyncPaginate` dependency:
```python
from fastapi import APIRouter
from fastapi_sqla import Base, Page, AsyncPaginate
from pydantic import BaseModel
from sqlalchemy import select
router = APIRouter()
class User(Base):
__tablename__ = "user"
class UserModel(BaseModel):
id: int
name: str
class Config:
orm_mode = True
@router.get("/users", response_model=Page[UserModel])
async def all_users(paginate: AsyncPaginate):
return await paginate(select(User))
```
Customize pagination by creating a dependency using `fastapi_sqla.AsyncPagination`:
```python
from fastapi import APIRouter, Depends
from fastapi_sqla import Base, Page, AsyncPagination, AsyncSession
from pydantic import BaseModel
from sqlalchemy import func, select
router = APIRouter()
class User(Base):
__tablename__ = "user"
class UserModel(BaseModel):
id: int
name: str
async def query_count(session: AsyncSession) -> int:
result = await session.execute(select(func.count()).select_from(User))
return result.scalar()
CustomPaginate = AsyncPagination(min_page_size=5, max_page_size=500, query_count=query_count)
@router.get("/users", response_model=Page[UserModel])
async def all_users(paginate: CustomPaginate = Depends()):
return await paginate(select(User))
```
### Multi-session support
Pagination supports multiple sessions as well. To paginate using a session
configured with a custom key:
```python
from typing import Annotated
from fastapi import APIRouter, Depends
from fastapi_sqla import (
AsyncPaginateSignature,
AsyncPagination,
Base,
Page,
PaginateSignature,
Pagination,
)
from pydantic import BaseModel
from sqlalchemy import func, select
router = APIRouter()
class User(Base):
__tablename__ = "user"
class UserModel(BaseModel):
id: int
name: str
# Preferred
ReadOnlyPaginate = Annotated[
PaginateSignature, Depends(Pagination(session_key="read_only"))
]
AsyncReadOnlyPaginate = Annotated[
AsyncPaginateSignature, Depends(AsyncPagination(session_key="read_only"))
]
@router.get("/users", response_model=Page[UserModel])
def all_users(paginate: ReadOnlyPaginate):
return paginate(select(User))
@router.get("/async_users", response_model=Page[UserModel])
async def async_all_users(paginate: AsyncReadOnlyPaginate):
return await paginate(select(User))
# Alternative
@router.get("/users/alt", response_model=Page[UserModel])
def all_users_alt(
paginate: PaginateSignature = Depends(
Pagination(session_key="read_only")
),
):
return paginate(select(User))
@router.get("/async_users/alt", response_model=Page[UserModel])
async def async_all_users_alt(
paginate: AsyncPaginateSignature = Depends(
AsyncPagination(session_key="read_only")
),
):
return await paginate(select(User))
```
# SQLModel support 🎉
If your project uses [SQLModel], then `Session` dependency is an SQLModel session::
```python
from http import HTTPStatus
from fastapi import FastAPI, HTTPException
from fastapi_sqla import Item, Page, Paginate, Session, setup
from sqlmodel import Field, SQLModel, select
class Hero(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
secret_name: str
age: int | None = None
app = FastAPI()
setup(app)
@app.get("/heros", response_model=Page[Hero])
def list_hero(paginate: Paginate) -> Page[Hero]:
return paginate(select(Hero))
@app.get("/heros/{hero_id}", response_model=Item[Hero])
def get_hero(hero_id: int, session: Session) -> Item[Hero]:
hero = session.get(Hero, hero_id)
if hero is None:
raise HTTPException(HTTPStatus.NOT_FOUND)
return {"data": hero}
```
# Pytest fixtures
This library provides a set of utility fixtures, through its PyTest plugin, which is
automatically installed with the library.
By default, no records are actually written to the database when running tests.
There currently is no way to change this behaviour.
## `sqla_modules`
You must define this fixture, in order for the plugin to reflect table metadata in your
SQLAlchemy entities. It should just import all of the application's modules which contain
SQLAlchemy models.
Example:
```python
# tests/conftest.py
from pytest import fixture
@fixture
def sqla_modules():
from app import sqla # noqa
```
## `db_url`
The DB url to use.
When `CI` key is set in environment variables, it defaults to using `postgres` as the
host name:
```
postgresql://postgres@postgres/postgres
```
In other cases, the host is set to `localhost`:
```
postgresql://postgres@localhost/postgres
```
Of course, you can override it by overloading the fixture:
```python
from pytest import fixture
@fixture(scope="session")
def db_url():
return "postgresql://postgres@localhost/test_database"
```
## `async_sqlalchemy_url`
DB url to use when using `asyncio` support. Defaults to `db_url` fixture with
`postgresql+asyncpg://` scheme.
## `session` & `async_session`
Sqla sessions to create db fixture:
* All changes done at test setup or during the test are rollbacked at test tear down;
* No record will actually be written in the database;
* Changes in one regular session need to be committed to be available from other regular
sessions;
* Changes in one async session need to be committed to be available from other async
sessions;
* Changes from regular sessions are not available from `async` session and vice-versa
even when committed;
Example:
```python
from pytest import fixture
@fixture
def patient(session):
from er.sqla import Patient
patient = Patient(first_name="Bob", last_name="David")
session.add(patient)
session.commit()
return patient
@fixture
async def doctor(async_session):
from er.sqla import Doctor
doctor = Doctor(name="who")
async_session.add(doctor)
await async_session.commit()
return doctor
```
## `db_migration`
A session scope fixture that runs `alembic upgrade` at test session setup and
`alembic downgrade` at tear down.
It depends on `alembic_ini_path` fixture to get the path of `alembic.ini` file.
To use in a test or test module:
```python
from pytest import mark
pytestmark = mark.usefixtures("db_migration")
```
To use globally, add to [pytest options]:
```ini
[pytest]
usefixtures =
db_migration
```
Or depends on it in top-level `conftest.py` and mark it as _auto-used_:
```python
from pytest import fixture
@fixture(scope="session", autouse=True)
def db_migration(db_migration):
pass
```
## `alembic_ini_path`
It returns the path of `alembic.ini` configuration file. By default, it returns
`./alembic.ini`.
# Development
## Prerequisites
- **Python >=3.9**
- [**Poetry**](https://poetry.eustace.io/) to install package dependencies.
- A postgres DB reachable at `postgresql://postgres@localhost/postgres`
## Setup
```bash
$ poetry install --extras tests --extras asyncpg --extras aws_rds_iam
```
## Running tests
```bash
$ poetry run pytest
```
#### Runing tests on multiple environments
```bash
$ poetry run tox
```
[`sqlalchemy.create_engine`]: https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine
[`Query.count`]: https://docs.sqlalchemy.org/en/20/orm/queryguide/query.html#sqlalchemy.orm.Query.count
[pytest options]: https://docs.pytest.org/en/stable/reference.html#confval-usefixtures
[FastAPI]: https://fastapi.tiangolo.com/
[FastAPI dependency injection]: https://fastapi.tiangolo.com/tutorial/dependencies/
[FastAPI background tasks]: https://fastapi.tiangolo.com/tutorial/background-tasks/
[SQLAlchemy]: http://sqlalchemy.org/
[SQLAlchemy 2.0]: https://docs.sqlalchemy.org/en/20/changelog/migration_20.html
[SQLModel]: https://sqlmodel.tiangolo.com
[`asyncpg`]: https://magicstack.github.io/asyncpg/current/
[scalars]: https://docs.sqlalchemy.org/en/20/core/connections.html#sqlalchemy.engine.Result.scalars
[alembic]: https://alembic.sqlalchemy.org/
[pytest]: https://docs.pytest.org/
[@dialoguemd]: https://github.com/dialoguemd
Raw data
{
"_id": null,
"home_page": "https://github.com/dialoguemd/fastapi-sqla",
"name": "fastapi-sqla",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.9,<4.0",
"maintainer_email": "",
"keywords": "FastAPI,SQLAlchemy,asyncio,pytest,alembic",
"author": "Hadrien David",
"author_email": "hadrien.david@dialogue.co",
"download_url": "https://files.pythonhosted.org/packages/d9/28/e8713da321a0ce32f4c11184969c9de4c13045e528ed60d9520189343525/fastapi_sqla-3.1.1.tar.gz",
"platform": null,
"description": "# Fastapi-SQLA\n\n[](https://codecov.io/gh/dialoguemd/fastapi-sqla)\n[](https://dl.circleci.com/status-badge/redirect/gh/dialoguemd/fastapi-sqla/tree/master)\n\n[](https://conventionalcommits.org)\n[](https://github.com/psf/black)\n\n\nFastapi-SQLA is an [SQLAlchemy] extension for [FastAPI] easy to setup with support for\npagination, asyncio, [SQLModel] and [pytest].\nIt supports SQLAlchemy>=1.3 and is fully compliant with [SQLAlchemy 2.0].\nIt is developped, maintained and used on production by the team at [@dialoguemd] with\nlove from Montreal \ud83c\udde8\ud83c\udde6.\n\n# Installing\n\nUsing [pip](https://pip.pypa.io/):\n```\npip install fastapi-sqla\n```\n\n# Quick Example\n\nAssuming it runs against a DB with a table `user` with 3 columns, `id`, `name` and\nunique `email`:\n\n```python\n# main.py\nfrom fastapi import FastAPI, HTTPException\nfrom fastapi_sqla import Base, Item, Page, Paginate, Session, setup\nfrom pydantic import BaseModel, EmailStr\nfrom sqlalchemy import select\nfrom sqlalchemy.exc import IntegrityError\n\napp = FastAPI()\n\nsetup(app)\n\n\nclass User(Base):\n __tablename__ = \"user\"\n\n\nclass UserIn(BaseModel):\n name: str\n email: EmailStr\n\n\nclass UserModel(UserIn):\n id: int\n\n class Config:\n orm_mode = True\n\n\n@app.get(\"/users\", response_model=Page[UserModel])\ndef list_users(paginate: Paginate):\n return paginate(select(User))\n\n\n@app.get(\"/users/{user_id}\", response_model=Item[UserModel])\ndef get_user(user_id: int, session: Session):\n user = session.get(User, user_id)\n if user is None:\n raise HTTPException(404)\n return {\"data\": user}\n\n\n@app.post(\"/users\", response_model=Item[UserModel])\ndef create_user(new_user: UserIn, session: Session):\n user = User(**new_user.model_dump())\n session.add(user)\n try:\n session.flush()\n except IntegrityError:\n raise HTTPException(409, \"Email is already taken.\")\n return {\"data\": user}\n```\n\nCreating a db using `sqlite3`:\n```bash\nsqlite3 db.sqlite <<EOF\nCREATE TABLE user (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n email TEXT NOT NULL,\n name TEXT NOT NULL\n);\nCREATE UNIQUE INDEX user_email_idx ON user (email);\nEOF\n```\n\nRunning the app:\n```bash\nsqlalchemy_url=sqlite:///db.sqlite?check_same_thread=false uvicorn main:app\n```\n\n# Configuration\n\n## Environment variables:\n\nThe keys of interest in `os.environ` are prefixed with `sqlalchemy_`.\nEach matching key (after the prefix is stripped) is treated as though it were the\ncorresponding keyword argument to [`sqlalchemy.create_engine`]\ncall.\n\nThe only required key is `sqlalchemy_url`, which provides the database URL, example:\n\n```bash\nexport sqlalchemy_url=postgresql://postgres@localhost\n```\n\n### Multi-session support\n\nIn order to configure multiple sessions for the application, \nset the environment variables with this prefix format: `fastapi_sqla__MY_KEY__`.\n\nAs with the default session, each matching key (after the prefix is stripped) \nis treated as though it were the corresponding keyword argument to [`sqlalchemy.create_engine`]\ncall.\n\nFor example, to configure a session with the `read_only` key:\n\n```bash\nexport fastapi_sqla__read_only__sqlalchemy_url=postgresql://postgres@localhost\n```\n\n### `asyncio` support using [`asyncpg`]\n\nSQLAlchemy `>= 1.4` supports `asyncio`.\nTo enable `asyncio` support against a Postgres DB, install `asyncpg`:\n\n```bash\npip install asyncpg\n```\n\nAnd define the environment variable `sqlalchemy_url` with `postgres+asyncpg` scheme:\n\n```bash\nexport sqlalchemy_url=postgresql+asyncpg://postgres@localhost\n```\n\n## Setup the app:\n\n```python\nimport fastapi_sqla\nfrom fastapi import FastAPI\n\napp = FastAPI()\nfastapi_sqla.setup(app)\n```\n\n# SQLAlchemy\n\n## Adding a new entity class:\n\n```python\nfrom fastapi_sqla import Base\n\n\nclass Entity(Base):\n __tablename__ = \"table-name-in-db\"\n```\n\n## Getting an sqla session\n\n### Using dependency injection\n\nUse [FastAPI dependency injection] to get a session as a parameter of a path operation\nfunction.\n\nThe SQLAlchemy session is committed before the response is returned or rollbacked if any\nexception occurred:\n\n```python\nfrom fastapi import APIRouter\nfrom fastapi_sqla import AsyncSession, Session\n\nrouter = APIRouter()\n\n\n@router.get(\"/example\")\ndef example(session: Session):\n return session.execute(\"SELECT now()\").scalar()\n\n\n@router.get(\"/async_example\")\nasync def async_example(session: AsyncSession):\n return await session.scalar(\"SELECT now()\")\n```\n\nIn order to get a session configured with a custom key:\n\n```python\nfrom typing import Annotated\n\nfrom fastapi import APIRouter, Depends\nfrom fastapi_sqla import (\n AsyncSessionDependency,\n SessionDependency,\n SqlaAsyncSession,\n SqlaSession,\n)\n\nrouter = APIRouter()\n\n\n# Preferred\n\nReadOnlySession = Annotated[SqlaSession, Depends(SessionDependency(key=\"read_only\"))]\nAsyncReadOnlySession = Annotated[\n SqlaAsyncSession, Depends(AsyncSessionDependency(key=\"read_only\"))\n]\n\n@router.get(\"/example\")\ndef example(session: ReadOnlySession):\n return session.execute(\"SELECT now()\").scalar()\n\n\n@router.get(\"/async_example\")\nasync def async_example(session: AsyncReadOnlySession):\n return await session.scalar(\"SELECT now()\")\n\n\n# Alternative\n\n@router.get(\"/example/alt\")\ndef example_alt(session: SqlaSession = Depends(SessionDependency(key=\"read_only\"))):\n return session.execute(\"SELECT now()\").scalar()\n\n\n@router.get(\"/async_example/alt\")\nasync def async_example_alt(\n session: SqlaAsyncSession = Depends(AsyncSessionDependency(key=\"read_only\")),\n):\n return await session.scalar(\"SELECT now()\")\n```\n\n### Using a context manager\n\nWhen needing a session outside of a path operation, like when using\n[FastAPI background tasks], use `fastapi_sqla.open_session` context manager.\nThe SQLAlchemy session is committed when exiting context or rollbacked if any exception\noccurred:\n\n```python\nfrom fastapi import APIRouter, BackgroundTasks\nfrom fastapi_sqla import open_async_session, open_session\n\nrouter = APIRouter()\n\n\n@router.get(\"/example\")\ndef example(bg: BackgroundTasks):\n bg.add_task(run_bg)\n bg.add_task(run_async_bg)\n\n\ndef run_bg():\n with open_session() as session:\n session.execute(\"SELECT now()\").scalar()\n\ndef run_bg_with_key():\n with open_session(key=\"read_only\") as session:\n session.execute(\"SELECT now()\").scalar()\n\nasync def run_async_bg():\n async with open_async_session() as session:\n await session.scalar(\"SELECT now()\")\n\nasync def run_async_bg_with_key():\n async with open_async_session(key=\"read_only\") as session:\n await session.scalar(\"SELECT now()\")\n```\n\n## Pagination\n\n```python\nfrom fastapi import APIRouter\nfrom fastapi_sqla import Base, Page, Paginate\nfrom pydantic import BaseModel\nfrom sqlalchemy import select\n\nrouter = APIRouter()\n\n\nclass User(Base):\n __tablename__ = \"user\"\n\n\nclass UserModel(BaseModel):\n id: int\n name: str\n\n class Config:\n orm_mode = True\n\n\n@router.get(\"/users\", response_model=Page[UserModel])\ndef all_users(paginate: Paginate):\n return paginate(select(User))\n```\n\nBy default:\n\n* It returns pages of 10 items, up to 100 items;\n* Total number of items in the collection is queried using [`Query.count`].\n* Response example for `/users?offset=40&limit=10`:\n\n ```json\n {\n \"data\": [\n {\n \"id\": 41,\n \"name\": \"Pat Thomas\"\n },\n {\n \"id\": 42,\n \"name\": \"Mulatu Astatke\"\n }\n ],\n \"meta\": {\n \"offset\": 40,\n \"total_items\": 42,\n \"total_pages\": 5,\n \"page_number\": 5\n }\n }\n ```\n\n### Paginating non-scalar results\n\nTo paginate a query which doesn't return [scalars], specify `scalars=False` when invoking\n`paginate`:\n\n```python\nfrom fastapi import APIRouter\nfrom fastapi_sqla import Base, Page, Paginate\nfrom pydantic import BaseModel\nfrom sqlalchemy import func, select\nfrom sqlalchemy.orm import relationship\n\nrouter = APIRouter()\n\n\nclass User(Base):\n __tablename__ = \"user\"\n notes = relationship(\"Note\")\n\n\nclass Note(Base):\n __tablename__ = \"note\"\n\n\nclass UserModel(BaseModel):\n id: int\n name: str\n notes_count: int\n\n\n@router.get(\"/users\", response_model=Page[UserModel])\ndef all_users(paginate: Paginate):\n query = (\n select(User.id, User.name, func.count(Note.id).label(\"notes_count\"))\n .join(Note)\n .group_by(User)\n )\n return paginate(query, scalars=False)\n```\n\n\n### Customize pagination\n\nYou can customize:\n- Minimum and maximum number of items per pages;\n- How the total number of items in the collection is queried;\n\nTo customize pagination, create a dependency using `fastapi_sqla.Pagination`:\n\n```python\nfrom fastapi import APIRouter, Depends\nfrom fastapi_sqla import Base, Page, Pagination, Session\nfrom pydantic import BaseModel\nfrom sqlalchemy import func, select\n\nrouter = APIRouter()\n\n\nclass User(Base):\n __tablename__ = \"user\"\n\n\nclass UserModel(BaseModel):\n id: int\n name: str\n\n\ndef query_count(session: Session) -> int:\n return session.execute(select(func.count()).select_from(User)).scalar()\n\n\nCustomPaginate = Pagination(min_page_size=5, max_page_size=500, query_count=query_count)\n\n\n@router.get(\"/users\", response_model=Page[UserModel])\ndef all_users(paginate: CustomPaginate = Depends()):\n return paginate(select(User))\n```\n\n### Async pagination\n\nWhen using the asyncio support, use the `AsyncPaginate` dependency:\n\n```python\nfrom fastapi import APIRouter\nfrom fastapi_sqla import Base, Page, AsyncPaginate\nfrom pydantic import BaseModel\nfrom sqlalchemy import select\n\nrouter = APIRouter()\n\n\nclass User(Base):\n __tablename__ = \"user\"\n\n\nclass UserModel(BaseModel):\n id: int\n name: str\n\n class Config:\n orm_mode = True\n\n\n@router.get(\"/users\", response_model=Page[UserModel])\nasync def all_users(paginate: AsyncPaginate):\n return await paginate(select(User))\n```\n\nCustomize pagination by creating a dependency using `fastapi_sqla.AsyncPagination`:\n\n```python\nfrom fastapi import APIRouter, Depends\nfrom fastapi_sqla import Base, Page, AsyncPagination, AsyncSession\nfrom pydantic import BaseModel\nfrom sqlalchemy import func, select\n\nrouter = APIRouter()\n\n\nclass User(Base):\n __tablename__ = \"user\"\n\n\nclass UserModel(BaseModel):\n id: int\n name: str\n\n\nasync def query_count(session: AsyncSession) -> int:\n result = await session.execute(select(func.count()).select_from(User))\n return result.scalar()\n\n\nCustomPaginate = AsyncPagination(min_page_size=5, max_page_size=500, query_count=query_count)\n\n\n@router.get(\"/users\", response_model=Page[UserModel])\nasync def all_users(paginate: CustomPaginate = Depends()):\n return await paginate(select(User))\n```\n\n### Multi-session support\n\nPagination supports multiple sessions as well. To paginate using a session \nconfigured with a custom key:\n\n```python\nfrom typing import Annotated\n\nfrom fastapi import APIRouter, Depends\nfrom fastapi_sqla import (\n AsyncPaginateSignature,\n AsyncPagination,\n Base,\n Page,\n PaginateSignature,\n Pagination,\n)\nfrom pydantic import BaseModel\nfrom sqlalchemy import func, select\n\nrouter = APIRouter()\n\n\nclass User(Base):\n __tablename__ = \"user\"\n\n\nclass UserModel(BaseModel):\n id: int\n name: str\n\n\n# Preferred\n\nReadOnlyPaginate = Annotated[\n PaginateSignature, Depends(Pagination(session_key=\"read_only\"))\n]\nAsyncReadOnlyPaginate = Annotated[\n AsyncPaginateSignature, Depends(AsyncPagination(session_key=\"read_only\"))\n]\n\n@router.get(\"/users\", response_model=Page[UserModel])\ndef all_users(paginate: ReadOnlyPaginate):\n return paginate(select(User))\n\n@router.get(\"/async_users\", response_model=Page[UserModel])\nasync def async_all_users(paginate: AsyncReadOnlyPaginate):\n return await paginate(select(User))\n\n\n# Alternative\n\n@router.get(\"/users/alt\", response_model=Page[UserModel])\ndef all_users_alt(\n paginate: PaginateSignature = Depends(\n Pagination(session_key=\"read_only\")\n ),\n):\n return paginate(select(User))\n\n@router.get(\"/async_users/alt\", response_model=Page[UserModel])\nasync def async_all_users_alt(\n paginate: AsyncPaginateSignature = Depends(\n AsyncPagination(session_key=\"read_only\")\n ),\n):\n return await paginate(select(User))\n```\n\n# SQLModel support \ud83c\udf89\n\nIf your project uses [SQLModel], then `Session` dependency is an SQLModel session::\n\n```python\n from http import HTTPStatus\n\n from fastapi import FastAPI, HTTPException\n from fastapi_sqla import Item, Page, Paginate, Session, setup\n from sqlmodel import Field, SQLModel, select\n\n class Hero(SQLModel, table=True):\n id: int | None = Field(default=None, primary_key=True)\n name: str\n secret_name: str\n age: int | None = None\n\n\n app = FastAPI()\n setup(app)\n\n @app.get(\"/heros\", response_model=Page[Hero])\n def list_hero(paginate: Paginate) -> Page[Hero]:\n return paginate(select(Hero))\n\n\n @app.get(\"/heros/{hero_id}\", response_model=Item[Hero])\n def get_hero(hero_id: int, session: Session) -> Item[Hero]:\n hero = session.get(Hero, hero_id)\n if hero is None:\n raise HTTPException(HTTPStatus.NOT_FOUND)\n return {\"data\": hero}\n\n```\n\n# Pytest fixtures\n\nThis library provides a set of utility fixtures, through its PyTest plugin, which is\nautomatically installed with the library.\n\nBy default, no records are actually written to the database when running tests.\nThere currently is no way to change this behaviour.\n\n## `sqla_modules`\n\nYou must define this fixture, in order for the plugin to reflect table metadata in your\nSQLAlchemy entities. It should just import all of the application's modules which contain\nSQLAlchemy models.\n\nExample:\n\n```python\n# tests/conftest.py\nfrom pytest import fixture\n\n\n@fixture\ndef sqla_modules():\n from app import sqla # noqa\n```\n\n## `db_url`\n\nThe DB url to use.\n\nWhen `CI` key is set in environment variables, it defaults to using `postgres` as the\nhost name:\n\n```\npostgresql://postgres@postgres/postgres\n```\n\nIn other cases, the host is set to `localhost`:\n\n```\npostgresql://postgres@localhost/postgres\n```\n\nOf course, you can override it by overloading the fixture:\n\n```python\nfrom pytest import fixture\n\n\n@fixture(scope=\"session\")\ndef db_url():\n return \"postgresql://postgres@localhost/test_database\"\n```\n\n## `async_sqlalchemy_url`\n\nDB url to use when using `asyncio` support. Defaults to `db_url` fixture with\n`postgresql+asyncpg://` scheme.\n\n\n## `session` & `async_session`\n\nSqla sessions to create db fixture:\n* All changes done at test setup or during the test are rollbacked at test tear down;\n* No record will actually be written in the database;\n* Changes in one regular session need to be committed to be available from other regular\n sessions;\n* Changes in one async session need to be committed to be available from other async\n sessions;\n* Changes from regular sessions are not available from `async` session and vice-versa\n even when committed;\n\nExample:\n```python\nfrom pytest import fixture\n\n\n@fixture\ndef patient(session):\n from er.sqla import Patient\n patient = Patient(first_name=\"Bob\", last_name=\"David\")\n session.add(patient)\n session.commit()\n return patient\n\n\n@fixture\nasync def doctor(async_session):\n from er.sqla import Doctor\n doctor = Doctor(name=\"who\")\n async_session.add(doctor)\n await async_session.commit()\n return doctor\n```\n\n## `db_migration`\n\nA session scope fixture that runs `alembic upgrade` at test session setup and\n`alembic downgrade` at tear down.\n\nIt depends on `alembic_ini_path` fixture to get the path of `alembic.ini` file.\n\nTo use in a test or test module:\n\n```python\nfrom pytest import mark\n\npytestmark = mark.usefixtures(\"db_migration\")\n```\n\nTo use globally, add to [pytest options]:\n\n ```ini\n [pytest]\n usefixtures =\n db_migration\n ```\n\nOr depends on it in top-level `conftest.py` and mark it as _auto-used_:\n\n```python\nfrom pytest import fixture\n\n\n@fixture(scope=\"session\", autouse=True)\ndef db_migration(db_migration):\n pass\n```\n\n## `alembic_ini_path`\n\nIt returns the path of `alembic.ini` configuration file. By default, it returns\n`./alembic.ini`.\n\n\n# Development\n\n## Prerequisites\n\n- **Python >=3.9**\n- [**Poetry**](https://poetry.eustace.io/) to install package dependencies.\n- A postgres DB reachable at `postgresql://postgres@localhost/postgres`\n\n\n## Setup\n\n```bash\n$ poetry install --extras tests --extras asyncpg --extras aws_rds_iam\n```\n\n## Running tests\n\n```bash\n$ poetry run pytest\n```\n\n#### Runing tests on multiple environments\n\n```bash\n$ poetry run tox\n```\n\n[`sqlalchemy.create_engine`]: https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine\n[`Query.count`]: https://docs.sqlalchemy.org/en/20/orm/queryguide/query.html#sqlalchemy.orm.Query.count\n[pytest options]: https://docs.pytest.org/en/stable/reference.html#confval-usefixtures\n[FastAPI]: https://fastapi.tiangolo.com/\n[FastAPI dependency injection]: https://fastapi.tiangolo.com/tutorial/dependencies/\n[FastAPI background tasks]: https://fastapi.tiangolo.com/tutorial/background-tasks/\n[SQLAlchemy]: http://sqlalchemy.org/\n[SQLAlchemy 2.0]: https://docs.sqlalchemy.org/en/20/changelog/migration_20.html\n[SQLModel]: https://sqlmodel.tiangolo.com\n[`asyncpg`]: https://magicstack.github.io/asyncpg/current/\n[scalars]: https://docs.sqlalchemy.org/en/20/core/connections.html#sqlalchemy.engine.Result.scalars\n[alembic]: https://alembic.sqlalchemy.org/\n[pytest]: https://docs.pytest.org/\n[@dialoguemd]: https://github.com/dialoguemd\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "SQLAlchemy extension for FastAPI with support for pagination, asyncio, and pytest, ready for production.",
"version": "3.1.1",
"project_urls": {
"Homepage": "https://github.com/dialoguemd/fastapi-sqla",
"Repository": "https://github.com/dialoguemd/fastapi-sqla"
},
"split_keywords": [
"fastapi",
"sqlalchemy",
"asyncio",
"pytest",
"alembic"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "51cb98c1ec4291ba8c4532c1a20934e1489d6bbab65da86633f9388dffa0becf",
"md5": "6ca0120b6b28d49317eea82474589c13",
"sha256": "1d389e0da1a05d61246d05f2c7e19d450f5eec73a1e0e7edea0bb9d874acb55b"
},
"downloads": -1,
"filename": "fastapi_sqla-3.1.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "6ca0120b6b28d49317eea82474589c13",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9,<4.0",
"size": 19863,
"upload_time": "2024-02-02T21:26:27",
"upload_time_iso_8601": "2024-02-02T21:26:27.729117Z",
"url": "https://files.pythonhosted.org/packages/51/cb/98c1ec4291ba8c4532c1a20934e1489d6bbab65da86633f9388dffa0becf/fastapi_sqla-3.1.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "d928e8713da321a0ce32f4c11184969c9de4c13045e528ed60d9520189343525",
"md5": "bb91b5ab3be5847991dc64df5d43bbdb",
"sha256": "ed29677197d4e02f05554160407d5b4079d2af3d250a3d2c5b12477bdbf8137c"
},
"downloads": -1,
"filename": "fastapi_sqla-3.1.1.tar.gz",
"has_sig": false,
"md5_digest": "bb91b5ab3be5847991dc64df5d43bbdb",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9,<4.0",
"size": 19839,
"upload_time": "2024-02-02T21:26:35",
"upload_time_iso_8601": "2024-02-02T21:26:35.258311Z",
"url": "https://files.pythonhosted.org/packages/d9/28/e8713da321a0ce32f4c11184969c9de4c13045e528ed60d9520189343525/fastapi_sqla-3.1.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-02-02 21:26:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "dialoguemd",
"github_project": "fastapi-sqla",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"circle": true,
"lcname": "fastapi-sqla"
}
JSON
