auto_rest.routers

API routers are responsible for redirecting incoming HTTP requests to the appropriate handling logic. Router objects are created using a factory pattern, with each router being responsible for a single application resource. Each factory returns an APIRouter instance preconfigured with request handling logic for the relevant resource. This allows routers to be added directly to an API application instance.

Example: Creating and Adding a Router

Care should be taken to avoid path conflicts when adding routers to an API application instance. Using a unique prefix value ensures that each router's endpoints are properly namespaced and unique.

from fastapi import FastAPI
from auto_rest.routers import create_welcome_router

app = FastAPI()
welcome_router = create_welcome_router()
app.include_router(welcome_router, prefix="/welcome")

create_about_handler(name, version)

Create an endpoint handler that returns the application name and version number.

Parameters:

Name	Type	Description	Default
name	str	The application name.	required
version	str	The returned version identifier.	required

Returns:

Type	Description
Callable[[], Awaitable[BaseModel]]	An async function that returns application info.

Source code in auto_rest/handlers.py

def create_about_handler(name: str, version: str) -> Callable[[], Awaitable[PydanticModel]]:
    """Create an endpoint handler that returns the application name and version number.

    Args:
        name: The application name.
        version: The returned version identifier.

    Returns:
        An async function that returns application info.
    """

    interface = create_model("Version", version=(str, version), name=(str, name))

    async def about_handler() -> interface:
        """Return the application name and version number."""

        return interface()

    return about_handler

create_delete_record_handler(engine, table)

Create a function for handling DELETE requests against a record in the database.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to use when executing queries.	required
table	Table	The database table to query against.	required

Returns:

Type	Description
Callable[..., Awaitable[None]]	An async function that handles record deletion.

Source code in auto_rest/handlers.py

def create_delete_record_handler(engine: DBEngine, table: Table) -> Callable[..., Awaitable[None]]:
    """Create a function for handling DELETE requests against a record in the database.

    Args:
        engine: Database engine to use when executing queries.
        table: The database table to query against.

    Returns:
        An async function that handles record deletion.
    """

    pk_interface = create_interface(table, pk_only=True, mode='required')

    async def delete_record_handler(
        pk: pk_interface = Depends(),
        session: DBSession = Depends(create_session_iterator(engine)),
    ) -> None:
        """Delete a record from the database."""

        query = select(table).filter_by(**pk.model_dump())
        result = await execute_session_query(session, query)
        record = get_record_or_404(result)

        await delete_session_record(session, record)
        await commit_session(session)

    return delete_record_handler

create_engine_handler(engine)

Create an endpoint handler that returns configuration details for a database engine.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to return the configuration for.	required

Returns:

Type	Description
Callable[[], Awaitable[BaseModel]]	An async function that returns database metadata.

Source code in auto_rest/handlers.py

def create_engine_handler(engine: DBEngine) -> Callable[[], Awaitable[PydanticModel]]:
    """Create an endpoint handler that returns configuration details for a database engine.

    Args:
        engine: Database engine to return the configuration for.

    Returns:
        An async function that returns database metadata.
    """

    interface = create_model("Meta",
        dialect=(str, engine.dialect.name),
        driver=(str, engine.dialect.driver),
        database=(str, engine.url.database),
    )

    async def meta_handler() -> interface:
        """Return metadata concerning the underlying application database."""

        return interface()

    return meta_handler

create_get_record_handler(engine, table)

Create a function for handling GET requests against a single record in the database.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to use when executing queries.	required
table	Table	The database table to query against.	required

Returns:

Type	Description
Callable[..., Awaitable[BaseModel]]	An async function that returns a single record from the given database table.

Source code in auto_rest/handlers.py

def create_get_record_handler(engine: DBEngine, table: Table) -> Callable[..., Awaitable[PydanticModel]]:
    """Create a function for handling GET requests against a single record in the database.

    Args:
        engine: Database engine to use when executing queries.
        table: The database table to query against.

    Returns:
        An async function that returns a single record from the given database table.
    """

    interface = create_interface(table)
    pk_interface = create_interface(table, pk_only=True, mode='required')

    async def get_record_handler(
        pk: pk_interface = Depends(),
        session: DBSession = Depends(create_session_iterator(engine)),
    ) -> interface:
        """Fetch a single record from the database."""

        query = select(table).filter_by(**pk.model_dump())
        result = await execute_session_query(session, query)
        record = get_record_or_404(result)
        return record

    return get_record_handler

create_list_records_handler(engine, table)

Create an endpoint handler that returns a list of records from a database table.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to use when executing queries.	required
table	Table	The database table to query against.	required

Returns:

Type	Description
Callable[..., Awaitable[list[BaseModel]]]	An async function that returns a list of records from the given database model.

Source code in auto_rest/handlers.py

def create_list_records_handler(engine: DBEngine, table: Table) -> Callable[..., Awaitable[list[PydanticModel]]]:
    """Create an endpoint handler that returns a list of records from a database table.

    Args:
        engine: Database engine to use when executing queries.
        table: The database table to query against.

    Returns:
        An async function that returns a list of records from the given database model.
    """

    interface = create_interface(table)
    interface_opt = create_interface(table, mode="optional")
    columns = tuple(table.columns.keys())

    async def list_records_handler(
        response: Response,
        session: DBSession = Depends(create_session_iterator(engine)),
        _limit_: int = Query(0, ge=0, description="The maximum number of records to return."),
        _offset_: int = Query(0, ge=0, description="The starting index of the returned records."),
        _order_by_: Optional[Literal[*columns]] = Query(None, description="The field name to sort by."),
        _direction_: Literal["asc", "desc"] = Query("asc", description="Sort results in 'asc' or 'desc' order.")
    ) -> list[interface]:
        """Fetch a list of records from the database.

        URL query parameters are used to enable filtering, ordering, and paginating returned values.
        """

        response.headers["X-Pagination-Limit"] = str(_limit_)
        response.headers["X-Pagination-Offset"] = str(_offset_)
        response.headers["X-Order-By"] = str(_order_by_)
        response.headers["X-Order-Direction"] = str(_direction_)

        query = select(table)
        query = apply_pagination_params(query, _limit_, _offset_)
        query = apply_ordering_params(query, _order_by_, _direction_)

        result = await execute_session_query(session, query)
        return [row._mapping for row in result.all()]

    return list_records_handler

create_meta_router(engine, metadata, name, version)

Create an API router for returning database metadata.

Includes routes for retrieving the database driver, database schema, and application/schema version.

Parameters:

Name	Type	Description	Default
engine	DBEngine	The database engine used to facilitate database interactions.	required
metadata	MetaData	The metadata object containing the database schema.	required
name	str	The application name.	required
version	str	The application versionnumber.	required

Returns:

Type	Description
APIRouter	An APIRouter with a routes for retrieving application metadata.

Source code in auto_rest/routers.py

def create_meta_router(engine: DBEngine, metadata: MetaData, name: str, version: str) -> APIRouter:
    """Create an API router for returning database metadata.

    Includes routes for retrieving the database driver, database schema,
    and application/schema version.

    Args:
        engine: The database engine used to facilitate database interactions.
        metadata: The metadata object containing the database schema.
        name: The application name.
        version: The application versionnumber.

    Returns:
        An `APIRouter` with a routes for retrieving application metadata.
    """

    logger.debug("Creating metadata endpoints.")

    router = APIRouter()
    tags = ["Application Metadata"]

    router.add_api_route(
        path="/app",
        methods=["GET"],
        endpoint=create_about_handler(name, version),
        summary="Fetch application metadata.",
        tags=tags
    )

    router.add_api_route(
        path="/engine",
        methods=["GET"],
        endpoint=create_engine_handler(engine),
        summary="Fetch database metadata.",
        tags=tags
    )

    router.add_api_route(
        path="/schema",
        methods=["GET"],
        endpoint=create_schema_handler(metadata),
        summary="Fetch the database schema.",
        tags=tags
    )

    return router

create_patch_record_handler(engine, table)

Create a function for handling PATCH requests against a record in the database.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to use when executing queries.	required
table	Table	The database table to query against.	required

Returns:

Type	Description
Callable[..., Awaitable[BaseModel]]	An async function that handles record updates.

Source code in auto_rest/handlers.py

def create_patch_record_handler(engine: DBEngine, table: Table) -> Callable[..., Awaitable[PydanticModel]]:
    """Create a function for handling PATCH requests against a record in the database.

    Args:
        engine: Database engine to use when executing queries.
        table: The database table to query against.

    Returns:
        An async function that handles record updates.
    """

    interface = create_interface(table)
    pk_interface = create_interface(table, pk_only=True, mode='required')

    async def patch_record_handler(
        data: interface,
        pk: pk_interface = Depends(),
        session: DBSession = Depends(create_session_iterator(engine)),
    ) -> interface:
        """Update record values in the database with the provided data."""

        query = select(table).filter_by(**pk.model_dump())
        result = await execute_session_query(session, query)
        record = get_record_or_404(result)

        for key, value in data.model_dump(exclude_unset=True).items():
            setattr(record, key, value)

        await commit_session(session)
        return record

    return patch_record_handler

create_post_record_handler(engine, table)

Create a function for handling POST requests against a record in the database.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to use when executing queries.	required
table	Table	The database table to query against.	required

Returns:

Type	Description
Callable[..., Awaitable[BaseModel]]	An async function that handles record creation.

Source code in auto_rest/handlers.py

def create_post_record_handler(engine: DBEngine, table: Table) -> Callable[..., Awaitable[PydanticModel]]:
    """Create a function for handling POST requests against a record in the database.

    Args:
        engine: Database engine to use when executing queries.
        table: The database table to query against.

    Returns:
        An async function that handles record creation.
    """

    interface = create_interface(table)

    async def post_record_handler(
        data: interface,
        session: DBSession = Depends(create_session_iterator(engine)),
    ) -> None:
        """Create a new record in the database."""

        query = insert(table).values(**data.dict())
        await execute_session_query(session, query)
        await commit_session(session)

    return post_record_handler

create_put_record_handler(engine, table)

Create a function for handling PUT requests against a record in the database.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to use when executing queries.	required
table	Table	The database table to query against.	required

Returns:

Type	Description
Callable[..., Awaitable[BaseModel]]	An async function that handles record updates.

Source code in auto_rest/handlers.py

def create_put_record_handler(engine: DBEngine, table: Table) -> Callable[..., Awaitable[PydanticModel]]:
    """Create a function for handling PUT requests against a record in the database.

    Args:
        engine: Database engine to use when executing queries.
        table: The database table to query against.

    Returns:
        An async function that handles record updates.
    """

    interface = create_interface(table)
    opt_interface = create_interface(table, mode='optional')
    pk_interface = create_interface(table, pk_only=True, mode='required')

    async def put_record_handler(
        data: opt_interface,
        pk: pk_interface = Depends(),
        session: DBSession = Depends(create_session_iterator(engine)),
    ) -> interface:
        """Replace record values in the database with the provided data."""

        query = select(table).filter_by(**pk.model_dump())
        result = await execute_session_query(session, query)
        record = get_record_or_404(result)

        for key, value in data.model_dump().items():
            setattr(record, key, value)

        await commit_session(session)
        return interface.model_validate(record.__dict__)

    return put_record_handler

create_schema_handler(metadata)

Create an endpoint handler that returns the database schema.

Parameters:

Name	Type	Description	Default
metadata	MetaData	Metadata object containing the database schema.	required

Returns:

Type	Description
Callable[[], Awaitable[BaseModel]]	An async function that returns the database schema.

Source code in auto_rest/handlers.py

def create_schema_handler(metadata: MetaData) -> Callable[[], Awaitable[PydanticModel]]:
    """Create an endpoint handler that returns the database schema.

    Args:
        metadata: Metadata object containing the database schema.

    Returns:
        An async function that returns the database schema.
    """

    # Define Pydantic models for column, table, and schema level data
    column_interface = create_model("Column",
        type=(str, ...),
        nullable=(bool, ...),
        default=(str | None, None),
        primary_key=(bool, ...),
    )

    table_interface = create_model("Table", columns=(dict[str, column_interface], ...))
    schema_interface = create_model("Schema", tables=(dict[str, table_interface], ...))

    async def schema_handler() -> schema_interface:
        """Return metadata concerning the underlying application database."""

        return schema_interface(
            tables={table_name: table_interface(columns={
                column.name: column_interface(
                    type=str(column.type),
                    nullable=column.nullable,
                    default=str(column.default.arg) if column.default else None,
                    primary_key=column.primary_key
                )
                for column in table.columns
            }) for table_name, table in metadata.tables.items()}
        )

    return schema_handler

create_table_router(engine, table)

Create an API router with endpoint handlers for a given database table.

Parameters:

Name	Type	Description	Default
engine	DBEngine	The SQLAlchemy engine connected to the database.	required
table	Table	The database table to create API endpoints for.	required

Returns:

Type	Description
APIRouter	An APIRouter instance with routes for database operations on the table.

Source code in auto_rest/routers.py

def create_table_router(engine: DBEngine, table: Table) -> APIRouter:
    """Create an API router with endpoint handlers for a given database table.

    Args:
        engine: The SQLAlchemy engine connected to the database.
        table: The database table to create API endpoints for.

    Returns:
        An APIRouter instance with routes for database operations on the table.
    """

    logger.debug(f"Creating endpoints for table `{table.name}`.")
    router = APIRouter()

    # Construct path parameters from primary key columns
    pk_columns = sorted(column.name for column in table.primary_key.columns)
    path_params_url = "/".join(f"{{{col_name}}}" for col_name in pk_columns)

    # Add routes for operations against the table
    router.add_api_route(
        path="/",
        methods=["GET"],
        endpoint=create_list_records_handler(engine, table),
        status_code=status.HTTP_200_OK,
        summary="Fetch multiple records from the table.",
        tags=[table.name],
    )

    router.add_api_route(
        path="/",
        methods=["POST"],
        endpoint=create_post_record_handler(engine, table),
        status_code=status.HTTP_201_CREATED,
        summary="Create a new record.",
        tags=[table.name],
    )

    # Add route for read operations against individual records
    if pk_columns:
        router.add_api_route(
            path=f"/{path_params_url}/",
            methods=["GET"],
            endpoint=create_get_record_handler(engine, table),
            status_code=status.HTTP_200_OK,
            summary="Fetch a single record from the table.",
            tags=[table.name],
        )

        router.add_api_route(
            path=f"/{path_params_url}/",
            methods=["PUT"],
            endpoint=create_put_record_handler(engine, table),
            status_code=status.HTTP_200_OK,
            summary="Replace a single record in the table.",
            tags=[table.name],
        )

        router.add_api_route(
            path=f"/{path_params_url}/",
            methods=["PATCH"],
            endpoint=create_patch_record_handler(engine, table),
            status_code=status.HTTP_200_OK,
            summary="Update a single record in the table.",
            tags=[table.name],
        )

        router.add_api_route(
            path=f"/{path_params_url}/",
            methods=["DELETE"],
            endpoint=create_delete_record_handler(engine, table),
            status_code=status.HTTP_200_OK,
            summary="Delete a single record from the table.",
            tags=[table.name],
        )

    return router

create_welcome_handler()

Create an endpoint handler that returns an application welcome message.

Returns:

Type	Description
Callable[[], Awaitable[BaseModel]]	An async function that returns a welcome message.

Source code in auto_rest/handlers.py

def create_welcome_handler() -> Callable[[], Awaitable[PydanticModel]]:
    """Create an endpoint handler that returns an application welcome message.

    Returns:
        An async function that returns a welcome message.
    """

    interface = create_model("Welcome", message=(str, "Welcome to Auto-Rest!"))

    async def welcome_handler() -> interface:
        """Return an application welcome message."""

        return interface()

    return welcome_handler

create_welcome_router()

Create an API router for returning a welcome message.

Returns:

Type	Description
APIRouter	An APIRouter with a single route for retrieving a welcome message.

Source code in auto_rest/routers.py

def create_welcome_router() -> APIRouter:
    """Create an API router for returning a welcome message.

    Returns:
        An `APIRouter` with a single route for retrieving a welcome message.
    """

    logger.debug("Creating welcome endpoint.")

    router = APIRouter()
    router.add_api_route(
        path="/",
        methods=["GET"],
        endpoint=create_welcome_handler(),
        include_in_schema=False
    )

    return router