Productivity
Powerful CRUD methods and automatic endpoint creation for FastAPI.
FastCRUD is a Python package for FastAPI, offering robust async CRUD operations and flexible endpoint creation utilities, streamlined through advanced features like auto-detected join conditions, dynamic sorting, and offset and cursor pagination.
Documentation: igorbenav.github.io/fastcrud
Before installing FastCRUD, ensure you have the following prerequisites:
To install, just run:
pip install fastcrud
Or, if using poetry:
poetry add fastcrud
FastCRUD offers two primary ways to use its functionalities:
crud_router
for automatic endpoint creation.FastCRUD
directly into your FastAPI endpoints for more control.Below are examples demonstrating both approaches:
Here's a quick example to get you started:
models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
class Item(Base):
__tablename__ = 'items'
id = Column(Integer, primary_key=True)
name = Column(String)
description = Column(String)
schemas.py
from pydantic import BaseModel
class ItemCreateSchema(BaseModel):
name: str
description: str
class ItemUpdateSchema(BaseModel):
name: str
description: str
main.py
from typing import AsyncGenerator
from fastapi import FastAPI
from fastcrud import FastCRUD, crud_router
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from yourapp.models import Base, Item
from yourapp.schemas import ItemCreateSchema, ItemUpdateSchema
# Database setup (Async SQLAlchemy)
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# Database session dependency
async def get_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
yield session
# Create tables before the app start
async def lifespan(app: FastAPI):
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# FastAPI app
app = FastAPI(lifespan=lifespan)
# CRUD router setup
item_router = crud_router(
session=get_session,
model=Item,
create_schema=ItemCreateSchema,
update_schema=ItemUpdateSchema,
path="/items",
tags=["Items"],
)
app.include_router(item_router)
For more control over your endpoints, you can use FastCRUD directly within your custom FastAPI route functions. Here's an example:
main.py
from typing import AsyncGenerator
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from fastcrud import FastCRUD
from models import Base, Item
from schemas import ItemCreateSchema, ItemUpdateSchema
# Database setup (Async SQLAlchemy)
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# Database session dependency
async def get_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
yield session
# Create tables before the app start
async def lifespan(app: FastAPI):
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# FastAPI app
app = FastAPI(lifespan=lifespan)
# Instantiate FastCRUD with your model
item_crud = FastCRUD(Item)
@app.post("/custom/items/")
async def create_item(
item_data: ItemCreateSchema, db: AsyncSession = Depends(get_session)
):
return await item_crud.create(db, item_data)
@app.get("/custom/items/{item_id}")
async def read_item(item_id: int, db: AsyncSession = Depends(get_session)):
item = await item_crud.get(db, id=item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return item
# You can add more routes for update and delete operations in a similar fashion
In this example, we define custom endpoints for creating and reading items using FastCRUD directly, providing more flexibility in how the endpoints are structured and how the responses are handled.
To read more detailed descriptions, go to the documentation.
FastAPI Microservices
by @kludex.Igor Magalhaes β @igormagalhaesr β igormagalhaesr@gmail.com github.com/igorbenav