Skip to content

Using FastCRUD for Enhanced CRUD Operations

FastCRUD is a versatile tool for handling CRUD (Create, Read, Update, Delete) operations in FastAPI applications with SQLAlchemy models. It leverages Pydantic schemas for data validation and serialization, offering a streamlined approach to database interactions.

Key Features

  • Simplified CRUD operations with SQLAlchemy models.
  • Data validation and serialization using Pydantic.
  • Support for complex queries including joins and pagination.

Getting Started

Step 1: Define Models and Schemas

Define your SQLAlchemy models and Pydantic schemas for data representation.

Models and Schemas Used Below
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
customer/model.py 
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class Customer(Base):
    __tablename__ = "customer"
    id = Column(Integer, primary_key=True)
    name = Column(String)
customer/schemas.py 
from pydantic import BaseModel


class ReadCustomerSchema(BaseModel):
    id: int
    name: str | None = None
product/model.py 
from sqlalchemy import Column, Integer, Numeric, String
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "product"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Numeric)
order/model.py 
from sqlalchemy import Column, ForeignKey, Integer
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class Order(Base):
    __tablename__ = "order"
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey("customer.id"))
    product_id = Column(Integer, ForeignKey("product.id"))
    quantity = Column(Integer)
order/schemas.py 
from pydantic import BaseModel


class ReadOrderSchema(BaseModel):
    id: int
    customer_id: int | None = None
    product_id: int | None = None
    quantity: int | None = None

Step 2: Initialize FastCRUD

Create a FastCRUD instance for your model to handle CRUD operations.

from fastcrud import FastCRUD

# Creating a FastCRUD instance
item_crud = FastCRUD(Item)
order_crud = FastCRUD(Order)

Step 3: Pick your Method

Then you just pick the method you need and use it like this:

# Creating a new record
new_record = await item_crud.create(db_session, create_schema_instance)

More on available methods below.

Understanding FastCRUD Methods

FastCRUD offers a comprehensive suite of methods for CRUD operations, each designed to handle different aspects of database interactions efficiently.

1. Create

create(
    db: AsyncSession,
    object: CreateSchemaType,
    commit: bool = True,
) -> ModelType

Purpose: To create a new record in the database.
Usage Example: Creates an item with name "New Item".

new_item = await item_crud.create(db, CreateItemSchema(name="New Item"))

Warning

Note that naive datetime such as datetime.utcnow is not supported by FastCRUD as it was deprecated.

Use timezone aware datetime, such as datetime.now(UTC) instead.

2. Get

get(
    db: AsyncSession,
    schema_to_select: Optional[type[BaseModel]] = None,
    return_as_model: bool = False,
    one_or_none: bool = False,
    **kwargs: Any,
) -> Optional[Union[dict, BaseModel]]

Purpose: To fetch a single record based on filters, with an option to select specific columns using a Pydantic schema.
Usage Example: Fetches the item with item_id as its id.

item = await item_crud.get(db, id=item_id)

3. Exists

exists(
    db: AsyncSession,
    **kwargs: Any,
) -> bool

Purpose: To check if a record exists based on provided filters.
Usage Example: Checks whether an item with name "Existing Item" exists.

exists = await item_crud.exists(db, name="Existing Item")

4. Count

count(
    db: AsyncSession,
    joins_config: Optional[list[JoinConfig]] = None,
    **kwargs: Any,
) -> int

Purpose: To count the number of records matching provided filters.
Usage Example: Counts the number of items with the "Books" category.

count = await item_crud.count(db, category="Books")

5. Get Multi

get_multi(
    db: AsyncSession,
    offset: int = 0,
    limit: Optional[int] = 100,
    schema_to_select: Optional[type[BaseModel]] = None,
    sort_columns: Optional[Union[str, list[str]]] = None,
    sort_orders: Optional[Union[str, list[str]]] = None,
    return_as_model: bool = False,
    return_total_count: bool = True,
    **kwargs: Any,
) -> dict[str, Any]

Purpose: To fetch multiple records with optional sorting, pagination, and model conversion.
Usage Example: Fetches a subset of 5 items, starting from the 11th item in the database.

items = await item_crud.get_multi(db, offset=10, limit=5)

6. Update

update(
    db: AsyncSession, 
    object: Union[UpdateSchemaType, dict[str, Any]], 
    allow_multiple: bool = False,
    commit: bool = True,
    return_columns: Optional[list[str]] = None,
    schema_to_select: Optional[type[BaseModel]] = None,
    return_as_model: bool = False,
    one_or_none: bool = False,
    **kwargs: Any,
) -> Optional[Union[dict, BaseModel]]

Purpose: To update an existing record in the database.
Usage Example: Updates the description of the item with item_id as its id.

await item_crud.update(
    db,
    UpdateItemSchema(description="Updated"),
    id=item_id,
)

7. Delete

delete(
    db: AsyncSession, 
    db_row: Optional[Row] = None, 
    allow_multiple: bool = False,
    commit: bool = True,
    **kwargs: Any,
) -> None

Purpose: To delete a record from the database, with support for soft delete.
Usage Example: Deletes the item with item_id as its id, performs a soft delete if the model has the is_deleted column.

await item_crud.delete(db, id=item_id)

8. Hard Delete

db_delete(
    db: AsyncSession, 
    allow_multiple: bool = False,
    commit: bool = True,
    **kwargs: Any,
) -> None

Purpose: To hard delete a record from the database.
Usage Example: Hard deletes the item with item_id as its id.

await item_crud.db_delete(db, id=item_id)

Advanced Methods for Complex Queries and Joins

FastCRUD extends its functionality with advanced methods tailored for complex query operations and handling joins. These methods cater to specific use cases where more sophisticated data retrieval and manipulation are required.

1. Get Multi

get_multi(
    db: AsyncSession,
    offset: int = 0,
    limit: Optional[int] = 100,
    schema_to_select: Optional[type[BaseModel]] = None,
    sort_columns: Optional[Union[str, list[str]]] = None,
    sort_orders: Optional[Union[str, list[str]]] = None,
    return_as_model: bool = False,
    return_total_count: bool = True,
    **kwargs: Any,
) -> dict[str, Any]

Purpose: To fetch multiple records based on specified filters, with options for sorting and pagination.
Usage Example: Gets the first 10 items sorted by name in ascending order.

items = await item_crud.get_multi(
    db,
    offset=0,
    limit=10,
    sort_columns=['name'],
    sort_orders=['asc'],
)

2. Get Joined

get_joined(
    db: AsyncSession,
    schema_to_select: Optional[type[BaseModel]] = None,
    join_model: Optional[ModelType] = None,
    join_on: Optional[Union[Join, BinaryExpression]] = None,
    join_prefix: Optional[str] = None,
    join_schema_to_select: Optional[type[BaseModel]] = None,
    join_type: str = "left",
    alias: Optional[AliasedClass] = None,
    join_filters: Optional[dict] = None,
    joins_config: Optional[list[JoinConfig]] = None,
    nest_joins: bool = False,
    relationship_type: Optional[str] = None,
    **kwargs: Any,
) -> Optional[dict[str, Any]]

Purpose: To fetch a single record with one or multiple joins on other models.
Usage Example: Fetches order details for a specific order by joining with the Customer table, selecting specific columns as defined in ReadOrderSchema and ReadCustomerSchema.

order_details = await order_crud.get_joined(
    db,
    schema_to_select=ReadOrderSchema,
    join_model=Customer,
    join_schema_to_select=ReadCustomerSchema,
    id=order_id,
)

3. Get Multi Joined

get_multi_joined(
    db: AsyncSession,
    schema_to_select: Optional[type[BaseModel]] = None,
    join_model: Optional[type[ModelType]] = None,
    join_on: Optional[Any] = None,
    join_prefix: Optional[str] = None,
    join_schema_to_select: Optional[type[BaseModel]] = None,
    join_type: str = "left",
    alias: Optional[AliasedClass[Any]] = None,
    join_filters: Optional[dict] = None,
    nest_joins: bool = False,
    offset: int = 0,
    limit: Optional[int] = 100,
    sort_columns: Optional[Union[str, list[str]]] = None,
    sort_orders: Optional[Union[str, list[str]]] = None,
    return_as_model: bool = False,
    joins_config: Optional[list[JoinConfig]] = None,
    return_total_count: bool = True,
    relationship_type: Optional[str] = None,
    **kwargs: Any,
) -> dict[str, Any]

Purpose: Similar to get_joined, but for fetching multiple records.
Usage Example: Retrieves a paginated list of orders (up to 5), joined with the Customer table, using specified schemas for selective column retrieval from both tables.

orders = await order_crud.get_multi_joined(
    db,
    schema_to_select=ReadOrderSchema,
    join_model=Customer,
    join_schema_to_select=ReadCustomerSchema,
    offset=0,
    limit=5,
)

4. Get Multi By Cursor

get_multi_by_cursor(
    db: AsyncSession,
    cursor: Any = None,
    limit: int = 100,
    schema_to_select: Optional[type[BaseModel]] = None,
    sort_column: str = "id",
    sort_order: str = "asc",
    **kwargs: Any,
) -> dict[str, Any]

Purpose: Implements cursor-based pagination for efficient data retrieval in large datasets.
Usage Example: Fetches the next 10 items after the last cursor for efficient pagination, sorted by creation date in descending order.

paginated_items = await item_crud.get_multi_by_cursor(
    db,
    cursor=last_cursor,
    limit=10,
    sort_column='created_at',
    sort_order='desc',
)

5. Select

async def select(
    db: AsyncSession,
    schema_to_select: Optional[type[BaseModel]] = None,
    sort_columns: Optional[Union[str, list[str]]] = None,
    sort_orders: Optional[Union[str, list[str]]] = None,
    **kwargs: Any,
) -> Select

Purpose: Constructs a SQL Alchemy Select statement with optional column selection, filtering, and sorting. Usage Example: Selects all items, filtering by name and sorting by id. Returns the Select statement.

stmt = await item_crud.select(
    schema_to_select=ItemSchema,
    sort_columns='id',
    name='John',
)
# Note: This method returns a SQL Alchemy Select object, not the actual query result.

6. Count for Joined Models

count(
    db: AsyncSession,
    joins_config: Optional[list[JoinConfig]] = None,
    **kwargs: Any,
) -> int

Purpose: To count records that match specified filters, especially useful in scenarios involving joins between models. This method supports counting unique entities across relationships, a common requirement in many-to-many or complex relationships.
Usage Example: Count the number of unique projects a participant is involved in, considering a many-to-many relationship between Project and Participant models.

# Assuming a Project model related to a Participant model through a many-to-many relationship
projects_count = await project_crud.count(
    db=session,
    joins_config=[
        JoinConfig(
            model=Participant,
            join_on=ProjectsParticipantsAssociation.project_id == Project.id,
            join_type="inner",
        ),
    ],
    participant_id=specific_participant_id,
)

Error Handling

FastCRUD provides mechanisms to handle common database errors, ensuring robust API behavior.