Designed for use with FastAPI to build a router for CRUD operations on a SQLAlchemy model. It automates the creation of API endpoints for these operations, making it easier to set up a RESTful API.
Endpoints are created for the following operations:
- Read all items
- Read one item by primary key
- Create one item
- Update one item by primary key
- Delete all items
- Delete one item by primary key
Endpoints created by CRUDBuilder are designed to support the OpenAPI documentation that FastAPI automatically generates.
Optionally, you can add security, caching and custom postprocessors to the generated endpoints. They're designed to attach to
an existing APIRouter object in FastAPI, or alternatively can be used to create a new router.
Documentation: (This file) https://github.com/TetricusLabs/crudbuilder
Source Code: https://github.com/TetricusLabs/crudbuilder
- Python 3.10+
- FastAPI
- SQLAlchemy
- [optional] Memcached (pymemcached client)
- Automatically generate CRUD endpoints for a given SQLAlchemy model
- Automatically generate OpenAPI documentation for the generated endpoints
- Optionally add security to the generated endpoints
- Optionally add caching to the generated endpoints
- Optionally add postprocessors to the generated endpoints
- Optionally infer the create and update models from the SQLAlchemy model, or provide custom Pydantic models
- Designed to be used with FastAPI
- Extendable to support other ORMs and caches
poetry add fastapi_crudbuilderfrom fastapi import APIRouter, Security
from fastapi_crudbuilder import CRUDBuilder
from src.postprocessors import YOUR_POSTPROCESSOR_1, YOUR_POSTPROCESSOR_2
from src.database import YOUR_DB_SESSION
from src.database.models import YOUR_MODEL
from src.security import YOUR_SECURITY_FUNCTION
example = APIRouter(prefix="/example", tags=["Example CRUD"]) # set up a FastAPI router to attach the CRUD endpoints to
@example.get("custom_non_crudbldr_route")
def custom_route():
return {"message": "Hello World"}
example = CRUDBuilder(
db_model=YOUR_MODEL,
db_func=YOUR_DB_SESSION,
infer_create=True, # Optionally infer the create model
infer_update=True, # Optionally infer the update model
read_security=Security(YOUR_SECURITY_FUNCTION.verify, scopes=["YOUR_MODEL:all:read"]),
# Optionally add custom security function and scope to the endpoint
create_security=Security(YOUR_SECURITY_FUNCTION.verify, scopes=["YOUR_MODEL:all:create"]),
update_security=Security(YOUR_SECURITY_FUNCTION.verify, scopes=["YOUR_MODEL:all:update"]),
delete_security=Security(YOUR_SECURITY_FUNCTION.verify, scopes=["YOUR_MODEL:all:delete"]),
response_postprocessors=[YOUR_POSTPROCESSOR_1(YOUR_MODEL), YOUR_POSTPROCESSOR_2(YOUR_MODEL)],
).build(example) # Attach the CRUD endpoints to the routerThe router must then be added to the FastAPI app (like any other router) in order to be used.
from fastapi import FastAPI
from src.routes.example import example
app = FastAPI()
app.include_router(example)- db_model: The SqlAlchemy model you want to create CRUD endpoints for
- db_func: The function that returns the SqlAlchemy session you'd like to use
The db_func parameter is expected to be a function that returns a SQLAlchemy session. This function will be injected using FastAPI's Depends function.
The generated endpoint will handle adding and committing the session, as well as rolling back the session in the event of an error.
This follows the FastAPI docs on dependency injection. See: https://fastapi.tiangolo.com/tutorial/dependencies/ and https://fastapi.tiangolo.com/reference/dependencies/
See https://fastapi.tiangolo.com/tutorial/sql-databases/ for more information on how to set up a database connection in FastAPI.
Here is an example of a db_func function that returns a SqlAlchemy session:
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
engine = create_engine("sqlite:///./test.db") # Example of a SqlAlchemy engine, replace with your own
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close() This will infer the model for the create and update endpoints from the SqlAlchemy model. Otherwise, you can pass in a custom Pydantic model for the data to be validated against. This will also be reflected in the automatically generated OpenAPI documentation.
You can pass in a FastAPI Security object to the CRUDBuilder class for each operation type (read, create, update, delete).
Post Processors are passed into the CRUDBuilder class as a list of functions that take the model as an argument and return a function that takes the response as an argument and returns a modified version of the response.
These are processed in order (e.g. in the example above YOUR_POSTPROCESSOR_1 would be applied to
the database result first, then YOUR_POSTPROCESSOR_2 etc.)
Post Processors are expected to receive a passed-in data model and return a callable that takes the response data and modifies it.
This allows you to adjust the response of the CRUD endpoints -- for example if you store dates in your Database as UTC but want to render them to users in their local time or a different format. See example below:
def transform_response_date(db_model: BaseModel) -> Callable:
def transform_dates(data: db_model.__class__, db_model: DeclarativeMeta = db_model):
columns = inspect(db_model).columns.items()
for name, column in columns:
if column.type.python_type.__name__ in (
"datetime",
"date",
"datetime.datetime",
):
_LOGGER.info(f"Transforming {name} to ISO 8601 date format.")
if getattr(data, name):
setattr(
data, name, convert_iso_datetime_format(getattr(data, name))
)
return data
return transform_datesYou can optionally pass a cache object to the CRUDBuilder class. Right now only Memcached is supported, and only the pymemcached client has been tested. See https://pymemcache.readthedocs.io/en/latest/getting_started.html for more information on how to set up a pymemcached client.
Caching relies upon the primary key of the model in question in the case of singe items, and takes into account query parameters like equals_field, equals_value, relationships and sorting. The primary key distinction is not useful in the case of read-all, but the query parameters are still taken into account.
Practically, this should result in a cache hit for the same item if the primary key is the same, or if the query parameters are the same, across any number of requests.
We chose to cache inside the endpoint, instead of caching the endpoint itself.
This allows for caching to be subject to the same access controls applied to the endpoint, and share that logic. It also means that if multiple CRUDBuilders are invoked on the same model, but with different postprocessors, they would share caching on the underlying data (because the generated cache key would be the same).
In theory any cache client library with the methods:
- get
- set
- delete
- delete_many
should work, but this has not been tested.
(Only SqlAlchemy and Memcached are supported for now)
- SqlAlchemy
- Memcached (pymemcached client)
GET /yourmodel
Get all items for your model
Query params:
limit: Number of items to return
skip: Number of items to skip
sort_field: Name of field to sort by, defaults to the primary key
sort_desc: True to sort descending, False for ascending
equals_field: Name of field to filter to a value, paired with equals_value
equals_value: Value of equals_field to filter by
relationships: Comma-separated names of fields that are relationships in the
SQLAlchemy model
GET /yourmodel/{item_id}
Get one item for your model by primary key (item_id)
Query params:
relationships: Comma-separated names of fields that are relationships in the
SQLAlchemy model
POST /yourmodel
Create one new item for your model
Request body:
JSON must match the given create_schema
PUT /yourmodel/{item_id}
Update fields for one item for your model by primary key (item_id)
Request body:
JSON must match the given update_schema
DELETE /yourmodel
Delete all contents in your model
DELETE /yourmodel/{item_id}
Delete one item for your model by primary key (item_id)
Notes:
* Ensure that the provided SQLAlchemy model and Pydantic schemas are compatible.
* The db_func should return a SQLAlchemy Session object. This will be injected
using FastAPI's Depends.
* The CRUD endpoints will reflect the attributes and relationships defined in
the SQLAlchemy model.
- Support for other cache clients like redis
- Async support
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
Please make sure to update tests as appropriate.
We use pytest for testing. To run the tests, simply run pytest in the root directory of the project. Tests are stored in the tests directory.