forge.crud

Template-based CRUD code generation for FastAPI from Pydantic schemas.

forge.crud

forge.crud — Template-based CRUD code generation for FastAPI.

Generates Create/Read/Update/Delete route handlers from a Pydantic schema with pagination, filtering, sorting, and optional soft-delete support.

Classes

CrudError

Bases: ForgeError

Base exception for all CRUD generation errors.

Source code in src/forge/crud/exceptions.py

class CrudError(ForgeError):
    """Base exception for all CRUD generation errors."""

CrudGenerationError

Bases: CrudError

Raised when CRUD code generation fails.

Source code in src/forge/crud/exceptions.py

class CrudGenerationError(CrudError):
    """Raised when CRUD code generation fails."""

CrudGenerator

Template-based CRUD code generator.

Analyzes a Pydantic schema and generates FastAPI route handler files with configurable operations, pagination, filtering, sorting, and soft-delete support.

Usage

generator = CrudGenerator(User, output_dir="app/crud/generated") generator.generate()

Source code in src/forge/crud/generator.py

class CrudGenerator:
    """
    Template-based CRUD code generator.

    Analyzes a Pydantic schema and generates FastAPI route handler files
    with configurable operations, pagination, filtering, sorting,
    and soft-delete support.

    Usage:
        generator = CrudGenerator(User, output_dir="app/crud/generated")
        generator.generate()
    """

    def __init__(
        self,
        schema: type[BaseModel],
        output_dir: str = ".",
        operations: set[CrudOperation] | None = None,
        auth_dependency: str | None = None,
        response_model: str | None = None,
        create_schema: str | None = None,
        update_schema: str | None = None,
        pagination: bool = True,
        filter_fields: list[str] | None = None,
        sort_fields: list[str] | None = None,
        soft_delete: bool = False,
        primary_key: str | None = None,
        table_name: str | None = None,
        config: CrudGeneratorConfig | None = None,
    ) -> None:
        if config is not None:
            self._config = config
        else:
            self._config = CrudGeneratorConfig(
                schema_name=schema.__name__,
                operations=operations
                if operations is not None
                else {
                    CrudOperation.CREATE,
                    CrudOperation.LIST,
                    CrudOperation.READ,
                    CrudOperation.UPDATE,
                    CrudOperation.DELETE,
                },
                auth_dependency=auth_dependency,
                response_model=response_model,
                create_schema=create_schema,
                update_schema=update_schema,
                output_dir=output_dir,
                pagination=pagination,
                filter_fields=filter_fields,
                sort_fields=sort_fields,
                soft_delete=soft_delete,
                primary_key=primary_key or "id",
                table_name=table_name,
            )
        self._schema = schema
        self._validate()

    def _validate(self) -> None:
        """Validate that the schema and configuration are usable."""
        if not self._config.operations:
            raise CrudGenerationError(
                "At least one CRUD operation must be specified in 'operations'."
            )

    @property
    def config(self) -> CrudGeneratorConfig:
        """Return the current generator configuration."""
        return self._config

    def build_context(self) -> dict[str, Any]:
        """Build the Jinja2 template context from the schema and config."""
        return build_template_context(self._schema, self._config)

    def render(self, template_name: str = "crud_router.py.jinja") -> str:
        """
        Render the CRUD template into generated source code.

        Args:
            template_name: Name of the Jinja2 template file to render.

        Returns:
            The rendered source code as a string.
        """
        env = _build_template_env()
        try:
            template = env.get_template(template_name)
        except jinja2.TemplateNotFound as exc:
            raise TemplateNotFoundError(
                f"Template '{template_name}' not found. "
                f"Available templates: {', '.join(_list_templates())}"
            ) from exc

        context = self.build_context()
        return template.render(**context)

    def generate(
        self,
        template_name: str = "crud_router.py.jinja",
        output_filename: str | None = None,
        force: bool = False,
    ) -> Path:
        """
        Generate the CRUD router file and write it to disk.

        Args:
            template_name: The template file to render.
            output_filename: Optional custom output filename.
                Defaults to '{table_name}.py'.
            force: Overwrite existing file if True.

        Returns:
            The path to the generated file.

        Raises:
            CrudGenerationError: If the output file already exists
                and force is False.
        """
        rendered = self.render(template_name)
        output_path = self._resolve_output_path(output_filename)

        if output_path.exists() and not force:
            raise CrudGenerationError(
                f"Output file already exists: {output_path}\n\n"
                f"Use force=True to overwrite, or specify a different output filename."
            )

        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(rendered, encoding="utf-8")
        return output_path

    def _resolve_output_path(self, output_filename: str | None = None) -> Path:
        """Resolve the output file path from config and optional filename."""
        base = Path(self._config.output_dir)
        name = output_filename or f"{self._config.resolved_table_name}.py"
        return base / name

Attributes

config property

config: CrudGeneratorConfig

Return the current generator configuration.

Methods:

build_context

build_context() -> dict[str, Any]

Build the Jinja2 template context from the schema and config.

Source code in src/forge/crud/generator.py

def build_context(self) -> dict[str, Any]:
    """Build the Jinja2 template context from the schema and config."""
    return build_template_context(self._schema, self._config)

generate

generate(template_name: str = 'crud_router.py.jinja', output_filename: str | None = None, force: bool = False) -> Path

Generate the CRUD router file and write it to disk.

Parameters:

Name	Type	Description	Default
template_name	str	The template file to render.	'crud_router.py.jinja'
output_filename	str | None	Optional custom output filename. Defaults to '{table_name}.py'.	None
force	bool	Overwrite existing file if True.	False

Returns:

Type	Description
Path	The path to the generated file.

Raises:

Type	Description
CrudGenerationError	If the output file already exists and force is False.

Source code in src/forge/crud/generator.py

def generate(
    self,
    template_name: str = "crud_router.py.jinja",
    output_filename: str | None = None,
    force: bool = False,
) -> Path:
    """
    Generate the CRUD router file and write it to disk.

    Args:
        template_name: The template file to render.
        output_filename: Optional custom output filename.
            Defaults to '{table_name}.py'.
        force: Overwrite existing file if True.

    Returns:
        The path to the generated file.

    Raises:
        CrudGenerationError: If the output file already exists
            and force is False.
    """
    rendered = self.render(template_name)
    output_path = self._resolve_output_path(output_filename)

    if output_path.exists() and not force:
        raise CrudGenerationError(
            f"Output file already exists: {output_path}\n\n"
            f"Use force=True to overwrite, or specify a different output filename."
        )

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(rendered, encoding="utf-8")
    return output_path

render

render(template_name: str = 'crud_router.py.jinja') -> str

Render the CRUD template into generated source code.

Parameters:

Name	Type	Description	Default
template_name	str	Name of the Jinja2 template file to render.	'crud_router.py.jinja'

Returns:

Type	Description
str	The rendered source code as a string.

Source code in src/forge/crud/generator.py

def render(self, template_name: str = "crud_router.py.jinja") -> str:
    """
    Render the CRUD template into generated source code.

    Args:
        template_name: Name of the Jinja2 template file to render.

    Returns:
        The rendered source code as a string.
    """
    env = _build_template_env()
    try:
        template = env.get_template(template_name)
    except jinja2.TemplateNotFound as exc:
        raise TemplateNotFoundError(
            f"Template '{template_name}' not found. "
            f"Available templates: {', '.join(_list_templates())}"
        ) from exc

    context = self.build_context()
    return template.render(**context)

CrudGeneratorConfig

Bases: BaseModel

Configuration for the CRUD generator.

Controls which operations are generated, how the output is structured, and what features (pagination, filtering, sorting, soft delete) are included.

Source code in src/forge/crud/models.py

class CrudGeneratorConfig(BaseModel):
    """
    Configuration for the CRUD generator.

    Controls which operations are generated, how the output is structured,
    and what features (pagination, filtering, sorting, soft delete) are included.
    """

    schema_name: str
    operations: set[CrudOperation] = Field(
        default_factory=lambda: {
            CrudOperation.CREATE,
            CrudOperation.LIST,
            CrudOperation.READ,
            CrudOperation.UPDATE,
            CrudOperation.DELETE,
        },
    )
    auth_dependency: str | None = None
    response_model: str | None = None
    create_schema: str | None = None
    update_schema: str | None = None
    output_dir: str = "."
    pagination: bool = True
    filter_fields: list[str] | None = None
    sort_fields: list[str] | None = None
    soft_delete: bool = False
    primary_key: str = "id"
    table_name: str | None = None
    package_name: str | None = None

    @property
    def resolved_table_name(self) -> str:
        """Return the table name, derived from schema_name if not set."""
        if self.table_name:
            return self.table_name
        return _to_snake(self.schema_name) + "s"

    @property
    def resolved_package_name(self) -> str:
        """Return the package name, derived from output_dir if not set."""
        if self.package_name:
            return self.package_name
        return self.output_dir.replace("/", ".").replace("\\", ".").strip(".")

    @property
    def resolved_create_schema(self) -> str:
        """Return the create schema class name."""
        return self.create_schema or f"{self.schema_name}Create"

    @property
    def resolved_update_schema(self) -> str:
        """Return the update schema class name."""
        return self.update_schema or f"{self.schema_name}Update"

    @property
    def resolved_response_model(self) -> str:
        """Return the response model class name."""
        return self.response_model or f"{self.schema_name}Response"

Attributes

resolved_create_schema property

resolved_create_schema: str

Return the create schema class name.

resolved_package_name property

resolved_package_name: str

Return the package name, derived from output_dir if not set.

resolved_response_model property

resolved_response_model: str

Return the response model class name.

resolved_table_name property

resolved_table_name: str

Return the table name, derived from schema_name if not set.

resolved_update_schema property

resolved_update_schema: str

Return the update schema class name.

CrudModule

Bases: ForgeModule

Forge module providing CRUD code generation capabilities.

The CrudModule integrates the CRUD generator into the forge runtime. It is a lightweight module that makes the CrudGenerator and generate_crud convenience function available through the runtime, enabling programmatic code generation during development workflows.

Source code in src/forge/crud/module.py

class CrudModule(ForgeModule):
    """
    Forge module providing CRUD code generation capabilities.

    The CrudModule integrates the CRUD generator into the forge runtime.
    It is a lightweight module that makes the ``CrudGenerator`` and
    ``generate_crud`` convenience function available through the runtime,
    enabling programmatic code generation during development workflows.
    """

    name = "crud"
    dependencies: ClassVar[list[str]] = []

    async def setup(self, runtime: Runtime) -> None:
        """Initialise the CRUD module."""

    async def teardown(self) -> None:
        """Teardown the CRUD module."""

    def health_check(self) -> HealthResult:
        """Return health status of the CRUD module."""
        return HealthResult.ok()

Methods:

health_check

health_check() -> HealthResult

Return health status of the CRUD module.

Source code in src/forge/crud/module.py

def health_check(self) -> HealthResult:
    """Return health status of the CRUD module."""
    return HealthResult.ok()

setup async

setup(runtime: ForgeRuntime) -> None

Initialise the CRUD module.

Source code in src/forge/crud/module.py

async def setup(self, runtime: Runtime) -> None:
    """Initialise the CRUD module."""

teardown async

teardown() -> None

Teardown the CRUD module.

Source code in src/forge/crud/module.py

async def teardown(self) -> None:
    """Teardown the CRUD module."""

CrudOperation

Bases: StrEnum

Supported CRUD operations for code generation.

Source code in src/forge/crud/models.py

class CrudOperation(StrEnum):
    """Supported CRUD operations for code generation."""

    CREATE = "create"
    LIST = "list"
    READ = "read"
    UPDATE = "update"
    DELETE = "delete"

FieldInfo

Bases: BaseModel

Describes a single field from a Pydantic schema for template context.

Source code in src/forge/crud/models.py

class FieldInfo(BaseModel):
    """Describes a single field from a Pydantic schema for template context."""

    name: str
    type_hint: str
    default: str | None = None
    required: bool = True
    is_primary_key: bool = False
    is_timestamp: bool = False
    is_soft_delete: bool = False

    def declaration(self, for_create: bool = True) -> str:
        """Return the Python field declaration line."""
        if (self.is_primary_key or self.is_timestamp or self.is_soft_delete) and for_create:
            return ""
        if self.required:
            return f"{self.name}: {self.type_hint}"
        if self.default is not None:
            return f"{self.name}: {self.type_hint} = {self.default}"
        if " | None" in self.type_hint or " | " in self.type_hint:
            return f"{self.name}: {self.type_hint} = None"
        return f"{self.name}: {self.type_hint} | None = None"

Methods:

declaration

declaration(for_create: bool = True) -> str

Return the Python field declaration line.

Source code in src/forge/crud/models.py

def declaration(self, for_create: bool = True) -> str:
    """Return the Python field declaration line."""
    if (self.is_primary_key or self.is_timestamp or self.is_soft_delete) and for_create:
        return ""
    if self.required:
        return f"{self.name}: {self.type_hint}"
    if self.default is not None:
        return f"{self.name}: {self.type_hint} = {self.default}"
    if " | None" in self.type_hint or " | " in self.type_hint:
        return f"{self.name}: {self.type_hint} = None"
    return f"{self.name}: {self.type_hint} | None = None"

SchemaValidationError

Bases: CrudError

Raised when the provided Pydantic schema is invalid for CRUD generation.

Source code in src/forge/crud/exceptions.py

class SchemaValidationError(CrudError):
    """Raised when the provided Pydantic schema is invalid for CRUD generation."""

TemplateNotFoundError

Bases: CrudError

Raised when a required template file is not found.

Source code in src/forge/crud/exceptions.py

class TemplateNotFoundError(CrudError):
    """Raised when a required template file is not found."""

Functions:

generate_crud

generate_crud(schema: type, output_dir: str = '.', operations: set[CrudOperation] | None = None, auth_dependency: str | None = None, response_model: str | None = None, create_schema: str | None = None, update_schema: str | None = None, pagination: bool = True, filter_fields: list[str] | None = None, sort_fields: list[str] | None = None, soft_delete: bool = False, primary_key: str | None = None, table_name: str | None = None, force: bool = False) -> Path

Convenience function to generate CRUD routes in one call.

Parameters:

Name	Type	Description	Default
schema	type	A Pydantic model class to generate CRUD routes for.	required
output_dir	str	Directory to write the generated file to.	'.'
operations	set[CrudOperation] | None	Set of CRUD operations to generate.	None
auth_dependency	str | None	Fully qualified auth dependency (e.g. 'app.auth.get_current_user').	None
response_model	str | None	Custom response model class name.	None
create_schema	str | None	Custom create schema class name.	None
update_schema	str | None	Custom update schema class name.	None
pagination	bool	Enable pagination for list endpoint.	True
filter_fields	list[str] | None	List of field names to enable filtering on.	None
sort_fields	list[str] | None	List of field names to enable sorting on.	None
soft_delete	bool	Enable soft-delete support.	False
primary_key	str | None	Primary key field name.	None
table_name	str | None	Database table name (used for URL prefix).	None
force	bool	Overwrite existing output file.	False

Returns:

Type	Description
Path	The path to the generated file.

Example

from pydantic import BaseModel from forge.crud import generate_crud

class User(BaseModel): ... id: int ... name: str ... email: str ... path = generate_crud( ... User, ... output_dir="app/crud/generated", ... auth_dependency="app.auth.get_current_user", ... soft_delete=True, ... )

Source code in src/forge/crud/generator.py

def generate_crud(
    schema: type,
    output_dir: str = ".",
    operations: set[CrudOperation] | None = None,
    auth_dependency: str | None = None,
    response_model: str | None = None,
    create_schema: str | None = None,
    update_schema: str | None = None,
    pagination: bool = True,
    filter_fields: list[str] | None = None,
    sort_fields: list[str] | None = None,
    soft_delete: bool = False,
    primary_key: str | None = None,
    table_name: str | None = None,
    force: bool = False,
) -> Path:
    """
    Convenience function to generate CRUD routes in one call.

    Args:
        schema: A Pydantic model class to generate CRUD routes for.
        output_dir: Directory to write the generated file to.
        operations: Set of CRUD operations to generate.
        auth_dependency: Fully qualified auth dependency (e.g. 'app.auth.get_current_user').
        response_model: Custom response model class name.
        create_schema: Custom create schema class name.
        update_schema: Custom update schema class name.
        pagination: Enable pagination for list endpoint.
        filter_fields: List of field names to enable filtering on.
        sort_fields: List of field names to enable sorting on.
        soft_delete: Enable soft-delete support.
        primary_key: Primary key field name.
        table_name: Database table name (used for URL prefix).
        force: Overwrite existing output file.

    Returns:
        The path to the generated file.

    Example:
        >>> from pydantic import BaseModel
        >>> from forge.crud import generate_crud
        >>>
        >>> class User(BaseModel):
        ...     id: int
        ...     name: str
        ...     email: str
        ...
        >>> path = generate_crud(
        ...     User,
        ...     output_dir="app/crud/generated",
        ...     auth_dependency="app.auth.get_current_user",
        ...     soft_delete=True,
        ... )
    """
    generator = CrudGenerator(
        schema=schema,
        output_dir=output_dir,
        operations=operations,
        auth_dependency=auth_dependency,
        response_model=response_model,
        create_schema=create_schema,
        update_schema=update_schema,
        pagination=pagination,
        filter_fields=filter_fields,
        sort_fields=sort_fields,
        soft_delete=soft_delete,
        primary_key=primary_key,
        table_name=table_name,
    )
    return generator.generate(force=force)