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
- ⚡️ Fully Async: Leverages Python's async capabilities for non-blocking database operations.
- 📚 SQLAlchemy 2.0: Works with the latest SQLAlchemy version for robust database interactions.
- 🦾 Powerful CRUD Functionality: Full suite of efficient CRUD operations with support for joins.
- ⚙️ Dynamic Query Building: Supports building complex queries dynamically, including filtering, sorting, and pagination.
- 🤝 Advanced Join Operations: Facilitates performing SQL joins with other models with automatic join condition detection.
- 📖 Built-in Offset Pagination: Comes with ready-to-use offset pagination.
- ➤ Cursor-based Pagination: Implements efficient pagination for large datasets, ideal for infinite scrolling interfaces.
- 🤸♂️ Modular and Extensible: Designed for easy extension and customization to fit your requirements.
- 🛣️ Auto-generated Endpoints: Streamlines the process of adding CRUD endpoints with custom dependencies and configurations.
Before installing FastCRUD, ensure you have the following prerequisites:
- Python: Version 3.9 or newer.
- FastAPI: FastCRUD is built to work with FastAPI, so having FastAPI in your project is essential.
- SQLAlchemy: Version 2.0.21 or newer. FastCRUD uses SQLAlchemy for database operations.
- Pydantic: Version 2.4.1 or newer. FastCRUD leverages Pydantic models for data validation and serialization.
- SQLAlchemy-Utils: Optional, but recommended for additional SQLAlchemy utilities.
To install, just run:
pip install fastcrudOr, if using poetry:
poetry add fastcrudFastCRUD offers two primary ways to use its functionalities:
- By using
crud_routerfor automatic endpoint creation. - By integrating
FastCRUDdirectly 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: strmain.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:https:///./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:https:///./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 fashionIn 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.
- This project was heavily inspired by CRUDBase in
FastAPI Microservicesby @kludex. - Thanks @ada0l for the PyPI package name!
- flask-muck - "I'd love something like this for flask" There you have it
- FastAPI CRUD Router - Supports multiple ORMs, but currently unmantained
- FastAPI Quick CRUD - Same purpose, but only for SQLAlchemy 1.4
Igor Magalhaes – @igormagalhaesr – [email protected] github.com/igorbenav