Database

SessionManager

genai_monitor.db.config.SessionManager

SessionManager(database_url: str = DEFAULT_DATABASE_URL)

Manages the database engine and provides session management.

Source code in src/genai_monitor/db/config.py

def __init__(self, database_url: str = DEFAULT_DATABASE_URL):  # noqa: ANN204,D107
    self.initialize(database_url=database_url)

initialize

initialize(
    database_url: str = DEFAULT_DATABASE_URL,
) -> SessionManager

Initializes the database engine and session factory. This should be called once on application start.

PARAMETER	DESCRIPTION
database_url	The URL for the database connection. Defaults to sqlite:///genai_eval.db. TYPE: str DEFAULT: DEFAULT_DATABASE_URL

RETURNS	DESCRIPTION
SessionManager	The SessionManager object.

Source code in src/genai_monitor/db/config.py

def initialize(self, database_url: str = DEFAULT_DATABASE_URL) -> "SessionManager":
    """Initializes the database engine and session factory. This should be called once on application start.

    Args:
        database_url: The URL for the database connection. Defaults to sqlite:///genai_eval.db.

    Returns:
        The `SessionManager` object.
    """
    if self._engine is None:
        self._engine = create_engine(database_url)
        self._session_factory = sessionmaker(bind=self._engine, expire_on_commit=False)
        BaseModel.metadata.create_all(bind=self._engine)

    return self

session_scope

session_scope() -> Generator[Session, None, None]

Provide a transactional scope around a series of operations. Commits or rolls back on error.

RAISES	DESCRIPTION
Exception	on any error during database transaction
ConnectionError	if the database connection has not been initialized yet

YIELDS	DESCRIPTION
Session	The database connection session

Source code in src/genai_monitor/db/config.py

@contextmanager
def session_scope(self) -> Generator[Session, None, None]:
    """Provide a transactional scope around a series of operations. Commits or rolls back on error.

    Raises:
        Exception: on any error during database transaction
        ConnectionError: if the database connection has not been initialized yet

    Yields:
        The database connection session
    """
    if self._session_factory is None:
        raise ConnectionError(
            "The connection to the database must be initialized through, SessionManager.initialize()"
        )

    session = self._session_factory(expire_on_commit=False)
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

Database Manager

genai_monitor.db.manager.DBManager

A database manager class.

Class provides basic operations for saving, updating and searching records in the database using SQLAlchemy ORM models.

session_manager class-attribute instance-attribute

session_manager: Optional[SessionManager] = None

save

save(instance: BaseModel) -> BaseModel

Saves an instance of a model to the database.

PARAMETER	DESCRIPTION
instance	The ORM model instance to be saved. TYPE: BaseModel

RETURNS	DESCRIPTION
BaseModel	The saved ORM model instance.

Source code in src/genai_monitor/db/manager.py

def save(self, instance: BaseModel) -> BaseModel:
    """Saves an instance of a model to the database.

    Args:
        instance: The ORM model instance to be saved.

    Returns:
        The saved ORM model instance.
    """
    with self.session_manager.session_scope() as session:
        session.add(instance)
        session.commit()
        self._eager_load_instance_relations(instance=instance)
        return instance

search

search(
    model: Type[BaseModel],
    filters: Optional[Dict[str, Any]] = None,
) -> Sequence[Row]

Searches for records in the database that match the given filters.

PARAMETER	DESCRIPTION
model	The ORM model class representing the database table to search. TYPE: Type[BaseModel]
filters	Dictionary of filter criteria to locate specific records. TYPE: Optional[Dict[str, Any]] DEFAULT: None

RETURNS	DESCRIPTION
Sequence[Row]	A sequence of rows matching the filter criteria.

Source code in src/genai_monitor/db/manager.py

def search(self, model: Type[BaseModel], filters: Optional[Dict[str, Any]] = None) -> Sequence[Row]:
    """Searches for records in the database that match the given filters.

    Args:
        model: The ORM model class representing the database table to search.
        filters: Dictionary of filter criteria to locate specific records.

    Returns:
        A sequence of rows matching the filter criteria.
    """
    with self.session_manager.session_scope() as session:
        query = session.query(model)
        if filters:
            query = query.filter_by(**filters)
        result = query.all()
        for instance in result:
            self._eager_load_instance_relations(instance)
        return result

update

update(
    instance: Optional[BaseModel] = None,
    model: Optional[Type[BaseModel]] = None,
    filters: Optional[Dict[str, Any]] = None,
    values: Optional[Dict[str, Any]] = None,
) -> Union[BaseModel, Sequence[Row]]

Updates records in the database.

If an instance is provided, it will be updated directly. Otherwise, model and filter criteria are used to locate records for updating.

PARAMETER	DESCRIPTION
instance	An existing ORM model instance to update. TYPE: Optional[BaseModel] DEFAULT: None
model	The ORM model class representing the table to update. TYPE: Optional[Type[BaseModel]] DEFAULT: None
filters	Dictionary of filter criteria to locate records to update. TYPE: Optional[Dict[str, Any]] DEFAULT: None
values	Dictionary of field names and values to update. TYPE: Optional[Dict[str, Any]] DEFAULT: None

RETURNS	DESCRIPTION
Union[BaseModel, Sequence[Row]]	The updated instance if instance was provided or the updated DB rows.

RAISES	DESCRIPTION
ValueError	If neither instance nor model is provided.

Source code in src/genai_monitor/db/manager.py

def update(
    self,
    instance: Optional[BaseModel] = None,
    model: Optional[Type[BaseModel]] = None,
    filters: Optional[Dict[str, Any]] = None,
    values: Optional[Dict[str, Any]] = None,
) -> Union[BaseModel, Sequence[Row]]:
    """Updates records in the database.

    If an instance is provided, it will be updated
    directly. Otherwise, model and filter criteria are used to locate records for updating.

    Args:
        instance: An existing ORM model instance to update.
        model: The ORM model class representing the table to update.
        filters: Dictionary of filter criteria to locate records to update.
        values: Dictionary of field names and values to update.

    Returns:
       The updated instance if `instance` was provided or the updated DB rows.

    Raises:
        ValueError: If neither `instance` nor `model` is provided.
    """
    if (instance is None) and (model is None):
        raise ValueError(
            "To update DB resource provide either an instance of the ORM class or an ORM model with filters."
        )

    if (instance is not None) and (model is not None):
        logger.warning(
            "Provided both instance of an ORM class and the ORM model for update, instance will be used."
        )

    with self.session_manager.session_scope() as session:
        if instance is not None:
            for field_name, field_value in values.items():  # type: ignore
                setattr(instance, field_name, field_value)
            session.add(instance)
            session.commit()
            return instance

        query = session.query(model).filter_by(**filters)
        query_results = query.all()
        for result in query_results:
            for field_name, field_value in values.items():  # type: ignore
                setattr(result, field_name, field_value)
        session.commit()
        for result in query_results:
            self._eager_load_instance_relations(result)
        return query_results

join_search

join_search(
    target_model: Type[BaseModel],
    join_model: Type[BaseModel],
    on_condition: Any,
    target_filters: Optional[Dict[str, Any]] = None,
    join_filters: Optional[Dict[str, Any]] = None,
) -> Sequence[Row]

Performs a join search between two models based on a join condition and optional filters.

PARAMETER	DESCRIPTION
target_model	The ORM model class representing the primary table to search. TYPE: Type[BaseModel]
join_model	The ORM model class representing the table to join with. TYPE: Type[BaseModel]
on_condition	The join condition specifying how to link the two tables. TYPE: Any
target_filters	Dictionary of filter criteria for the target model. TYPE: Optional[Dict[str, Any]] DEFAULT: None
join_filters	Dictionary of filter criteria for the join model. TYPE: Optional[Dict[str, Any]] DEFAULT: None

RETURNS	DESCRIPTION
Sequence[Row]	A sequence of rows resulting from the join search, filtered as specified.

Source code in src/genai_monitor/db/manager.py

def join_search(
    self,
    target_model: Type[BaseModel],
    join_model: Type[BaseModel],
    on_condition: Any,
    target_filters: Optional[Dict[str, Any]] = None,
    join_filters: Optional[Dict[str, Any]] = None,
) -> Sequence[Row]:
    """Performs a join search between two models based on a join condition and optional filters.

    Args:
        target_model: The ORM model class representing the primary table to search.
        join_model: The ORM model class representing the table to join with.
        on_condition: The join condition specifying how to link the two tables.
        target_filters: Dictionary of filter criteria for the target model.
        join_filters: Dictionary of filter criteria for the join model.

    Returns:
        A sequence of rows resulting from the join search, filtered as specified.
    """
    with self.session_manager.session_scope() as session:
        query = session.query(target_model).join(join_model, on_condition)

        if target_filters:
            target_conditions = [getattr(target_model, key) == value for key, value in target_filters.items()]
            query = query.filter(and_(*target_conditions))

        if join_filters:
            join_conditions = [getattr(join_model, key) == value for key, value in join_filters.items()]
            query = query.filter(and_(*join_conditions))

        results = query.all()
        return results

Tables

genai_monitor.db.schemas.tables.ConditioningTable

Bases: BaseModel

Database table representing the value of conditioning.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

type_id class-attribute instance-attribute

type_id: Mapped[Optional[str]] = mapped_column(
    ForeignKey("conditioning_type.id")
)

value class-attribute instance-attribute

value: Mapped[dict] = mapped_column(JSON)

hash class-attribute instance-attribute

hash: Mapped[str] = mapped_column()

value_metadata class-attribute instance-attribute

value_metadata: Mapped[Optional[dict]] = mapped_column(JSON)

samples class-attribute instance-attribute

samples: Mapped[List[SampleTable]] = relationship(
    back_populates="conditioning"
)

genai_monitor.db.schemas.tables.ConditioningTypeTable

Bases: BaseModel

Database table representing the type of conditioning.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

type class-attribute instance-attribute

type: Mapped[str] = mapped_column()

genai_monitor.db.schemas.tables.SampleTable

Bases: BaseModel

Database table representing the sample - an atomic unit of data in the system.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

model_id class-attribute instance-attribute

model_id: Mapped[Optional[int]] = mapped_column(
    ForeignKey("model.id")
)

conditioning_id class-attribute instance-attribute

conditioning_id: Mapped[Optional[int]] = mapped_column(
    ForeignKey("conditioning.id")
)

user_id class-attribute instance-attribute

user_id: Mapped[Optional[int]] = mapped_column(
    ForeignKey("user.id")
)

name class-attribute instance-attribute

name: Mapped[Optional[str]] = mapped_column()

hash class-attribute instance-attribute

hash: Mapped[str] = mapped_column()

meta class-attribute instance-attribute

meta: Mapped[Optional[dict]] = mapped_column(JSON)

generation_id class-attribute instance-attribute

generation_id: Mapped[Optional[int]] = mapped_column()

status class-attribute instance-attribute

status: Mapped[str] = mapped_column()

version class-attribute instance-attribute

version: Mapped[str] = mapped_column()

conditioning class-attribute instance-attribute

conditioning: Mapped[ConditioningTable] = relationship(
    back_populates="samples"
)

user class-attribute instance-attribute

user: Mapped[UserTable] = relationship(
    back_populates="samples"
)

artifacts class-attribute instance-attribute

artifacts: Mapped[List[ArtifactTable]] = relationship(
    back_populates="sample"
)

genai_monitor.db.schemas.tables.ModelTable

Bases: BaseModel

Database table representing the generative model.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

hash class-attribute instance-attribute

hash: Mapped[Optional[str]] = mapped_column()

model_class class-attribute instance-attribute

model_class: Mapped[Optional[str]] = mapped_column()

checkpoint_location class-attribute instance-attribute

checkpoint_location: Mapped[Optional[str]] = mapped_column()

checkpoint_metadata class-attribute instance-attribute

checkpoint_metadata: Mapped[Optional[dict]] = mapped_column(
    JSON
)

training_step class-attribute instance-attribute

training_step: Mapped[Optional[int]] = mapped_column()

model_metadata class-attribute instance-attribute

model_metadata: Mapped[Optional[dict]] = mapped_column(JSON)

genai_monitor.db.schemas.tables.ConfigurationTable

Bases: BaseModel

Database table representing system configuration.

Default values are set at the database level to ensure consistency across all instances.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

key class-attribute instance-attribute

key: Mapped[str] = mapped_column(unique=True)

value class-attribute instance-attribute

value: Mapped[str] = mapped_column()

description class-attribute instance-attribute

description: Mapped[Optional[str]] = mapped_column()

updated_at class-attribute instance-attribute

updated_at: Mapped[str] = mapped_column()

is_default class-attribute instance-attribute

is_default: Mapped[bool] = mapped_column(
    server_default=text("0")
)

default_value class-attribute instance-attribute

default_value: Mapped[Optional[str]] = mapped_column()

genai_monitor.db.schemas.tables.UserTable

Bases: BaseModel

Database table representing the users.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

name class-attribute instance-attribute

name: Mapped[str] = mapped_column()

hash class-attribute instance-attribute

hash: Mapped[str] = mapped_column()

samples class-attribute instance-attribute

samples: Mapped[List[SampleTable]] = relationship(
    back_populates="user"
)

genai_monitor.db.schemas.tables.ArtifactTable

Bases: BaseModel

Database table representing the artifact - an atomic unit of data in the system.

id class-attribute instance-attribute

id: Mapped[int] = mapped_column(
    primary_key=True, autoincrement=True
)

name class-attribute instance-attribute

name: Mapped[str] = mapped_column()

value class-attribute instance-attribute

value: Mapped[Optional[str]] = mapped_column()

hash class-attribute instance-attribute

hash: Mapped[str] = mapped_column()

sample_id class-attribute instance-attribute

sample_id: Mapped[int] = mapped_column(
    ForeignKey("sample.id")
)

sample class-attribute instance-attribute

sample: Mapped[SampleTable] = relationship(
    back_populates="artifacts"
)