FastCRUD

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.

---

Features

• 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.
• SQLModel Support: You can optionally use SQLModel 0.14 or newer instead of SQLAlchemy.
• 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.
• Clean Architecture: Built with a six-level dependency hierarchy that prevents circular dependencies and supports future extensibility.

Quick Start

Using SQLModel?

Since SQLModel combines SQLAlchemy and Pydantic, you can replace the separate model and schema definitions below with SQLModel classes. Wherever you see SQLAlchemy models or Pydantic schemas in the documentation, just use SQLModel instead - it works seamlessly with FastCRUD.

1. Define Your Model and Schema

models.py

import datetime
from sqlalchemy import Column, DateTime, Integer, Numeric, String, func
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from pydantic import BaseModel

class Base(DeclarativeBase):
    pass

# SQLAlchemy Model
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    description = Column(String)
    category = Column(String)
    price = Column(Numeric)
    last_sold = Column(DateTime)
    created_at = Column(DateTime, default=func.now())



# Pydantic Schema
class ItemSchema(BaseModel):
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None



# Database connection
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

2. Set Up FastAPI with FastCRUD

main.py

from typing import AsyncGenerator
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
from fastcrud import crud_router

# Database session dependency
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

# Create tables before 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)

# Create CRUD router
item_router = crud_router(
    session=get_session,
    model=Item,
    create_schema=ItemSchema,
    update_schema=ItemSchema,
    path="/items",
    tags=["Items"]
)

app.include_router(item_router)

3. That's It!

Start your FastAPI application and visit /docs - your CRUD endpoints are automatically created and documented!

Requirements

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 or SQLModel: FastCRUD uses SQLAlchemy 2.0 for database operations, so you need SQLAlchemy 2.0 or newer or SQLModel 0.14 or newer.
• Pydantic V2 or SQLModel: FastCRUD leverages Pydantic models for data validation and serialization, so you need Pydantic 2.0 or newer or SQLModel 0.14 or newer.

Installing

To install, just run:

pip install fastcrud

Or, if using UV:

uv add fastcrud

Usage

FastCRUD offers two primary ways to use its functionalities:

1. By using crud_router for automatic endpoint creation.
2. By integrating FastCRUD directly into your FastAPI endpoints for more control.

Below are examples demonstrating both approaches:

Using crud_router for Automatic Endpoint Creation

Here's a quick example to get you started:

Define Your Model and Schemas

item/model.py

from sqlalchemy import Column, DateTime, Integer, Numeric, String, func
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)
    category = Column(String)
    price = Column(Numeric)
    last_sold = Column(DateTime)
    created_at = Column(DateTime, default=func.now())

item/schemas.py

import datetime

from pydantic import BaseModel


class CreateItemSchema(BaseModel):
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None


class UpdateItemSchema(BaseModel):
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None

Set Up FastAPI and FastCRUD

main.py

from typing import AsyncGenerator

from fastapi import FastAPI
from fastcrud import crud_router
from fastcrud import FastCRUD
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from .item.model import Base, Item
from .item.schemas import CreateItemSchema, UpdateItemSchema

# 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 operations setup
item_crud = FastCRUD(Item)

# CRUD router setup
item_router = crud_router(
    session=get_session,
    model=Item,
    create_schema=CreateItemSchema,
    update_schema=UpdateItemSchema,
    crud=item_crud,
    path="/items",
    tags=["Items"],
)

app.include_router(item_router)

Using FastCRUD in User-Defined FastAPI Endpoints

For more control over your endpoints, you can use FastCRUD directly within your custom FastAPI route functions. Here's an example:

api/v1/items.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from fastcrud import FastCRUD

from .item.model import Item
from .item.schemas import CreateItemSchema, UpdateItemSchema

# Assume async_session is already set up as per the previous example

# Instantiate FastCRUD with your model
item_crud = FastCRUD(Item)

@app.post("/custom/items/")
async def create_item(item_data: CreateItemSchema, 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

License

MIT

---

Why Choose FastCRUD? →