EndpointCreator API Reference

EndpointCreator is a class designed to create and register CRUD endpoints for a FastAPI application. It simplifies the process of adding CRUD (Create, Read, Update, Delete) endpoints to a FastAPI router.

Class Definition

A class to create and register CRUD endpoints for a FastAPI application.

This class simplifies the process of adding create, read, update, and delete (CRUD) endpoints to a FastAPI router. It is initialized with a SQLAlchemy session, model, CRUD operations, and Pydantic schemas, and allows for custom dependency injection for each endpoint. The method assumes id is the primary key for path parameters.

Attributes:

Name	Type	Description
session		The SQLAlchemy async session.
model		The SQLAlchemy model.
create_schema		Pydantic schema for creating an item.
update_schema		Pydantic schema for updating an item.
crud		An optional FastCRUD instance. If not provided, uses FastCRUD(model).
include_in_schema		Whether to include the created endpoints in the OpenAPI schema.
delete_schema		Optional Pydantic schema for deleting an item.
path		Base path for the CRUD endpoints.
tags		List of tags for grouping endpoints in the documentation.
is_deleted_column		Optional column name to use for indicating a soft delete. Defaults to "is_deleted".
deleted_at_column		Optional column name to use for storing the timestamp of a soft delete. Defaults to "deleted_at".
updated_at_column		Optional column name to use for storing the timestamp of an update. Defaults to "updated_at".
endpoint_names		Optional dictionary to customize endpoint names for CRUD operations. Keys are operation types ("create", "read", "update", "delete", "db_delete", "read_multi"), and values are the custom names to use. Unspecified operations will use default names.
filter_config	FilterConfig | None	Optional FilterConfig instance or dictionary to configure filters for the read_multi endpoint.
select_schema		Optional Pydantic schema for selecting an item.
custom_filters		Optional dictionary of custom filter operators. Keys are operator names (e.g., 'year'), values are callables that take a column and return a filter function.
include_relationships		If True, automatically detect and include related data from SQLAlchemy relationships in read endpoints. Can also be a list of relationship names to include specific relationships. Defaults to False.
nest_joins		If True, nested data structures will be returned where joined model data are nested as dictionaries or lists. Defaults to True.
include_one_to_many		If True, include one-to-many relationships when auto-detecting. Defaults to False for safety since one-to-many JOINs can return unbounded data. This is ignored when specific relationship names are provided.

Raises:

Type	Description
ValueError	If both included_methods and deleted_methods are provided.

Examples:

Basic Setup:

mymodel/model.py

from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class MyModel(Base):
    __tablename__ = "my_model"
    id = Column(Integer, primary_key=True)
    name = Column(String)

mymodel/schemas.py

import datetime

from pydantic import BaseModel


class CreateMyModelSchema(BaseModel):
    name: str | None = None


class UpdateMyModelSchema(BaseModel):
    name: str | None = None

from fastapi import FastAPI
from fastcrud import EndpointCreator

from .database import async_session
from .mymodel.model import MyModel
from .mymodel.schemas import CreateMyModelSchema, UpdateMyModelSchema

app = FastAPI()
endpoint_creator = EndpointCreator(
    session=async_session,
    model=MyModel,
    create_schema=CreateMyModelSchema,
    update_schema=UpdateMyModelSchema,
)
endpoint_creator.add_routes_to_router()
app.include_router(endpoint_creator.router, prefix="/mymodel")

With Custom Dependencies:

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    return ...

endpoint_creator.add_routes_to_router(
    read_deps=[get_current_user],
    update_deps=[get_current_user],
)

Selective Endpoint Creation (inclusion):

# Only create 'create' and 'read' endpoints
endpoint_creator.add_routes_to_router(
    included_methods=["create", "read"],
)

Selective Endpoint Creation (deletion):

# Create all but 'update' and 'delete' endpoints
endpoint_creator.add_routes_to_router(
    deleted_methods=["update", "delete"],
)

Integrating with Multiple Models:

# Assuming definitions for OtherModel, OtherModelCRUD, etc.

other_model_crud = OtherModelCRUD(OtherModel)
other_endpoint_creator = EndpointCreator(
    session=async_session,
    model=OtherModel,
    create_schema=CreateOtherModelSchema,
    update_schema=UpdateOtherModelSchema,
    crud=other_model_crud,
)
other_endpoint_creator.add_routes_to_router()
app.include_router(other_endpoint_creator.router, prefix="/othermodel")

Customizing Endpoint Names:

endpoint_creator = EndpointCreator(
    session=async_session,
    model=MyModel,
    create_schema=CreateMyModelSchema,
    update_schema=UpdateMyModelSchema,
    path="/mymodel",
    tags=["MyModel"],
    endpoint_names={
        "create": "add",  # Custom endpoint name for creating items
        "read": "fetch",  # Custom endpoint name for reading a single item
        "update": "change",  # Custom endpoint name for updating items
        # The delete operation will use the default name "delete"
    },
)
endpoint_creator.add_routes_to_router()

Using filter_config with dict:

from fastapi import FastAPI
from fastcrud import EndpointCreator, FilterConfig

from .database import async_session
from .mymodel.model import MyModel
from .mymodel.schemas import CreateMyModelSchema, UpdateMyModelSchema

app = FastAPI()
endpoint_creator = EndpointCreator(
    session=async_session,
    model=MyModel,
    create_schema=CreateMyModelSchema,
    update_schema=UpdateMyModelSchema,
    filter_config=FilterConfig(filters={"id": None, "name": "default"}),
)
# Adds CRUD routes with filtering capabilities
endpoint_creator.add_routes_to_router()
# Include the internal router into the FastAPI app
app.include_router(endpoint_creator.router, prefix="/mymodel")

# Explanation:
# The FilterConfig specifies that 'id' should be a query parameter with no default value
# and 'name' should be a query parameter with a default value of 'default'.
# When fetching multiple items, you can filter by these parameters.
# Example GET request: /mymodel/get_multi?id=1&name=example

Using filter_config with keyword arguments:

from fastapi import FastAPI
from fastcrud import EndpointCreator, FilterConfig

from .database import async_session
from .mymodel.model import MyModel
from .mymodel.schemas import CreateMyModelSchema, UpdateMyModelSchema

app = FastAPI()
endpoint_creator = EndpointCreator(
    session=async_session,
    model=MyModel,
    create_schema=CreateMyModelSchema,
    update_schema=UpdateMyModelSchema,
    filter_config=FilterConfig(id=None, name="default"),
)
# Adds CRUD routes with filtering capabilities
endpoint_creator.add_routes_to_router()
# Include the internal router into the FastAPI app
app.include_router(endpoint_creator.router, prefix="/mymodel")

# Explanation:
# The FilterConfig specifies that 'id' should be a query parameter with no default value
# and 'name' should be a query parameter with a default value of 'default'.
# When fetching multiple items, you can filter by these parameters.
# Example GET request: /mymodel/get_multi?id=1&name=example

Source code in fastcrud/endpoint/endpoint_creator.py

class EndpointCreator:
    """
    A class to create and register CRUD endpoints for a FastAPI application.

    This class simplifies the process of adding create, read, update, and delete (CRUD) endpoints
    to a FastAPI router. It is initialized with a SQLAlchemy session, model, CRUD operations,
    and Pydantic schemas, and allows for custom dependency injection for each endpoint.
    The method assumes `id` is the primary key for path parameters.

    Attributes:
        session: The SQLAlchemy async session.
        model: The SQLAlchemy model.
        create_schema: Pydantic schema for creating an item.
        update_schema: Pydantic schema for updating an item.
        crud: An optional FastCRUD instance. If not provided, uses `FastCRUD(model)`.
        include_in_schema: Whether to include the created endpoints in the OpenAPI schema.
        delete_schema: Optional Pydantic schema for deleting an item.
        path: Base path for the CRUD endpoints.
        tags: List of tags for grouping endpoints in the documentation.
        is_deleted_column: Optional column name to use for indicating a soft delete. Defaults to `"is_deleted"`.
        deleted_at_column: Optional column name to use for storing the timestamp of a soft delete. Defaults to `"deleted_at"`.
        updated_at_column: Optional column name to use for storing the timestamp of an update. Defaults to `"updated_at"`.
        endpoint_names: Optional dictionary to customize endpoint names for CRUD operations. Keys are operation types
                        (`"create"`, `"read"`, `"update"`, `"delete"`, `"db_delete"`, `"read_multi"`), and
                        values are the custom names to use. Unspecified operations will use default names.
        filter_config: Optional `FilterConfig` instance or dictionary to configure filters for the `read_multi` endpoint.
        select_schema: Optional Pydantic schema for selecting an item.
        custom_filters: Optional dictionary of custom filter operators. Keys are operator names (e.g., 'year'),
                        values are callables that take a column and return a filter function.
        include_relationships: If `True`, automatically detect and include related data from SQLAlchemy relationships in read endpoints.
            Can also be a list of relationship names to include specific relationships. Defaults to `False`.
        nest_joins: If `True`, nested data structures will be returned where joined model data are nested as dictionaries or lists. Defaults to `True`.
        include_one_to_many: If `True`, include one-to-many relationships when auto-detecting. Defaults to `False` for safety since
            one-to-many JOINs can return unbounded data. This is ignored when specific relationship names are provided.

    Raises:
        ValueError: If both `included_methods` and `deleted_methods` are provided.

    Examples:
        Basic Setup:

        ??? example "`mymodel/model.py`"

            ```python
            --8<--
            fastcrud/examples/mymodel/model.py:imports
            fastcrud/examples/mymodel/model.py:model_simple
            --8<--
            ```

        ??? example "`mymodel/schemas.py`"

            ```python
            --8<--
            fastcrud/examples/mymodel/schemas.py:imports
            fastcrud/examples/mymodel/schemas.py:createschema
            fastcrud/examples/mymodel/schemas.py:updateschema
            --8<--
            ```

        ```python
        from fastapi import FastAPI
        from fastcrud import EndpointCreator

        from .database import async_session
        from .mymodel.model import MyModel
        from .mymodel.schemas import CreateMyModelSchema, UpdateMyModelSchema

        app = FastAPI()
        endpoint_creator = EndpointCreator(
            session=async_session,
            model=MyModel,
            create_schema=CreateMyModelSchema,
            update_schema=UpdateMyModelSchema,
        )
        endpoint_creator.add_routes_to_router()
        app.include_router(endpoint_creator.router, prefix="/mymodel")
        ```

        With Custom Dependencies:

        ```python
        from fastapi.security import OAuth2PasswordBearer

        oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

        def get_current_user(token: str = Depends(oauth2_scheme)):
            return ...

        endpoint_creator.add_routes_to_router(
            read_deps=[get_current_user],
            update_deps=[get_current_user],
        )
        ```

        Selective Endpoint Creation (inclusion):

        ```python
        # Only create 'create' and 'read' endpoints
        endpoint_creator.add_routes_to_router(
            included_methods=["create", "read"],
        )
        ```

        Selective Endpoint Creation (deletion):

        ```python
        # Create all but 'update' and 'delete' endpoints
        endpoint_creator.add_routes_to_router(
            deleted_methods=["update", "delete"],
        )
        ```

        Integrating with Multiple Models:

        ```python
        # Assuming definitions for OtherModel, OtherModelCRUD, etc.

        other_model_crud = OtherModelCRUD(OtherModel)
        other_endpoint_creator = EndpointCreator(
            session=async_session,
            model=OtherModel,
            create_schema=CreateOtherModelSchema,
            update_schema=UpdateOtherModelSchema,
            crud=other_model_crud,
        )
        other_endpoint_creator.add_routes_to_router()
        app.include_router(other_endpoint_creator.router, prefix="/othermodel")
        ```

        Customizing Endpoint Names:

        ```python
        endpoint_creator = EndpointCreator(
            session=async_session,
            model=MyModel,
            create_schema=CreateMyModelSchema,
            update_schema=UpdateMyModelSchema,
            path="/mymodel",
            tags=["MyModel"],
            endpoint_names={
                "create": "add",  # Custom endpoint name for creating items
                "read": "fetch",  # Custom endpoint name for reading a single item
                "update": "change",  # Custom endpoint name for updating items
                # The delete operation will use the default name "delete"
            },
        )
        endpoint_creator.add_routes_to_router()
        ```

        Using `filter_config` with `dict`:

        ```python
        from fastapi import FastAPI
        from fastcrud import EndpointCreator, FilterConfig

        from .database import async_session
        from .mymodel.model import MyModel
        from .mymodel.schemas import CreateMyModelSchema, UpdateMyModelSchema

        app = FastAPI()
        endpoint_creator = EndpointCreator(
            session=async_session,
            model=MyModel,
            create_schema=CreateMyModelSchema,
            update_schema=UpdateMyModelSchema,
            filter_config=FilterConfig(filters={"id": None, "name": "default"}),
        )
        # Adds CRUD routes with filtering capabilities
        endpoint_creator.add_routes_to_router()
        # Include the internal router into the FastAPI app
        app.include_router(endpoint_creator.router, prefix="/mymodel")

        # Explanation:
        # The FilterConfig specifies that 'id' should be a query parameter with no default value
        # and 'name' should be a query parameter with a default value of 'default'.
        # When fetching multiple items, you can filter by these parameters.
        # Example GET request: /mymodel/get_multi?id=1&name=example
        ```

        Using `filter_config` with keyword arguments:

        ```python
        from fastapi import FastAPI
        from fastcrud import EndpointCreator, FilterConfig

        from .database import async_session
        from .mymodel.model import MyModel
        from .mymodel.schemas import CreateMyModelSchema, UpdateMyModelSchema

        app = FastAPI()
        endpoint_creator = EndpointCreator(
            session=async_session,
            model=MyModel,
            create_schema=CreateMyModelSchema,
            update_schema=UpdateMyModelSchema,
            filter_config=FilterConfig(id=None, name="default"),
        )
        # Adds CRUD routes with filtering capabilities
        endpoint_creator.add_routes_to_router()
        # Include the internal router into the FastAPI app
        app.include_router(endpoint_creator.router, prefix="/mymodel")

        # Explanation:
        # The FilterConfig specifies that 'id' should be a query parameter with no default value
        # and 'name' should be a query parameter with a default value of 'default'.
        # When fetching multiple items, you can filter by these parameters.
        # Example GET request: /mymodel/get_multi?id=1&name=example
        ```
    """

    def __init__(
        self,
        session: Callable,
        model: ModelType,
        create_schema: type[CreateSchemaType],
        update_schema: type[UpdateSchemaType],
        crud: FastCRUD | None = None,
        include_in_schema: bool = True,
        delete_schema: type[DeleteSchemaType] | None = None,
        path: str = "",
        tags: list[str | Enum] | None = None,
        is_deleted_column: str = "is_deleted",
        deleted_at_column: str = "deleted_at",
        updated_at_column: str = "updated_at",
        endpoint_names: dict[str, str] | None = None,
        filter_config: FilterConfig | dict | None = None,
        select_schema: type[SelectSchemaType] | None = None,
        create_config: CreateConfig | None = None,
        update_config: UpdateConfig | None = None,
        delete_config: DeleteConfig | None = None,
        custom_filters: dict[str, FilterCallable] | None = None,
        include_relationships: bool | Sequence[str] = False,
        joins_config: Sequence[JoinConfig] | None = None,
        nest_joins: bool = True,
        default_nested_limit: int | None = None,
        include_one_to_many: bool = False,
    ) -> None:
        self._primary_keys = get_primary_key_columns(model)
        self._primary_keys_types = {
            pk.name: pk_type
            for pk in self._primary_keys
            if (pk_type := get_python_type(pk)) is not None
        }
        self.primary_key_names = [pk.name for pk in self._primary_keys]
        self.session = session
        self.custom_filters = custom_filters
        self.crud = crud or FastCRUD(
            model=model,
            is_deleted_column=is_deleted_column,
            deleted_at_column=deleted_at_column,
            updated_at_column=updated_at_column,
            custom_filters=custom_filters,
        )
        self.model = model
        self.create_schema = create_schema
        self.update_schema = update_schema
        self.delete_schema = delete_schema
        self.select_schema = select_schema
        self.include_in_schema = include_in_schema
        self.path = path
        self.tags = tags or []
        self.router = APIRouter()
        self.is_deleted_column = is_deleted_column
        self.deleted_at_column = deleted_at_column
        self.updated_at_column = updated_at_column
        self.default_endpoint_names = {
            "create": "",
            "read": "",
            "update": "",
            "delete": "",
            "db_delete": "db_delete",
            "read_multi": "",
        }
        self.endpoint_names = {**self.default_endpoint_names, **(endpoint_names or {})}
        if filter_config:
            if isinstance(filter_config, dict):
                filter_config = FilterConfig(**filter_config)
            self._validate_filter_config(filter_config)
            self.filter_config: FilterConfig | None = filter_config
        else:
            self.filter_config = None
        self.create_config = create_config
        self.update_config = update_config
        self.delete_config = delete_config
        self.column_types = dict(get_column_types(model))

        if include_relationships and joins_config:
            raise ValueError(
                "Cannot use both 'include_relationships' and 'joins_config'. "
                "Use 'include_relationships' for auto-detection or 'joins_config' for manual control."
            )

        if include_relationships and include_relationships is not True:
            relationship_names = list(include_relationships)
            available_relationships = discover_model_relationships(model)
            available_names = {name for name, _ in available_relationships}

            invalid_names = set(relationship_names) - available_names
            if invalid_names:
                raise ValueError(
                    f"Invalid relationship name(s): {sorted(invalid_names)}. "
                    f"Available relationships on '{model.__name__}': {sorted(available_names)}. "
                    f"Tip: Use `include_relationships=True` to include all relationships, "
                    f"or check your model's relationship definitions."
                )

        self.include_relationships = include_relationships
        self.joins_config: list[JoinConfig] | None = (
            list(joins_config) if joins_config else None
        )
        self.nest_joins = nest_joins
        self.default_nested_limit = default_nested_limit
        self.include_one_to_many = include_one_to_many

        if select_schema is not None:
            response_key = getattr(self.crud, "multi_response_key", "data")
            self.list_response_model = create_list_response(select_schema, response_key)
            self.paginated_response_model = create_paginated_response(
                select_schema, response_key
            )
        else:
            self.list_response_model = None  # type: ignore
            self.paginated_response_model = None  # type: ignore

    def _validate_filter_config(self, filter_config: FilterConfig) -> None:
        model_columns = self.crud.model_col_names
        supported_filters = {**SUPPORTED_FILTERS, **(self.custom_filters or {})}
        for key, value in filter_config.filters.items():
            if callable(value):
                continue

            if filter_config.is_joined_filter(key):
                try:
                    relationship_path, final_field, operator = (
                        filter_config.parse_joined_filter(key)
                    )

                    if (
                        operator and operator not in supported_filters
                    ):  # pragma: no cover
                        raise ValueError(
                            f"Invalid filter op '{operator}': following filter ops are allowed: {supported_filters.keys()}"
                        )

                    if not validate_joined_filter_path(
                        self.model, relationship_path, final_field
                    ):  # pragma: no cover
                        raise ValueError(
                            f"Invalid joined filter '{key}': relationship path {'.'.join(relationship_path + [final_field])} not found in model '{self.model.__name__}'"
                        )
                except ValueError as e:  # pragma: no cover
                    raise ValueError(f"Invalid joined filter '{key}': {str(e)}")
            else:
                if "__" in key:
                    field_name, op = key.rsplit("__", 1)
                    if op not in supported_filters:
                        raise ValueError(
                            f"Invalid filter op '{op}': following filter ops are allowed: {supported_filters.keys()}"
                        )
                else:
                    field_name = key

                if field_name not in model_columns:
                    raise ValueError(
                        f"Invalid filter column '{key}': not found in model '{self.model.__name__}' columns"
                    )

    def _should_include_relationships(self) -> bool:
        """Check if relationships should be included in responses."""
        return bool(self.include_relationships or self.joins_config)

    def _get_join_params(self) -> dict[str, Any]:
        """
        Get the join parameters for CRUD methods.

        Returns a dict with either:
        - `joins_config` if manually specified or built from auto-detection
        - `auto_detect_relationships` for backward compatibility (deprecated path)

        Note: When using auto-detection, one-to-many relationships are excluded
        by default unless `include_one_to_many=True` is set.
        """
        if self.joins_config:
            return {"joins_config": self.joins_config}
        else:
            config = resolve_relationship_config(
                self.model,
                self.include_relationships,
                default_nested_limit=self.default_nested_limit,
                include_one_to_many=self.include_one_to_many,
            )
            if config:
                return {"joins_config": config}
            return {}

    def _create_item(self) -> Callable[..., Awaitable[Any]]:
        """Creates an endpoint for creating items in the database.

        Returns:
            - When auto_fields are used: SQLAlchemy model instance (legacy behavior)
            - When select_schema is provided: dict by default, or Pydantic model if return_as_model=True (recommended - gets data back in one call)
            - When select_schema is None: None (v0.20.0 behavior)
        """
        auto_field_injector = create_auto_field_injector(self.create_config)

        request_schema: type[BaseModel] = self.create_schema
        if self.create_config and self.create_config.exclude_from_schema:
            request_schema = create_modified_schema(
                self.create_schema,
                tuple(self.create_config.exclude_from_schema),
                f"{self.create_schema.__name__}Modified",
            )

        async def endpoint(
            db: AsyncSession = Depends(self.session),
            item: BaseModel = Body(...),
            auto_fields: dict = Depends(auto_field_injector),
        ):
            unique_columns = get_unique_columns(self.model)

            for column in unique_columns:
                col_name = column.name
                if hasattr(item, col_name):
                    value = getattr(item, col_name)
                    exists = await self.crud.exists(db, **{col_name: value})
                    if exists:  # pragma: no cover
                        raise DuplicateValueException(
                            f"Value {value} is already registered"
                        )

            if auto_fields:
                item_dict = item.model_dump()
                item_dict.update(auto_fields)
                db_object = self.model(**item_dict)
                db.add(db_object)
                await db.commit()
                await db.refresh(db_object)

                join_params = self._get_join_params()
                if self._should_include_relationships() and join_params:
                    pk_values = {
                        pk: getattr(db_object, pk) for pk in self.primary_key_names
                    }
                    result = await self.crud.get_joined(
                        db,
                        schema_to_select=cast(type[BaseModel], self.select_schema)
                        if self.select_schema
                        else None,
                        return_as_model=True if self.select_schema else False,
                        nest_joins=self.nest_joins,
                        **join_params,
                        **pk_values,
                    )
                    return result
                return db_object

            created_item = await self.crud.create(
                db, item, schema_to_select=self.select_schema
            )

            join_params = self._get_join_params()
            if self._should_include_relationships() and join_params and created_item:
                if isinstance(created_item, dict):
                    pk_values = {
                        pk: created_item.get(pk) for pk in self.primary_key_names
                    }
                else:
                    pk_values = {
                        pk: getattr(created_item, pk, None)
                        for pk in self.primary_key_names
                    }

                result = await self.crud.get_joined(
                    db,
                    schema_to_select=cast(type[BaseModel], self.select_schema)
                    if self.select_schema
                    else None,
                    return_as_model=True if self.select_schema else False,
                    nest_joins=self.nest_joins,
                    **join_params,
                    **pk_values,
                )
                return result

            return created_item

        endpoint.__annotations__["item"] = request_schema

        return endpoint

    def _read_item(self) -> Callable[..., Awaitable[Any]]:
        """Creates an endpoint for reading a single item from the database."""

        @apply_model_pk(**self._primary_keys_types)
        async def endpoint(db: AsyncSession = Depends(self.session), **pkeys):
            join_params = self._get_join_params()
            if self._should_include_relationships() and join_params:
                item = await self.crud.get_joined(
                    db,
                    schema_to_select=cast(type[BaseModel], self.select_schema)
                    if self.select_schema
                    else None,
                    return_as_model=True if self.select_schema else False,
                    nest_joins=self.nest_joins,
                    **join_params,
                    **pkeys,
                )
            else:
                if self.select_schema is not None:
                    item = await self.crud.get(
                        db,
                        schema_to_select=cast(type[BaseModel], self.select_schema),
                        return_as_model=True,
                        **pkeys,
                    )
                else:
                    item = await self.crud.get(db, **pkeys)
            if not item:  # pragma: no cover
                raise NotFoundException(detail="Item not found")
            return item  # pragma: no cover

        return cast(Callable[..., Awaitable[Any]], endpoint)

    def _read_items(
        self,
    ) -> Callable[
        ...,
        Awaitable[dict[str, Any] | PaginatedListResponse[Any] | ListResponse[Any]],
    ]:
        """Creates an endpoint for reading multiple items from the database.

        The created endpoint supports:
        - Pagination (using page/itemsPerPage or offset/limit)
        - Filtering based on configured filter parameters
        - Sorting by one or more fields (supports both ascending and descending order)

        Sorting can be applied using the 'sort' query parameter:
        - Single field ascending: ?sort=field_name
        - Single field descending: ?sort=-field_name
        - Multiple fields: ?sort=field1,-field2 (field1 asc, field2 desc)

        The query parameters are encapsulated in PaginatedRequestQuery schema,
        which can be reused in custom endpoints.
        """
        dynamic_filters = create_dynamic_filters(self.filter_config, self.column_types)

        async def endpoint(
            db: AsyncSession = Depends(self.session),
            query: PaginatedRequestQuery = Depends(),
            filters: dict = Depends(dynamic_filters),
        ) -> dict[str, Any] | PaginatedListResponse | ListResponse:
            is_paginated = (query.page is not None) or (
                query.items_per_page is not None
            )
            has_offset_limit = (query.offset is not None) and (query.limit is not None)
            default_offset = 0
            default_limit = 100

            if is_paginated and has_offset_limit:
                raise BadRequestException(
                    detail="Conflicting parameters: Use either 'page' and 'itemsPerPage' for paginated results or 'offset' and 'limit' for specific range queries."
                )

            if is_paginated:
                page = query.page if query.page else 1
                items_per_page = query.items_per_page if query.items_per_page else 10
                offset = compute_offset(page=page, items_per_page=items_per_page)  # type: ignore
                limit = items_per_page
            elif not has_offset_limit:
                offset = default_offset
                limit = default_limit
            else:
                offset = query.offset if query.offset is not None else default_offset
                limit = query.limit if query.limit is not None else default_limit

            sort_columns: list[str] = []
            sort_orders: list[str] = []
            if query.sort:
                for s in query.sort.split(","):
                    s = s.strip()
                    if not s:
                        continue
                    if s.startswith("-"):
                        sort_columns.append(s[1:])
                        sort_orders.append("desc")
                    else:
                        sort_columns.append(s)
                        sort_orders.append("asc")

            join_params = self._get_join_params()
            if self._should_include_relationships() and join_params:
                crud_data = await self.crud.get_multi_joined(
                    db,
                    offset=offset,  # type: ignore
                    limit=limit,  # type: ignore
                    schema_to_select=self.select_schema,
                    return_as_model=True if self.select_schema else False,
                    nest_joins=self.nest_joins,
                    sort_columns=sort_columns,
                    sort_orders=sort_orders,
                    **join_params,
                    **filters,
                )
            else:
                if self.select_schema is not None:
                    crud_data = await self.crud.get_multi(
                        db,
                        offset=offset,  # type: ignore
                        limit=limit,  # type: ignore
                        schema_to_select=self.select_schema,
                        sort_columns=sort_columns,
                        sort_orders=sort_orders,
                        return_as_model=True,
                        **filters,
                    )
                else:
                    crud_data = await self.crud.get_multi(
                        db,
                        offset=offset,  # type: ignore
                        limit=limit,  # type: ignore
                        sort_columns=sort_columns,
                        sort_orders=sort_orders,
                        **filters,
                    )

            if is_paginated:
                return paginated_response(
                    crud_data=crud_data,
                    page=page,  # type: ignore
                    items_per_page=items_per_page,  # type: ignore
                    multi_response_key=self.crud.multi_response_key,
                )

            return cast(dict[str, Any], crud_data)  # pragma: no cover

        return endpoint

    def _update_item(self) -> Callable[..., Awaitable[Any]]:
        """Creates an endpoint for updating an existing item in the database."""
        auto_field_injector = create_auto_field_injector(self.update_config)

        request_schema: type[BaseModel] = self.update_schema
        if self.update_config and self.update_config.exclude_from_schema:
            request_schema = create_modified_schema(
                self.update_schema,
                tuple(self.update_config.exclude_from_schema),
                f"{self.update_schema.__name__}Modified",
            )

        async def endpoint(
            item: BaseModel = Body(...),
            db: AsyncSession = Depends(self.session),
            auto_fields: dict = Depends(auto_field_injector),
            **pkeys,
        ):
            try:
                if auto_fields:
                    item_dict = item.model_dump(exclude_unset=True)
                    item_dict.update(auto_fields)
                    updated_item = await self.crud.update(db, item_dict, **pkeys)
                else:
                    updated_item = await self.crud.update(db, item, **pkeys)

                join_params = self._get_join_params()
                if self._should_include_relationships() and join_params:
                    result = await self.crud.get_joined(
                        db,
                        schema_to_select=cast(type[BaseModel], self.select_schema)
                        if self.select_schema
                        else None,
                        return_as_model=True if self.select_schema else False,
                        nest_joins=self.nest_joins,
                        **join_params,
                        **pkeys,
                    )
                    return result

                return updated_item
            except NoResultFound:
                raise NotFoundException(detail="Item not found")

        endpoint.__annotations__["item"] = request_schema

        endpoint = apply_model_pk(**self._primary_keys_types)(endpoint)

        return endpoint

    def _delete_item(self) -> Callable[..., Awaitable[Any]]:
        """Creates an endpoint for deleting (soft delete) an item from the database."""
        auto_field_injector = create_auto_field_injector(self.delete_config)

        async def endpoint(
            db: AsyncSession = Depends(self.session),
            auto_fields: dict = Depends(auto_field_injector),
            **pkeys,
        ):
            try:
                await self.crud.delete(db, **pkeys)

                if auto_fields:
                    await self.crud.update(
                        db, auto_fields, allow_multiple=False, **pkeys
                    )

                join_params = self._get_join_params()
                if self._should_include_relationships() and join_params:
                    result = await self.crud.get_joined(
                        db,
                        schema_to_select=cast(type[BaseModel], self.select_schema)
                        if self.select_schema
                        else None,
                        return_as_model=True if self.select_schema else False,
                        nest_joins=self.nest_joins,
                        **join_params,
                        **pkeys,
                    )
                    return result

                return {"message": "Item deleted successfully"}  # pragma: no cover
            except NoResultFound:
                raise NotFoundException(detail="Item not found")

        endpoint = apply_model_pk(**self._primary_keys_types)(endpoint)

        return endpoint

    def _db_delete(self) -> Callable[..., Awaitable[Any]]:
        """
        Creates an endpoint for hard deleting an item from the database.

        This endpoint is only added if the `delete_schema` is provided during initialization.
        The endpoint expects an item ID as a path parameter and uses the provided SQLAlchemy
        async session to permanently delete the item from the database.
        """

        @apply_model_pk(**self._primary_keys_types)
        async def endpoint(db: AsyncSession = Depends(self.session), **pkeys):
            item_to_delete = None
            join_params = self._get_join_params()
            if self._should_include_relationships() and join_params:
                item_to_delete = await self.crud.get_joined(
                    db,
                    schema_to_select=cast(type[BaseModel], self.select_schema)
                    if self.select_schema
                    else None,
                    return_as_model=True if self.select_schema else False,
                    nest_joins=self.nest_joins,
                    **join_params,
                    **pkeys,
                )

            await self.crud.db_delete(db, **pkeys)

            if item_to_delete:
                return item_to_delete

            return {
                "message": "Item permanently deleted from the database"
            }  # pragma: no cover

        return cast(Callable[..., Awaitable[Any]], endpoint)

    def _get_endpoint_path(self, operation: str) -> str:
        endpoint_name = self.endpoint_names.get(
            operation, self.default_endpoint_names.get(operation, operation)
        )
        path = f"{self.path}/{endpoint_name}" if endpoint_name else self.path

        if operation in {"read", "update", "delete", "db_delete"}:
            _primary_keys_path_suffix = "/".join(
                f"{{{n}}}" for n in self.primary_key_names
            )
            path = f"{path}/{_primary_keys_path_suffix}"

        return path

    def add_routes_to_router(
        self,
        create_deps: Sequence[Callable] = [],
        read_deps: Sequence[Callable] = [],
        read_multi_deps: Sequence[Callable] = [],
        update_deps: Sequence[Callable] = [],
        delete_deps: Sequence[Callable] = [],
        db_delete_deps: Sequence[Callable] = [],
        included_methods: Sequence[str] | None = None,
        deleted_methods: Sequence[str] | None = None,
    ):
        """
        Adds CRUD operation routes to the FastAPI router with specified dependencies for each type of operation.

        This method registers routes for create, read, update, and delete operations with the FastAPI router,
        allowing for custom dependency injection for each type of operation.

        Args:
            create_deps: List of functions to be injected as dependencies for the create endpoint.
            read_deps: List of functions to be injected as dependencies for the read endpoint.
            read_multi_deps: List of functions to be injected as dependencies for the read multiple items endpoint.
            update_deps: List of functions to be injected as dependencies for the update endpoint.
            delete_deps: List of functions to be injected as dependencies for the delete endpoint.
            db_delete_deps: List of functions to be injected as dependencies for the hard delete endpoint.
            included_methods: Optional list of methods to include. Defaults to all CRUD methods.
            deleted_methods: Optional list of methods to exclude. Defaults to `None`.

        Raises:
            ValueError: If both `included_methods` and `deleted_methods` are provided.

        Examples:
            Selective Endpoint Creation:

            ```python
            # Only create 'create' and 'read' endpoints
            endpoint_creator.add_routes_to_router(
                included_methods=["create", "read"],
            )
            ```

            Excluding Specific Endpoints:

            ```python
            # Create all endpoints except 'delete' and 'db_delete'
            endpoint_creator.add_routes_to_router(
                deleted_methods=["delete", "db_delete"],
            )
            ```

            With Custom Dependencies and Selective Endpoints:

            ```python
            def get_current_user(...):
                ...

            # Create only 'read' and 'update' endpoints with custom dependencies
            endpoint_creator.add_routes_to_router(
                read_deps=[get_current_user],
                update_deps=[get_current_user],
                included_methods=["read", "update"],
            )
            ```

        Note:
            This method should be called to register the endpoints with the FastAPI application.
            If `delete_schema` is provided on class instantiation, a hard delete endpoint is also registered.
            This method assumes `id` is the primary key for path parameters.
        """
        if (included_methods is not None) and (deleted_methods is not None):
            raise ValueError(
                "Cannot use both 'included_methods' and 'deleted_methods' simultaneously."
            )

        if included_methods is None:
            included_methods = [
                "create",
                "read",
                "read_multi",
                "update",
                "delete",
                "db_delete",
            ]
        else:
            try:
                included_methods = CRUDMethods(
                    valid_methods=included_methods
                ).valid_methods
            except ValidationError as e:
                raise ValueError(f"Invalid CRUD methods in included_methods: {e}")

        if deleted_methods is None:
            deleted_methods = []
        else:
            try:
                deleted_methods = CRUDMethods(
                    valid_methods=deleted_methods
                ).valid_methods
            except ValidationError as e:
                raise ValueError(f"Invalid CRUD methods in deleted_methods: {e}")

        delete_description = "Delete a"
        if self.delete_schema:
            delete_description = "Soft delete a"

        if ("create" in included_methods) and ("create" not in deleted_methods):
            self.router.add_api_route(
                self._get_endpoint_path(operation="create"),
                self._create_item(),
                methods=["POST"],
                include_in_schema=self.include_in_schema,
                tags=self.tags,
                dependencies=inject_dependencies(create_deps),
                name=f"{self.model.__name__.lower()}_create",
                description=f"Create a new {self.model.__name__} row in the database.",
            )

        if ("read" in included_methods) and ("read" not in deleted_methods):
            self.router.add_api_route(
                self._get_endpoint_path(operation="read"),
                self._read_item(),
                methods=["GET"],
                include_in_schema=self.include_in_schema,
                tags=self.tags,
                dependencies=inject_dependencies(read_deps),
                response_model=self.select_schema if self.select_schema else None,
                name=f"{self.model.__name__.lower()}_read",
                description=f"Read a single {self.model.__name__} row from the database by its primary keys: {self.primary_key_names}.",
            )

        if ("read_multi" in included_methods) and ("read_multi" not in deleted_methods):
            if self.select_schema is not None:
                response_model: (
                    type[PaginatedListResponse[Any] | ListResponse[Any]] | None
                ) = (
                    self.paginated_response_model | self.list_response_model  # type: ignore
                )
            else:
                response_model = None

            self.router.add_api_route(
                self._get_endpoint_path(operation="read_multi"),
                self._read_items(),
                methods=["GET"],
                include_in_schema=self.include_in_schema,
                tags=self.tags,
                dependencies=inject_dependencies(read_multi_deps),
                response_model=response_model,
                name=f"{self.model.__name__.lower()}_read_multi",
                description=(
                    f"Read multiple {self.model.__name__} rows from the database.\n\n"
                    f"**Pagination Options:**\n"
                    f"- Use `page` & `itemsPerPage` for paginated results\n"
                    f"- Use `offset` & `limit` for specific ranges\n\n"
                    f"**Sorting:**\n"
                    f"- Use `sort` parameter to sort results by one or more fields\n"
                    f"- Format: `field1,-field2` (comma-separated, `-` prefix for descending)\n"
                    f"- Examples: `name` (ascending), `-age` (descending), `name,-age` (mixed)\n\n"
                    f"**Response Format:**\n"
                    f"- Returns paginated response when using page/itemsPerPage\n"
                    f"- Returns simple list response when using offset/limit"
                ),
            )

        if ("update" in included_methods) and ("update" not in deleted_methods):
            self.router.add_api_route(
                self._get_endpoint_path(operation="update"),
                self._update_item(),
                methods=["PATCH"],
                include_in_schema=self.include_in_schema,
                tags=self.tags,
                dependencies=inject_dependencies(update_deps),
                name=f"{self.model.__name__.lower()}_update",
                description=f"Update an existing {self.model.__name__} row in the database by its primary keys: {self.primary_key_names}.",
            )

        if ("delete" in included_methods) and ("delete" not in deleted_methods):
            path = self._get_endpoint_path(operation="delete")
            self.router.add_api_route(
                path,
                self._delete_item(),
                methods=["DELETE"],
                include_in_schema=self.include_in_schema,
                tags=self.tags,
                dependencies=inject_dependencies(delete_deps),
                name=f"{self.model.__name__.lower()}_delete",
                description=f"{delete_description} {self.model.__name__} row from the database by its primary keys: {self.primary_key_names}.",
            )

        if (
            ("db_delete" in included_methods)
            and ("db_delete" not in deleted_methods)
            and self.delete_schema
        ):
            self.router.add_api_route(
                self._get_endpoint_path(operation="db_delete"),
                self._db_delete(),
                methods=["DELETE"],
                include_in_schema=self.include_in_schema,
                tags=self.tags,
                dependencies=inject_dependencies(db_delete_deps),
                name=f"{self.model.__name__.lower()}_db_delete",
                description=f"Permanently delete a {self.model.__name__} row from the database by its primary keys: {self.primary_key_names}.",
            )

    def add_custom_route(
        self,
        endpoint: Callable,
        methods: set[str] | list[str] | None,
        path: str | None = None,
        dependencies: Sequence[Callable] | None = None,
        include_in_schema: bool = True,
        tags: list[str | Enum] | None = None,
        summary: str | None = None,
        description: str | None = None,
        response_description: str = "Successful Response",
    ) -> None:
        """
        Adds a custom route to the FastAPI router.

        Args:
            endpoint: The endpoint function to execute when the route is called.
            methods: A list of HTTP methods for the route (e.g., `['GET', 'POST']`).
            path: URL path for the custom route.
            dependencies: A list of functions to be injected as dependencies for the route.
            include_in_schema: Whether to include this route in the OpenAPI schema.
            tags: Tags for grouping and categorizing the route in documentation.
            summary: A short summary of the route, for documentation.
            description: A detailed description of the route, for documentation.
            response_description: A description of the response, for documentation.

        Example:
            ```python
            async def custom_endpoint(foo: int, bar: str):
                # custom logic here
                return {"foo": foo, "bar": bar}

            endpoint_creator.add_custom_route(
                endpoint=custom_endpoint,
                methods=["GET"],
                path="/custom",
                tags=["custom"],
                summary="Custom Endpoint",
                description="This is a custom endpoint.",
            )
            ```
        """
        path = path or self.path
        full_path = f"{self.path}{path}"
        self.router.add_api_route(
            path=full_path,
            endpoint=endpoint,
            methods=methods,
            dependencies=inject_dependencies(dependencies) or [],
            include_in_schema=include_in_schema,
            tags=tags or self.tags,
            summary=summary,
            description=description,
            response_description=response_description,
        )

add_custom_route(endpoint, methods, path=None, dependencies=None, include_in_schema=True, tags=None, summary=None, description=None, response_description='Successful Response')

Adds a custom route to the FastAPI router.

Parameters:

Name	Type	Description	Default
endpoint	Callable	The endpoint function to execute when the route is called.	required
methods	set[str] | list[str] | None	A list of HTTP methods for the route (e.g., ['GET', 'POST']).	required
path	str | None	URL path for the custom route.	None
dependencies	Sequence[Callable] | None	A list of functions to be injected as dependencies for the route.	None
include_in_schema	bool	Whether to include this route in the OpenAPI schema.	True
tags	list[str | Enum] | None	Tags for grouping and categorizing the route in documentation.	None
summary	str | None	A short summary of the route, for documentation.	None
description	str | None	A detailed description of the route, for documentation.	None
response_description	str	A description of the response, for documentation.	'Successful Response'

Example

async def custom_endpoint(foo: int, bar: str):
    # custom logic here
    return {"foo": foo, "bar": bar}

endpoint_creator.add_custom_route(
    endpoint=custom_endpoint,
    methods=["GET"],
    path="/custom",
    tags=["custom"],
    summary="Custom Endpoint",
    description="This is a custom endpoint.",
)

Source code in fastcrud/endpoint/endpoint_creator.py

def add_custom_route(
    self,
    endpoint: Callable,
    methods: set[str] | list[str] | None,
    path: str | None = None,
    dependencies: Sequence[Callable] | None = None,
    include_in_schema: bool = True,
    tags: list[str | Enum] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
) -> None:
    """
    Adds a custom route to the FastAPI router.

    Args:
        endpoint: The endpoint function to execute when the route is called.
        methods: A list of HTTP methods for the route (e.g., `['GET', 'POST']`).
        path: URL path for the custom route.
        dependencies: A list of functions to be injected as dependencies for the route.
        include_in_schema: Whether to include this route in the OpenAPI schema.
        tags: Tags for grouping and categorizing the route in documentation.
        summary: A short summary of the route, for documentation.
        description: A detailed description of the route, for documentation.
        response_description: A description of the response, for documentation.

    Example:
        ```python
        async def custom_endpoint(foo: int, bar: str):
            # custom logic here
            return {"foo": foo, "bar": bar}

        endpoint_creator.add_custom_route(
            endpoint=custom_endpoint,
            methods=["GET"],
            path="/custom",
            tags=["custom"],
            summary="Custom Endpoint",
            description="This is a custom endpoint.",
        )
        ```
    """
    path = path or self.path
    full_path = f"{self.path}{path}"
    self.router.add_api_route(
        path=full_path,
        endpoint=endpoint,
        methods=methods,
        dependencies=inject_dependencies(dependencies) or [],
        include_in_schema=include_in_schema,
        tags=tags or self.tags,
        summary=summary,
        description=description,
        response_description=response_description,
    )

add_routes_to_router(create_deps=[], read_deps=[], read_multi_deps=[], update_deps=[], delete_deps=[], db_delete_deps=[], included_methods=None, deleted_methods=None)

Adds CRUD operation routes to the FastAPI router with specified dependencies for each type of operation.

This method registers routes for create, read, update, and delete operations with the FastAPI router, allowing for custom dependency injection for each type of operation.

Parameters:

Name	Type	Description	Default
create_deps	Sequence[Callable]	List of functions to be injected as dependencies for the create endpoint.	[]
read_deps	Sequence[Callable]	List of functions to be injected as dependencies for the read endpoint.	[]
read_multi_deps	Sequence[Callable]	List of functions to be injected as dependencies for the read multiple items endpoint.	[]
update_deps	Sequence[Callable]	List of functions to be injected as dependencies for the update endpoint.	[]
delete_deps	Sequence[Callable]	List of functions to be injected as dependencies for the delete endpoint.	[]
db_delete_deps	Sequence[Callable]	List of functions to be injected as dependencies for the hard delete endpoint.	[]
included_methods	Sequence[str] | None	Optional list of methods to include. Defaults to all CRUD methods.	None
deleted_methods	Sequence[str] | None	Optional list of methods to exclude. Defaults to None.	None

Raises:

Type	Description
ValueError	If both included_methods and deleted_methods are provided.

Examples:

Selective Endpoint Creation:

# Only create 'create' and 'read' endpoints
endpoint_creator.add_routes_to_router(
    included_methods=["create", "read"],
)

Excluding Specific Endpoints:

# Create all endpoints except 'delete' and 'db_delete'
endpoint_creator.add_routes_to_router(
    deleted_methods=["delete", "db_delete"],
)

With Custom Dependencies and Selective Endpoints:

def get_current_user(...):
    ...

# Create only 'read' and 'update' endpoints with custom dependencies
endpoint_creator.add_routes_to_router(
    read_deps=[get_current_user],
    update_deps=[get_current_user],
    included_methods=["read", "update"],
)

Note

This method should be called to register the endpoints with the FastAPI application. If delete_schema is provided on class instantiation, a hard delete endpoint is also registered. This method assumes id is the primary key for path parameters.

Source code in fastcrud/endpoint/endpoint_creator.py

def add_routes_to_router(
    self,
    create_deps: Sequence[Callable] = [],
    read_deps: Sequence[Callable] = [],
    read_multi_deps: Sequence[Callable] = [],
    update_deps: Sequence[Callable] = [],
    delete_deps: Sequence[Callable] = [],
    db_delete_deps: Sequence[Callable] = [],
    included_methods: Sequence[str] | None = None,
    deleted_methods: Sequence[str] | None = None,
):
    """
    Adds CRUD operation routes to the FastAPI router with specified dependencies for each type of operation.

    This method registers routes for create, read, update, and delete operations with the FastAPI router,
    allowing for custom dependency injection for each type of operation.

    Args:
        create_deps: List of functions to be injected as dependencies for the create endpoint.
        read_deps: List of functions to be injected as dependencies for the read endpoint.
        read_multi_deps: List of functions to be injected as dependencies for the read multiple items endpoint.
        update_deps: List of functions to be injected as dependencies for the update endpoint.
        delete_deps: List of functions to be injected as dependencies for the delete endpoint.
        db_delete_deps: List of functions to be injected as dependencies for the hard delete endpoint.
        included_methods: Optional list of methods to include. Defaults to all CRUD methods.
        deleted_methods: Optional list of methods to exclude. Defaults to `None`.

    Raises:
        ValueError: If both `included_methods` and `deleted_methods` are provided.

    Examples:
        Selective Endpoint Creation:

        ```python
        # Only create 'create' and 'read' endpoints
        endpoint_creator.add_routes_to_router(
            included_methods=["create", "read"],
        )
        ```

        Excluding Specific Endpoints:

        ```python
        # Create all endpoints except 'delete' and 'db_delete'
        endpoint_creator.add_routes_to_router(
            deleted_methods=["delete", "db_delete"],
        )
        ```

        With Custom Dependencies and Selective Endpoints:

        ```python
        def get_current_user(...):
            ...

        # Create only 'read' and 'update' endpoints with custom dependencies
        endpoint_creator.add_routes_to_router(
            read_deps=[get_current_user],
            update_deps=[get_current_user],
            included_methods=["read", "update"],
        )
        ```

    Note:
        This method should be called to register the endpoints with the FastAPI application.
        If `delete_schema` is provided on class instantiation, a hard delete endpoint is also registered.
        This method assumes `id` is the primary key for path parameters.
    """
    if (included_methods is not None) and (deleted_methods is not None):
        raise ValueError(
            "Cannot use both 'included_methods' and 'deleted_methods' simultaneously."
        )

    if included_methods is None:
        included_methods = [
            "create",
            "read",
            "read_multi",
            "update",
            "delete",
            "db_delete",
        ]
    else:
        try:
            included_methods = CRUDMethods(
                valid_methods=included_methods
            ).valid_methods
        except ValidationError as e:
            raise ValueError(f"Invalid CRUD methods in included_methods: {e}")

    if deleted_methods is None:
        deleted_methods = []
    else:
        try:
            deleted_methods = CRUDMethods(
                valid_methods=deleted_methods
            ).valid_methods
        except ValidationError as e:
            raise ValueError(f"Invalid CRUD methods in deleted_methods: {e}")

    delete_description = "Delete a"
    if self.delete_schema:
        delete_description = "Soft delete a"

    if ("create" in included_methods) and ("create" not in deleted_methods):
        self.router.add_api_route(
            self._get_endpoint_path(operation="create"),
            self._create_item(),
            methods=["POST"],
            include_in_schema=self.include_in_schema,
            tags=self.tags,
            dependencies=inject_dependencies(create_deps),
            name=f"{self.model.__name__.lower()}_create",
            description=f"Create a new {self.model.__name__} row in the database.",
        )

    if ("read" in included_methods) and ("read" not in deleted_methods):
        self.router.add_api_route(
            self._get_endpoint_path(operation="read"),
            self._read_item(),
            methods=["GET"],
            include_in_schema=self.include_in_schema,
            tags=self.tags,
            dependencies=inject_dependencies(read_deps),
            response_model=self.select_schema if self.select_schema else None,
            name=f"{self.model.__name__.lower()}_read",
            description=f"Read a single {self.model.__name__} row from the database by its primary keys: {self.primary_key_names}.",
        )

    if ("read_multi" in included_methods) and ("read_multi" not in deleted_methods):
        if self.select_schema is not None:
            response_model: (
                type[PaginatedListResponse[Any] | ListResponse[Any]] | None
            ) = (
                self.paginated_response_model | self.list_response_model  # type: ignore
            )
        else:
            response_model = None

        self.router.add_api_route(
            self._get_endpoint_path(operation="read_multi"),
            self._read_items(),
            methods=["GET"],
            include_in_schema=self.include_in_schema,
            tags=self.tags,
            dependencies=inject_dependencies(read_multi_deps),
            response_model=response_model,
            name=f"{self.model.__name__.lower()}_read_multi",
            description=(
                f"Read multiple {self.model.__name__} rows from the database.\n\n"
                f"**Pagination Options:**\n"
                f"- Use `page` & `itemsPerPage` for paginated results\n"
                f"- Use `offset` & `limit` for specific ranges\n\n"
                f"**Sorting:**\n"
                f"- Use `sort` parameter to sort results by one or more fields\n"
                f"- Format: `field1,-field2` (comma-separated, `-` prefix for descending)\n"
                f"- Examples: `name` (ascending), `-age` (descending), `name,-age` (mixed)\n\n"
                f"**Response Format:**\n"
                f"- Returns paginated response when using page/itemsPerPage\n"
                f"- Returns simple list response when using offset/limit"
            ),
        )

    if ("update" in included_methods) and ("update" not in deleted_methods):
        self.router.add_api_route(
            self._get_endpoint_path(operation="update"),
            self._update_item(),
            methods=["PATCH"],
            include_in_schema=self.include_in_schema,
            tags=self.tags,
            dependencies=inject_dependencies(update_deps),
            name=f"{self.model.__name__.lower()}_update",
            description=f"Update an existing {self.model.__name__} row in the database by its primary keys: {self.primary_key_names}.",
        )

    if ("delete" in included_methods) and ("delete" not in deleted_methods):
        path = self._get_endpoint_path(operation="delete")
        self.router.add_api_route(
            path,
            self._delete_item(),
            methods=["DELETE"],
            include_in_schema=self.include_in_schema,
            tags=self.tags,
            dependencies=inject_dependencies(delete_deps),
            name=f"{self.model.__name__.lower()}_delete",
            description=f"{delete_description} {self.model.__name__} row from the database by its primary keys: {self.primary_key_names}.",
        )

    if (
        ("db_delete" in included_methods)
        and ("db_delete" not in deleted_methods)
        and self.delete_schema
    ):
        self.router.add_api_route(
            self._get_endpoint_path(operation="db_delete"),
            self._db_delete(),
            methods=["DELETE"],
            include_in_schema=self.include_in_schema,
            tags=self.tags,
            dependencies=inject_dependencies(db_delete_deps),
            name=f"{self.model.__name__.lower()}_db_delete",
            description=f"Permanently delete a {self.model.__name__} row from the database by its primary keys: {self.primary_key_names}.",
        )