fastapi-sa

fastapi-sa provides a simple integration between FastAPI and SQLAlchemy in your application. you can use decorators or middleware to transaction management.

Installing

install and update using pip:

pip install fastapi-sa

Examples

Create models for examples, models.py

from pydantic import BaseModel
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()


class User(Base):
    """UserModel"""
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    age = Column(Integer)


class UserSchema(BaseModel):
    """user schema"""
    id: int
    name: str
    age: int

    class Config:
        """config"""
        orm_mode = True

Database migrations for examples

code for create tables, also you can use database migrations.

from sqlalchemy import create_engine
from models import Base

engine = create_engine('sqlite+aiosqlite:////tmp/test.db')
Base.metadata.create_all(engine) 

DB init for examples

from fastapi_sa.database import db

db.init(url='sqlite+aiosqlite:////tmp/test.db')

Usage 1: fastapi middleware

from fastapi import FastAPI
from sqlalchemy import select

from fastapi_sa.database import db
from fastapi_sa.middleware import DBSessionMiddleware
from tests.example.db import User, UserSchema

app = FastAPI()
app.add_middleware(DBSessionMiddleware)


@app.get('/users')
async def get_users():
    """get all users"""
    result = await db.session.scalars(select(User))
    objs = [UserSchema.from_orm(i) for i in result.all()]
    return objs

Usage 2: other asynchronous database operations

from sqlalchemy import select

from fastapi_sa.database import db, session_ctx
from tests.example.db import User, UserSchema


@session_ctx
async def get_users():
    """get users"""
    results = await db.session.scalars(select(User))
    objs = [UserSchema.from_orm(i) for i in results.all()]
    return objs

Usage 3: with fixtures in pytest

import pytest
from fastapi_sa.database import db


@pytest.fixture()
def db_session_ctx():
    """db session context"""
    token = db.set_session_ctx()
    yield
    db.reset_session_ctx(token)


@pytest.fixture()
async def session(db_session_ctx):
    """session fixture"""
    async with db.session.begin():
        yield db.session

If you initialize data in fixture, please use

from fastapi_sa.database import db
from models import User

async with db():
    users = User(name='aoo', age=12)
    db.session.add(users)
    await db.session.flush()

if you test class methods, please use

import pytest
from sqlalchemy import func, select
from models import User, UserSchema
from fastapi_sa.database import db


class UserRepository:
    """user repository"""

    @property
    def model(self):
        """model"""
        return User

    async def get_all(self):
        """get all"""
        result = await db.session.scalars(select(self.model))
        objs = [UserSchema.from_orm(i) for i in result.all()]
        return objs


# the test case is as follows    

@pytest.fixture()
def repo():
    """repo"""
    return UserRepository()


@pytest.mark.asyncio
async def test_get_all(session, init_user, repo):
    """test get all"""
    objs = await repo.get_all()
    length = await session.scalar(select(func.count()).select_from(User))
    assert len(objs) == length

Similar design

• FastAPI-SQLAlchemy

Based on

• FastAPI
• SQLAlchemy

Develop

You may need to read the develop document to use SRC Layout in your IDE.