FastAPI-Async-SQLA

FastAPI-Async-SQLA是SQLAlchemy的扩展，用于FastAPI。它支持使用SQLAlchemy >= 2.0的异步SQLAlchemy会话，并提供分页支持。

安装

使用pip

pip install fastapi-async-sqla

快速示例

假设它运行在一个具有表user的数据库上，该表有2列id和name

# main.py
from fastapi import FastAPI, HTTPException
from fastapi_async_sqla import Base, Item, Page, Paginate, Session, lifespan
from pydantic import BaseModel
from sqlalchemy import select


app = FastAPI(lifespan=lifespan)


class User(Base):
    __tablename__ = "user"


class UserIn(BaseModel):
    name: str


class UserModel(UserIn):
    id: int


@app.get("/users", response_model=Page[UserModel])
async def list_users(paginate: Paginate):
    return await paginate(select(User))


@app.get("/users/{user_id}", response_model=Item[UserModel])
async def get_user(user_id: int, session: Session):
    user = await session.get(User, user_id)
    if user is None:
        raise HTTPException(404)
    return {"data": user}


@app.post("/users", response_model=Item[UserModel])
async def create_user(new_user: UserIn, session: Session):
    user = User(**new_user.model_dump())
    session.add(user)
    await session.flush()
    return {"data": user}

使用sqlite3创建数据库

sqlite3 db.sqlite <<EOF
CREATE TABLE user (
    id    INTEGER PRIMARY KEY AUTOINCREMENT,
    name  TEXT NOT NULL
);
EOF

安装aiosqlite以异步连接到sqlite数据库

pip install aiosqlite

运行应用程序

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