FastCRUD is a Python package for FastAPI, offering robust async CRUD operations and flexible endpoint creation utilities.
one_or_none parameter to FastCRUD get method (default False)nest_joins parameter to FastCRUD get_joined and get_multi_joined (default False)getBy default, the get method in FastCRUD returns the first object matching all the filters it finds.
If you want to ensure the one_or_none behavior, you may pass the parameter as True:
crud.get(
async_session,
one_or_none=True,
category_id=1
)
get_joined and get_multi_joinedBy default, FastCRUD joins all the data and returns it in a single dictionary.
Let's define two tables:
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
name = Column(String)
tier_id = Column(Integer, ForeignKey("tier.id"))
class Tier(Base):
__tablename__ = "tier"
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
And join them with FastCRUD:
user_tier = await user_crud.get_joined(
db=db,
model=Tier,
join_on=User.tier_id == Tier.id,
join_type="left",
join_prefix="tier_",,
id=1
)
We'll get:
{
"id": 1,
"name": "Example",
"tier_id": 1,
"tier_name": "Free",
}
Now, if you want the joined data in a nested dictionary instead, you may just pass nest_joins=True:
user_tier = await user_crud.get_joined(
db=db,
model=Tier,
join_on=User.tier_id == Tier.id,
join_type="left",
join_prefix="tier_",
nest_joins=True,
id=1,
)
And you will get:
{
"id": 1,
"name": "Example",
"tier": {
"id": 1,
"name": "Free",
},
}
This works for both get_joined and get_multi_joined.
[!WARNING] Note that the final
"_"in the passed"tier_"is stripped.
get method by @dubusster in https://github.com/igorbenav/fastcrud/pull/54
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.11.0...v0.11.1
get_multi and get_multi_joined methodsreturn_as_model is True fixeddb_row to methods fixedvalid_methods bug fixed (raising wrong error type)FastAPI raised up to 0.111.0
Now, for a model such as :
class MultiPkModel(Base):
__tablename__ = "multi_pk"
id = Column(Integer, primary_key=True)
uuid = Column(String(32), primary_key=True)
name = Column(String, unique=True)
the endpoint creator should give a path like /multi_pk/get/{id}/{uuid} for the different endpoints (get, update and delete) that need primary keys to interact with the database. The order of path parameter is given by the column order.
Primary keys named other than id are now supported as well.
Now, in get_multi and get_multi_joined you may pass return_total_count=False to get the data only, and not the count, like this:
crud.get_multi(
db=db,
return_total_count=False,
)
Returning
{"data": [...]}
Standard behavior is still return_total_count=True, with:
crud.get_multi(
db=db
)
Returning
{"data": [...], "total_count": 999}
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.10.0...v0.11.0
select statement by @dubusster in #28 🚀count method (passing joins_config)filters parameter in JoinConfig)mypy) plus fixes for typingruff) 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
) -> Selectable
This method constructs a SQL Alchemy Select statement, offering optional column selection, filtering, and sorting. It's designed for flexibility, allowing you to chain additional SQLAlchemy methods for even more complex queries.
stmt = await my_model_crud.select(schema_to_select=MySchema, sort_columns='name', name__like='%example%')
stmt = stmt.where(additional_conditions).limit(10)
results = await db.execute(stmt)
JoinConfig is a detailed configuration mechanism for specifying joins between models in FastCRUD queries. It contains the following key attributes:
model: The SQLAlchemy model to join.join_on: The condition defining how the join connects to other models.join_prefix: An optional prefix for the joined columns to avoid column name conflicts.schema_to_select: An optional Pydantic schema for selecting specific columns from the joined model.join_type: The type of join (e.g., "left", "inner").alias: An optional SQLAlchemy AliasedClass for complex scenarios like self-referential joins or multiple joins on the same model.filters: An optional dictionary to apply filters directly to the joined model.count Method with JoinsThe count method can be enhanced with join operations to perform complex aggregate queries. While count primarily returns the number of records matching a given condition, introducing joins allows for counting records across related models based on specific relationships and conditions.
JoinConfigFor join requirements, the count method can be invoked with join parameters passed as a list of JoinConfig to the joins_config parameter:
from fastcrud import JoinConfig
# Count the number of tasks assigned to users in a specific department
task_count = await task_crud.count(
db=db,
joins_config=[
JoinConfig(
model=User,
join_on=Task.assigned_user_id == User.id
),
JoinConfig(
model=Department,
join_on=User.department_id == Department.id,
filters={"name": "Engineering"}
)
]
)
get_joined and get_multi_joinedThese methods are essential for retrieving records from a primary model while including related data from one or more joined models. They support both simple and complex joining scenarios, including self-referential joins and many-to-many relationships.
For simpler join requirements, FastCRUD allows specifying join parameters directly:
model: The target model to join.join_on: The join condition.join_type: Specifies the SQL join type.aliased: When True, uses an alias for the model in the join.join_prefix: Optional prefix for columns from the joined model.filters: Additional filters for the joined model.# Fetch tasks with user details, specifying a left join
tasks_with_users = await task_crud.get_joined(
db=db,
model=User,
join_on=Task.user_id == User.id,
join_type="left"
)
JoinConfigWhen dealing with more complex join conditions, such as multiple joins, self-referential joins, or needing to specify aliases and filters, JoinConfig instances become the norm. They offer granular control over each join's aspects, enabling precise and efficient data retrieval.
# Fetch users with details from related departments and roles, using aliases for self-referential joins
users = await user_crud.get_multi_joined(
db=db,
schema_to_select=UserSchema,
joins_config=[
JoinConfig(
model=Department,
join_on=User.department_id == Department.id,
join_prefix="dept_"
),
JoinConfig(
model=Role,
join_on=User.role_id == Role.id,
join_prefix="role_"
),
JoinConfig(
model=User,
alias=manager_alias,
join_on=User.manager_id == manager_alias.id,
join_prefix="manager_"
)
]
)
get_multi_joinedFastCRUD simplifies dealing with many-to-many relationships by allowing easy fetch operations with joined models. Here, we demonstrate using get_multi_joined to handle a many-to-many relationship between Project and Participant models, linked through an association table.
Note on Handling Many-to-Many Relationships:
When using get_multi_joined for many-to-many relationships, it's essential to maintain a specific order in your joins_config:
This order ensures that the SQL joins are structured correctly to reflect the many-to-many relationship and retrieve the desired data accurately.
!!! TIP
Note that the first one can be the model defined in `FastCRUD(Model)`.
Imagine a scenario where projects have multiple participants, and participants can be involved in multiple projects. This many-to-many relationship is facilitated through an association table.
Our models include Project, Participant, and an association model ProjectsParticipantsAssociation:
from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
# Association table for the many-to-many relationship
projects_participants_association = Table('projects_participants_association', Base.metadata,
Column('project_id', Integer, ForeignKey('projects.id'), primary_key=True),
Column('participant_id', Integer, ForeignKey('participants.id'), primary_key=True)
)
class Project(Base):
__tablename__ = 'projects'
id = Column(Integer, primary_key=True)
name = Column(String)
description = Column(String)
# Relationship to Participant through the association table
participants = relationship("Participant", secondary=projects_participants_association)
class Participant(Base):
__tablename__ = 'participants'
id = Column(Integer, primary_key=True)
name = Column(String)
role = Column(String)
# Relationship to Project through the association table
projects = relationship("Project", secondary=projects_participants_association)
get_multi_joinedTo fetch projects along with their participants, we utilize get_multi_joined with appropriate JoinConfig settings:
from fastcrud import FastCRUD, JoinConfig
# Initialize FastCRUD for the Project model
crud_project = FastCRUD(Project)
# Define join conditions and configuration
joins_config = [
JoinConfig(
model=ProjectsParticipantsAssociation,
join_on=Project.id == ProjectsParticipantsAssociation.project_id,
join_type="inner",
join_prefix="pp_"
),
JoinConfig(
model=Participant,
join_on=ProjectsParticipantsAssociation.participant_id == Participant.id,
join_type="inner",
join_prefix="participant_"
)
]
# Fetch projects with their participants
projects_with_participants = await crud_project.get_multi_joined(
db_session,
joins_config=joins_config
)
# Now, `projects_with_participants['data']` will contain projects along with their participant information.
join_prefix attribute to avoid column name collisions, especially in complex joins involving multiple models or self-referential joins.alias attribute for disambiguating joins on the same model or for self-referential joins.filters attribute in JoinConfig to refine the data set returned by the query.JoinConfig entries to ensure logical and efficient SQL join construction.Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.9.1...v0.10.0
In complex query scenarios, particularly when you need to join a table to itself or perform multiple joins on the same table for different purposes, aliasing becomes crucial. Aliasing allows you to refer to the same table in different contexts with unique identifiers, avoiding conflicts and ambiguity in your queries.
For both get_joined and get_multi_joined methods, when you need to join the same model multiple times, you can utilize the alias parameter within your JoinConfig to differentiate between the joins. This parameter expects an instance of AliasedClass, which can be created using the aliased function from SQLAlchemy (also in fastcrud for convenience).
Consider a task management application where tasks have both an owner and an assigned user, represented by the same UserModel. To fetch tasks with details of both users, we use aliases to join the UserModel twice, distinguishing between owners and assigned users.
Let's start by creating the aliases and passing them to the join configuration. Don't forget to use the alias for join_on:
from fastcrud import FastCRUD, JoinConfig, aliased
# Create aliases for UserModel to distinguish between the owner and the assigned user
owner_alias = aliased(UserModel, name="owner")
assigned_user_alias = aliased(UserModel, name="assigned_user")
# Configure joins with aliases
joins_config = [
JoinConfig(
model=UserModel,
join_on=Task.owner_id == owner_alias.id,
join_prefix="owner_",
schema_to_select=UserSchema,
join_type="inner",
alias=owner_alias # Pass the aliased class instance
),
JoinConfig(
model=UserModel,
join_on=Task.assigned_user_id == assigned_user_alias.id,
join_prefix="assigned_",
schema_to_select=UserSchema,
join_type="inner",
alias=assigned_user_alias # Pass the aliased class instance
)
]
# Initialize your FastCRUD instance for TaskModel
task_crud = FastCRUD(TaskModel)
# Fetch tasks with joined user details
tasks = await task_crud.get_multi_joined(
db=session,
schema_to_select=TaskSchema,
joins_config=joins_config,
offset=0,
limit=10
)
Then just pass this joins_config to get_multi_joined:
from fastcrud import FastCRUD, JoinConfig, aliased
...
# Configure joins with aliases
joins_config = [
...
]
# Initialize your FastCRUD instance for TaskModel
task_crud = FastCRUD(TaskModel)
# Fetch tasks with joined user details
tasks = await task_crud.get_multi_joined(
db=session,
schema_to_select=TaskSchema,
joins_config=joins_config,
offset=0,
limit=10
)
In this example, owner_alias and assigned_user_alias are created from UserModel to distinguish between the task's owner and the assigned user within the task management system. By using aliases, you can join the same model multiple times for different purposes in your queries, enhancing expressiveness and eliminating ambiguity.
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.9.0...v0.9.1
To facilitate complex data relationships, get_joined and get_multi_joined can be configured to handle joins with multiple models. This is achieved using the joins_config parameter, where you can specify a list of JoinConfig instances, each representing a distinct join configuration.
Consider a scenario where you want to retrieve users along with their associated tier and department information. Here's how you can achieve this using get_multi_joined.
Start by creating a list of the multiple models to be joined:
from fastcrud import JoinConfig
joins_config = [
JoinConfig(
model=Tier,
join_on=User.tier_id == Tier.id,
join_prefix="tier_",
schema_to_select=TierSchema,
join_type="left",
),
JoinConfig(
model=Department,
join_on=User.department_id == Department.id,
join_prefix="dept_",
schema_to_select=DepartmentSchema,
join_type="inner",
)
]
users = await user_crud.get_multi_joined(
db=session,
schema_to_select=UserSchema,
joins_config=joins_config,
offset=0,
limit=10,
sort_columns='username',
sort_orders='asc'
)
Then just pass this list to joins_config:
from fastcrud import JoinConfig
joins_config = [
...
]
users = await user_crud.get_multi_joined(
db=session,
schema_to_select=UserSchema,
joins_config=joins_config,
offset=0,
limit=10,
sort_columns='username',
sort_orders='asc'
)
In this example, users are joined with the Tier and Department models. The join_on parameter specifies the condition for the join, join_prefix assigns a prefix to columns from the joined models (to avoid naming conflicts), and join_type determines whether it's a left or inner join.
[!WARNING] If both single join parameters and
joins_configare used simultaneously, an error will be raised.
get_joined and get_multi_joined for multiple models by @igorbenav in https://github.com/igorbenav/fastcrud/pull/26
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.8.0...v0.9.0
endpoint_names parameter to customize auto generated endpoints (both to crud_router and EndpointCreator)You can customize the names of the auto generated endpoints by passing an endpoint_names dictionary when initializing the EndpointCreator or calling the crud_router function. This dictionary should map the CRUD operation names (create, read, update, delete, db_delete, read_multi, read_paginated) to your desired endpoint names.
crud_routerHere's how you can customize endpoint names using the crud_router function:
from fastapi import FastAPI
from yourapp.crud import crud_router
from yourapp.models import YourModel
from yourapp.schemas import CreateYourModelSchema, UpdateYourModelSchema
from yourapp.database import async_session
app = FastAPI()
# Custom endpoint names
custom_endpoint_names = {
"create": "add",
"read": "fetch",
"update": "modify",
"delete": "remove",
"read_multi": "list",
"read_paginated": "paginate"
}
# Setup CRUD router with custom endpoint names
app.include_router(crud_router(
session=async_session,
model=YourModel,
create_schema=CreateYourModelSchema,
update_schema=UpdateYourModelSchema,
path="/yourmodel",
tags=["YourModel"],
endpoint_names=custom_endpoint_names
))
In this example, the standard CRUD endpoints will be replaced with /add, /fetch/{id}, /modify/{id}, /remove/{id}, /list, and /paginate.
EndpointCreatorIf you are using EndpointCreator, you can also pass the endpoint_names dictionary to customize the endpoint names similarly:
# Custom endpoint names
custom_endpoint_names = {
"create": "add_new",
"read": "get_single",
"update": "change",
"delete": "erase",
"db_delete": "hard_erase",
"read_multi": "get_all",
"read_paginated": "get_page"
}
# Initialize and use the custom EndpointCreator
endpoint_creator = EndpointCreator(
session=async_session,
model=YourModel,
create_schema=CreateYourModelSchema,
update_schema=UpdateYourModelSchema,
path="/yourmodel",
tags=["YourModel"],
endpoint_names=custom_endpoint_names
)
endpoint_creator.add_routes_to_router()
app.include_router(endpoint_creator.router)
!!! TIP
You only need to pass the names of the endpoints you want to change in the endpoint_names dict.
paginated to handle utility functionsget_paginated endpointEndpoint: /get_paginated
Method: GET
Description: Retrieves multiple items with pagination.
Query Parameters:
page: The page number, starting from 1.itemsPerPage: The number of items per page.Example Request: GET /yourmodel/get_paginated?page=1&itemsPerPage=3.
Example Return:
{
"data": [
{"id": 1, "name": "Item 1", "description": "Description of item 1"},
{"id": 2, "name": "Item 2", "description": "Description of item 2"},
{"id": 3, "name": "Item 3", "description": "Description of item 3"}
],
"total_count": 50,
"has_more": true,
"page": 1,
"items_per_page": 10
}
paginated modulepaginated_response(
crud_data: dict, page: int, items_per_page: int
) -> dict[str, Any]
Usage and Return example:
crud = FastCRUD(MyModel)
paginated_response(crud_data=crud.get_multi(MyModel), page=1, items_per_page=3)
returns
{
"data": [
{"id": 1, "name": "Item 1", "description": "Description of item 1"},
{"id": 2, "name": "Item 2", "description": "Description of item 2"},
{"id": 3, "name": "Item 3", "description": "Description of item 3"}
],
"total_count": 50,
"has_more": true,
"page": 1,
"items_per_page": 10
}
compute_offset(page: int, items_per_page: int) -> int
Usage and Return example:
offset(3, 10)
returns
20
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.6.0...v0.7.0
updated_at column namecrud_router and EndpointCreator now optionalhttp_exceptions now also in exceptions moduleupdated_at columnYou can also customize your `updated_at` column:
```python hl_lines="9"
app.include_router(endpoint_creator(
session=async_session,
model=MyModel,
create_schema=CreateMyModelSchema,
update_schema=UpdateMyModelSchema,
delete_schema=DeleteMyModelSchema,
path="/mymodel",
tags=["MyModel"],
updated_at_column='date_updated'
))
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.5.0...v0.6.0
FastCRUD supports advanced filtering options, allowing you to query records using operators such as greater than (__gt), less than (__lt), and their inclusive counterparts (__gte, __lte). These filters can be used in any method that retrieves or operates on records, including get, get_multi, exists, count, update, and delete.
The following examples demonstrate how to use advanced filters for querying and manipulating data:
# Fetch items priced between $5 and $20
items = await item_crud.get_multi(
db=db,
price__gte=5,
price__lte=20
)
# Count items added in the last month
item_count = await item_crud.count(
db=db,
added_at__gte=datetime.datetime.now() - datetime.timedelta(days=30)
)
EndpointCreator and crud_router with Custom Soft Delete ColumnsWhen initializing crud_router or creating a custom EndpointCreator, you can pass the names of your custom soft delete columns through the FastCRUD initialization. This informs FastCRUD which columns to check and update for soft deletion operations.
Here's an example of using crud_router with custom soft delete columns:
from fastapi import FastAPI
from fastcrud import FastCRUD, crud_router
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()
# Assuming async_session is your AsyncSession generator
# and MyModel is your SQLAlchemy model
# Initialize FastCRUD with custom soft delete columns
my_model_crud = FastCRUD(MyModel,
is_deleted_column='archived', # Custom 'is_deleted' column name
deleted_at_column='archived_at' # Custom 'deleted_at' column name
)
# Setup CRUD router with the FastCRUD instance
app.include_router(crud_router(
session=async_session,
model=MyModel,
crud=my_model_crud,
create_schema=CreateMyModelSchema,
update_schema=UpdateMyModelSchema,
delete_schema=DeleteMyModelSchema,
path="/mymodel",
tags=["MyModel"]
))
This setup ensures that the soft delete functionality within your application utilizes the archived and archived_at columns for marking records as deleted, rather than the default is_deleted and deleted_at fields.
To update multiple records, you can set the allow_multiple=True parameter in the update method. This allows FastCRUD to apply the update to all records matching the given filters.
# Assuming setup for FastCRUD instance `item_crud` and SQLAlchemy async session `db`
# Update all items priced below $10 to a new price
await item_crud.update(
db=db,
object={"price": 9.99},
allow_multiple=True,
price__lt=10
)
Similarly, you can delete multiple records by using the allow_multiple=True parameter in the delete or db_delete method, depending on whether you're performing a soft or hard delete.
# Soft delete all items not sold in the last year
await item_crud.delete(
db=db,
allow_multiple=True,
last_sold__lt=datetime.datetime.now() - datetime.timedelta(days=365)
)
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.4.0...v0.5.0
Since SQLModel is just a combination of SQLAlchemy and Pydantic, the process simplifies as SQLModel combines the model and schema definitions.
Wherever in the docs you see a SQLAlchemy model or Pydantic schema being used, you may just replace it with SQLModel and it will work. For the quick start:
Define your SQLModel model
from sqlmodel import SQLModel, Field
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
class Item(SQLModel):
__tablename__ = 'items'
id: int = Field(primary_key=True)
name: str
class ItemCreateSchema(SQLModel):
name: str
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Then your schemas
from sqlmodel import SQLModel, Field
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
class Item(SQLModel):
__tablename__ = 'items'
id: int = Field(primary_key=True)
name: str
class ItemCreateSchema(SQLModel):
name: str
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
And, finally, your database connection
from sqlmodel import SQLModel, Field
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
class Item(SQLModel):
__tablename__ = 'items'
id: int = Field(primary_key=True)
name: str
class ItemCreateSchema(SQLModel):
name: str
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Use crud_router and include it in your FastAPI application
from fastcrud import FastCRUD, crud_router
app = FastAPI()
item_router = crud_router(
session=session,
model=Item,
crud=FastCRUD(Item),
create_schema=ItemSchema,
update_schema=ItemSchema,
path="/items",
tags=["Items"]
)
app.include_router(item_router)
By following the above setup, FastCRUD auto-generates CRUD endpoints for your model, accessible through the /docs route of your FastAPI application.
Full Changelog: https://github.com/igorbenav/fastcrud/compare/v0.3.0...v0.4.0