auto_rest.models

The models module facilitates communication with relational databases via dynamically generated object relational mappers (ORMs). Building on the popular SQLAlchemy package, it natively supports multiple Database Management Systems (DBMS) without requiring custom configuration or setup.

Example: Mapping Database Metadata

Utility functions are provided for connecting to the database and mapping the underlying schema.

connection_args = dict(...)
db_url = create_db_url(**connection_args)
db_conn = create_db_engine(db_url)
db_meta = create_db_metadata(db_conn)

Support for asynchronous operations is automatically determined based on the chosen database. If the driver supports asynchronous operations, the connection and session handling are configured accordingly.

Developer Note

When working with database objects, the returned object type may vary depending on whether the underlying driver is synchronous or asynchronous. Of particular note are database engines (Engine / AsyncEngine) and sessions (Session / AsyncSession).

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_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

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 {}