Utils

Auto mode configuration

genai_monitor.utils.auto_mode_configuration.load_config

load_config(
    db_url: str,
    default_settings: Dict[
        str, dict[str, str]
    ] = DEFAULT_SETTINGS,
) -> Config

Load configuration from database.

PARAMETER	DESCRIPTION
db_url	The database URL. TYPE: str
default_settings	The default settings to use if no configuration is found. TYPE: Dict[str, dict[str, str]] DEFAULT: DEFAULT_SETTINGS

RETURNS	DESCRIPTION
Config	The configuration as a dictionary.

RAISES	DESCRIPTION
RuntimeError	If the configuration could not be loaded.

Source code in src/genai_monitor/utils/auto_mode_configuration.py

def load_config(db_url: str, default_settings: Dict[str, dict[str, str]] = DEFAULT_SETTINGS) -> Config:
    """Load configuration from database.

    Args:
        db_url: The database URL.
        default_settings: The default settings to use if no configuration is found.

    Returns:
        The configuration as a dictionary.

    Raises:
        RuntimeError: If the configuration could not be loaded.
    """
    logger.info(f"Using database at: {db_url}")

    # Initialize database with the url
    session_manager = init_db(database_url=db_url)

    try:
        with session_manager.session_scope() as session:
            # Initialize default configuration if needed
            _init_default_config(session, default_settings)

            # Load configuration, preferring non-default values over defaults
            config_entries = (
                session.query(ConfigurationTable).order_by(ConfigurationTable.is_default).all()  # Non-defaults first
            )

            # Build config dict, letting non-default values override defaults
            settings_dict = {}
            for entry in config_entries:
                if entry.key not in settings_dict:  # Only set if not already set by a non-default
                    settings_dict[entry.key] = entry.value

            return Config(
                persistency={  # type: ignore
                    "enabled": settings_dict["persistency.enabled"].lower() == "true",
                    "path": settings_dict["persistency.path"],
                },
                db={"url": settings_dict["db.url"]},  # type: ignore
                version=settings_dict["version"],
            )

    except SQLAlchemyError as e:
        logger.error(f"Failed to load configuration from database: {e}")
        raise RuntimeError("Failed to load configuration") from e

Data hashing

genai_monitor.utils.model_hashing.default_model_hashing_function

default_model_hashing_function(
    model: Any,
) -> str | Literal[UNKNOWN_MODEL_HASH]

Default model hashing function.

PARAMETER	DESCRIPTION
model	The model to hash. TYPE: Any

RETURNS	DESCRIPTION
str | Literal[UNKNOWN_MODEL_HASH]	str | Literal[UNKNOWN_MODEL_HASH]: The hash of the model.

Source code in src/genai_monitor/utils/model_hashing.py

def default_model_hashing_function(model: Any) -> str | Literal[UNKNOWN_MODEL_HASH]:  # type: ignore
    """Default model hashing function.

    Args:
        model: The model to hash.

    Returns:
        str | Literal[UNKNOWN_MODEL_HASH]: The hash of the model.
    """
    try:
        return get_component_hash(model)
    except Exception as e:
        logger.error(f"Failed to hash model: {e}")
        return UNKNOWN_MODEL_HASH

User registration

genai_monitor.utils.user_registration.register_user

register_user(
    db_manager: DBManager, runtime_manager: RuntimeManager
)

Register a new user if not already registered.

PARAMETER	DESCRIPTION
db_manager	The database manager. TYPE: DBManager
runtime_manager	The runtime manager. TYPE: RuntimeManager

Source code in src/genai_monitor/utils/user_registration.py

def register_user(db_manager: DBManager, runtime_manager: RuntimeManager):
    """Register a new user if not already registered.

    Args:
        db_manager: The database manager.
        runtime_manager: The runtime manager.
    """
    user_hash = generate_user_hash()

    existing_user = db_manager.search(
        UserTable,
        {"hash": user_hash},
    )

    if existing_user:
        runtime_manager.set_user_id(existing_user[0].id)
        logger.info(f"User {existing_user[0].name} found in the database.")
        return

    logger.info("User not found in the database. Registering new user.")
    username = getpass.getuser()
    user = User(name=username, hash=user_hash)
    user = db_manager.save(instance=user.to_orm())
    runtime_manager.set_user_id(user.id)
    logger.success(f"User {user.name} registered successfully.")