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.

Minimal Example

Assuming you have your model, schemas and database connection:

# imports here

# define your 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())



# your schemas
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)

Use crud_router and include it in your FastAPI application

from fastcrud import crud_router

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

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)

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

app.include_router(item_router)

And it's all done, just go to /docs and the crud endpoints are created.

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 poetry:

poetry 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