db

Version agnostic database API module.

This module serves as the import point for functions defined for various versions of the database API. Functions can be imported from this module instead of worrying about handling imports from the various API sub-packages.

import_function(func_name, api_version) cached

Import a database function from the appropriate API version.

Parameters:

Name	Type	Description	Default
func_name	str	The name of the function to import.	required
api_version	int	The API version. Currently either 0x000001 or 0x000002.	required

Returns:

Name	Type	Description
func	Callable	The requested function.

Raises:

Type	Description
DatabaseError	Raised if the api_version is not supported.
AttributeError	Raised if the function requested does not exist.

Source code in lute/io/db.py

@lru_cache
def import_function(func_name: str, api_version: int) -> Callable:
    """Import a database function from the appropriate API version.

    Args:
        func_name (str): The name of the function to import.

        api_version (int): The API version. Currently either 0x000001 or 0x000002.

    Returns:
        func (Callable): The requested function.

    Raises:
        DatabaseError: Raised if the api_version is not supported.

        AttributeError: Raised if the function requested does not exist.
    """
    if api_version not in (1, 2):
        raise common_sqlite.DatabaseError(
            "Unrecognized database specification version! Set LUTE_DB_SPEC_VERSION appropriately! "
            "Supported versions: 0x000001 and 0x000002"
        )
    api_mod: ModuleType = importlib.import_module(f"lute.io._db.v{api_version}.api")
    try:
        func: Callable = getattr(api_mod, func_name)
    except AttributeError:
        logging.error(
            f"Attempting to retrieve database API non-existent function {func_name}!"
        )
        raise
    return func

lazy_import(func_name, api_version)

Return a lazily loaded version of the function.

Source code in lute/io/db.py

def lazy_import(func_name: str, api_version: int) -> Callable:
    """Return a lazily loaded version of the function."""

    def wrapper(*args, **kwargs):
        func: Callable = import_function(func_name=func_name, api_version=api_version)
        return func(*args, **kwargs)

    return wrapper