auto_rest.handlers

An endpoint handler is a function designed to process incoming HTTP requests for single API endpoint. In auto_rest, handlers are created dynamically using a factory pattern. This approach allows handler logic to be customized and reused across multiple endpoints.

Example: Creating a Handler

New endpoint handlers are created dynamically using factory methods.

welcome_handler = create_welcome_handler()

Handler functions are defined as asynchronous coroutines. This provides improved performance when handling large numbers of incoming requests.

Example: Async Handlers

Python requires asynchronous coroutines to be run from an asynchronous context. In the following example, this is achieved using asyncio.run.

import asyncio

return_value = asyncio.run(welcome_handler())

Handlers are specifically designed to integrate with the FastAPI framework, including support for FastAPI's type hinting and data validation capabilities. This makes it easy to incorporate handlers into a FastAPI application.

Example: Adding a Handler to an Application

Use the add_api_route method to dynamically add handler functions to an existing application instance.

app = FastAPI(...)

handler = create_welcome_handler()
app.add_api_route("/", handler, methods=["GET"], ...)

Developer Note

FastAPI internally performs post-processing on values returned by endpoint handlers before sending them in an HTTP response. For this reason, handlers should always be tested within the context of a FastAPI application.

apply_ordering_params(query, order_by=None, direction='asc')

Apply ordering to a database query.

Returns a copy of the provided query with ordering parameters applied.

Parameters:

Name	Type	Description	Default
query	Select	The database query to apply parameters to.	required
order_by	str | None	The name of the column to order by.	None
direction	Literal['desc', 'asc']	The direction to order by (defaults to "asc").	'asc'

Returns:

Type	Description
Select	A copy of the query modified to return ordered values.

Source code in auto_rest/queries.py

def apply_ordering_params(
    query: Select,
    order_by: str | None = None,
    direction: Literal["desc", "asc"] = "asc"
) -> Select:
    """Apply ordering to a database query.

    Returns a copy of the provided query with ordering parameters applied.

    Args:
        query: The database query to apply parameters to.
        order_by: The name of the column to order by.
        direction: The direction to order by (defaults to "asc").

    Returns:
        A copy of the query modified to return ordered values.
    """

    if order_by is None:
        return query

    if order_by not in query.columns:
        raise ValueError(f"Invalid column name: {order_by}")

    # Default to ascending order for an invalid ordering direction
    if direction == "desc":
        return query.order_by(desc(order_by))

    elif direction == "asc":
        return query.order_by(asc(order_by))

    raise ValueError(f"Invalid direction, use 'asc' or 'desc': {direction}")

apply_pagination_params(query, limit=0, offset=0)

Apply pagination to a database query.

Returns a copy of the provided query with offset and limit parameters applied.

Parameters:

Name	Type	Description	Default
query	Select	The database query to apply parameters to.	required
limit	int	The number of results to return.	0
offset	int	The offset to start with.	0

Returns:

Type	Description
Select	A copy of the query modified to only return the paginated values.

Source code in auto_rest/queries.py

def apply_pagination_params(query: Select, limit: int = 0, offset: int = 0) -> Select:
    """Apply pagination to a database query.

    Returns a copy of the provided query with offset and limit parameters applied.

    Args:
        query: The database query to apply parameters to.
        limit: The number of results to return.
        offset: The offset to start with.

    Returns:
        A copy of the query modified to only return the paginated values.
    """

    if offset < 0 or limit < 0:
        raise ValueError("Pagination parameters cannot be negative")

    # Do not apply pagination if not requested
    if limit == 0:
        return query

    return query.offset(offset or 0).limit(limit)

commit_session(session) async

Commit a SQLAlchemy session.

Supports synchronous and asynchronous sessions.

Parameters:

Name	Type	Description	Default
session	DBSession	The session to commit.	required

Source code in auto_rest/queries.py

async def commit_session(session: DBSession) -> None:
    """Commit a SQLAlchemy session.

    Supports synchronous and asynchronous sessions.

    Args:
        session: The session to commit.
    """

    if isinstance(session, AsyncSession):
        await session.commit()

    else:
        session.commit()

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_db_engine(url, **kwargs)

Initialize a new database engine.

Instantiates and returns an Engine or AsyncEngine instance depending on whether the database URL uses a driver with support for async operations.

Parameters:

Name	Type	Description	Default
url	URL	A fully qualified database URL.	required
**kwargs	dict[str:any]	Keyword arguments passed to create_engine.	{}

Returns:

Type	Description
DBEngine	A SQLAlchemy Engine or AsyncEngine instance.

Source code in auto_rest/models.py

def create_db_engine(url: URL, **kwargs: dict[str: any]) -> DBEngine:
    """Initialize a new database engine.

    Instantiates and returns an `Engine` or `AsyncEngine` instance depending
    on whether the database URL uses a driver with support for async operations.

    Args:
        url: A fully qualified database URL.
        **kwargs: Keyword arguments passed to `create_engine`.

    Returns:
        A SQLAlchemy `Engine` or `AsyncEngine` instance.
    """

    logger.debug(f"Building database engine for {url}.")

    if url.get_dialect().is_async:
        engine = create_async_engine(url, **kwargs)
        logger.debug("Asynchronous connection established.")
        return engine

    else:
        engine = create_engine(url, **kwargs)
        logger.debug("Synchronous connection established.")
        return engine

create_db_metadata(engine)

Create and reflect metadata for the database connection.

Parameters:

Name	Type	Description	Default
engine	DBEngine	The database engine to use for reflection.	required

Returns:

Type	Description
MetaData	A MetaData object reflecting the database schema.

Source code in auto_rest/models.py

def create_db_metadata(engine: DBEngine) -> MetaData:
    """Create and reflect metadata for the database connection.

    Args:
        engine: The database engine to use for reflection.

    Returns:
        A MetaData object reflecting the database schema.
    """

    logger.debug("Loading database metadata.")
    metadata = MetaData()

    if isinstance(engine, AsyncEngine):
        asyncio.run(_async_reflect_metadata(engine, metadata))

    else:
        metadata.reflect(bind=engine, views=True)

    return metadata

create_db_url(driver, database, host=None, port=None, username=None, password=None)

Create a database URL from the provided parameters.

Parameters:

Name	Type	Description	Default
driver	str	The SQLAlchemy-compatible database driver.	required
database	str	The database name or file path (for SQLite).	required
host	str | None	The database server hostname or IP address.	None
port	int | None	The database server port number.	None
username	str | None	The username for authentication.	None
password	str | None	The password for the database user.	None

Returns:

Type	Description
URL	A fully qualified database URL.

Source code in auto_rest/models.py

def create_db_url(
    driver: str,
    database: str,
    host: str | None = None,
    port: int | None = None,
    username: str | None = None,
    password: str | None = None,
) -> URL:
    """Create a database URL from the provided parameters.

    Args:
        driver: The SQLAlchemy-compatible database driver.
        database: The database name or file path (for SQLite).
        host: The database server hostname or IP address.
        port: The database server port number.
        username: The username for authentication.
        password: The password for the database user.

    Returns:
        A fully qualified database URL.
    """

    # Handle special case where SQLite uses file paths.
    if "sqlite" in driver:
        path = Path(database).resolve()
        url = URL.create(drivername=driver, database=str(path))

    else:
        url = URL.create(
            drivername=driver,
            username=username,
            password=password,
            host=host,
            port=port,
            database=database,
        )

    logger.debug(f"Resolved URL: {url}")
    return url

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_interface(table, pk_only=False, mode='default')

Create a Pydantic interface for a SQLAlchemy model where all fields are required.

Modes

default: Values are marked as (not)required based on the column schema. required: Values are always marked required. required: Values are always marked optional.

Parameters:

Name	Type	Description	Default
table	Table	The SQLAlchemy table to create an interface for.	required
pk_only	bool	If True, only include primary key columns.	False
mode	MODE_TYPE	Whether to force fields to all be optional or required.	'default'

Returns:

Type	Description
type[BaseModel]	A dynamically generated Pydantic model with all fields required.

Source code in auto_rest/interfaces.py

def create_interface(table: Table, pk_only: bool = False, mode: MODE_TYPE = "default") -> type[PydanticModel]:
    """Create a Pydantic interface for a SQLAlchemy model where all fields are required.

    Modes:
        default: Values are marked as (not)required based on the column schema.
        required: Values are always marked required.
        required: Values are always marked optional.

    Args:
        table: The SQLAlchemy table to create an interface for.
        pk_only: If True, only include primary key columns.
        mode: Whether to force fields to all be optional or required.

    Returns:
        A dynamically generated Pydantic model with all fields required.
    """

    # Map field names to the column type and default value.
    fields = {
        col.name: create_field_definition(col, mode) for col in iter_columns(table, pk_only)
    }

    # Create a unique name for the interface
    name = f"{table.name}-{mode.title()}"
    if pk_only:
        name += '-PK'

    return create_model(name, __config__={'arbitrary_types_allowed': True}, **fields)

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_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_session_iterator(engine)

Create a generator for database sessions.

Returns a synchronous or asynchronous function depending on whether the database engine supports async operations. The type of session returned also depends on the underlying database engine, and will either be a Session or AsyncSession instance.

Parameters:

Name	Type	Description	Default
engine	DBEngine	Database engine to use when generating new sessions.	required

Returns:

Type	Description
Callable[[], DBSession]	A function that yields a single new database session.

Source code in auto_rest/models.py

def create_session_iterator(engine: DBEngine) -> Callable[[], DBSession]:
    """Create a generator for database sessions.

    Returns a synchronous or asynchronous function depending on whether
    the database engine supports async operations. The type of session
    returned also depends on the underlying database engine, and will
    either be a `Session` or `AsyncSession` instance.

    Args:
        engine: Database engine to use when generating new sessions.

    Returns:
        A function that yields a single new database session.
    """

    if isinstance(engine, AsyncEngine):
        async def session_iterator() -> AsyncSession:
            async with AsyncSession(bind=engine, autocommit=False, autoflush=True) as session:
                yield session

    else:
        def session_iterator() -> Session:
            with Session(bind=engine, autocommit=False, autoflush=True) as session:
                yield session

    return session_iterator

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

delete_session_record(session, record) async

Delete a record from the database using an existing session.

Does not automatically commit the session. Supports synchronous and asynchronous sessions.

Parameters:

Name	Type	Description	Default
session	DBSession	The session to use for deletion.	required
record	Result	The record to be deleted.	required

Source code in auto_rest/queries.py

async def delete_session_record(session: DBSession, record: Result) -> None:
    """Delete a record from the database using an existing session.

    Does not automatically commit the session.
    Supports synchronous and asynchronous sessions.

    Args:
        session: The session to use for deletion.
        record: The record to be deleted.
    """

    if isinstance(session, AsyncSession):
        await session.delete(record)

    else:
        session.delete(record)

execute_session_query(session, query) async

Execute a query in the given session and return the result.

Supports synchronous and asynchronous sessions.

Parameters:

Name	Type	Description	Default
session	DBSession	The SQLAlchemy session to use for executing the query.	required
query	Executable	The query to be executed.	required

Returns:

Type	Description
Result	The result of the executed query.

Source code in auto_rest/queries.py

async def execute_session_query(session: DBSession, query: Executable) -> Result:
    """Execute a query in the given session and return the result.

    Supports synchronous and asynchronous sessions.

    Args:
        session: The SQLAlchemy session to use for executing the query.
        query: The query to be executed.

    Returns:
        The result of the executed query.
    """

    if isinstance(session, AsyncSession):
        return await session.execute(query)

    return session.execute(query)

get_record_or_404(result)

Retrieve a scalar record from a query result or raise a 404 error.

Parameters:

Name	Type	Description	Default
result	Result	The query result to extract the scalar record from.	required

Returns:

Type	Description
any	The scalar record if it exists.

Raises:

Type	Description
HTTPException	If the record is not found.

Source code in auto_rest/queries.py

def get_record_or_404(result: Result) -> any:
    """Retrieve a scalar record from a query result or raise a 404 error.

    Args:
        result: The query result to extract the scalar record from.

    Returns:
        The scalar record if it exists.

    Raises:
        HTTPException: If the record is not found.
    """

    if record := result.fetchone():
        return record

    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Record not found")

parse_db_settings(path)

Parse engine configuration settings from a given file path.

Parameters:

Name	Type	Description	Default
path	Path | None	Path to the configuration file.	required

Returns:

Type	Description
dict[str, any]	Engine configuration settings.

Source code in auto_rest/models.py

def parse_db_settings(path: Path | None) -> dict[str, any]:
    """Parse engine configuration settings from a given file path.

    Args:
        path: Path to the configuration file.

    Returns:
        Engine configuration settings.
    """

    if path is not None:
        logger.debug(f"Parsing engine configuration from {path}.")
        return yaml.safe_load(path.read_text()) or dict()

    logger.debug("No connection file specified.")
    return {}