ovo.core.database.sql_db

Module Contents

Classes

SqlDBEngine

API

class ovo.core.database.sql_db.SqlDBEngine(db_url: str, verbose: bool = False, read_only: bool = False)

Bases: ovo.core.database.cache_clearing.CacheClearingEngine

init()

automigrate()

_create_session() → sqlalchemy.orm.Session

_check_updated(obj: ovo.core.database.models.Base)

check_read_only()

save(obj: ovo.core.database.models.Base)

save_all(objs: Sequence[ovo.core.database.models.Base])

remove(model: Type[ovo.core.database.base_db.T], *id_args, **kwargs)

save_value(model: Type[ovo.core.database.base_db.T], column: str, value, **kwargs)

select(model: Type[ovo.core.database.base_db.T], limit: int = None, order_by=None, **kwargs) → Sequence[ovo.core.database.base_db.T]

count(model: Type[ovo.core.database.base_db.T], **kwargs) → int

count_distinct(model: Type[ovo.core.database.base_db.T], field='id', group_by=None, **kwargs) → int | dict[Any, int]

_create_query(session, model: Type[ovo.core.database.base_db.T], columns=None, order_by=None, limit=None, id_args=None, **kwargs)

_create_order_by(model: Type[ovo.core.database.base_db.T], order_by)

_create_order_by_single(model: Type[ovo.core.database.base_db.T], order_by)

_create_filters(model: Type[ovo.core.database.base_db.T], kwargs)

get(model: Type[ovo.core.database.base_db.T], *id_args, **kwargs) → ovo.core.database.base_db.T

get_value(model: Type[ovo.core.database.base_db.T], column: str, *id_args, raw=False, **kwargs)

select_values(model: Type[ovo.core.database.base_db.T], column: str, order_by=None, **kwargs) → list

select_dict(model: Type[ovo.core.database.base_db.T], key_column: str, value_column: str, order_by=None, **kwargs) → dict

select_unique_values(model: Type[ovo.core.database.base_db.T], column: str, **kwargs) → set

select_dataframe(model: Type[ovo.core.database.base_db.T], index_col: str = 'id', order_by=None, limit='unset', **kwargs) → pandas.DataFrame

select_descriptor_values(descriptor_key: str, design_ids: list[str], descriptor_job_id: str | None = None) → pandas.Series

Select values of a single descriptor for multiple designs.

select_design_descriptors(design_id: str, descriptor_keys: list[str], descriptor_job_id: str | None = None) → pandas.Series

Select values of multiple descriptors for a single design.

select_wide_descriptor_table(design_ids: list[str], descriptor_keys: list[str], descriptor_job_id: str | None = None, **kwargs) → pandas.DataFrame

get_design_accepted_values(design_ids: list[str])

get_or_create_labeling(label: str, username: str, explanation: str | None = None) → ovo.core.database.models.Labeling

Get an existing labeling or create a new one if it doesn’t exist.

add_label(label: str, design_ids: list[str], username: str, explanation: str | None = None)

Add a label to multiple designs.

Args: label: The label to add design_ids: List of design IDs to add the label to username: The username of the person adding the label explanation: Optional explanation for adding the label (for audit/logging purposes)

remove_label(label: str, design_ids: list[str], **kwargs)

Remove a label from multiple designs.

Args: label: The label to remove design_ids: List of design IDs to remove the label from

remove_designs_labeling(labeling_id: str, design_ids: list[str])

Remove a specific design labeling by its ID.

Args: labeling_id: The ID of the design labeling to remove design_ids: List of design IDs to remove the labeling from

get_designs_with_all_labels(label_names: list[str], design_ids: list[str] = None) → list[str]

Get design IDs that have ALL of the specified labels using a single optimized query.

get_designs_with_any_labels(label_names: list[str], design_ids: list[str] = None) → list[str]

Get design IDs that have ANY of the specified labels using a single optimized query.

get_labelings_for_design(design_id: str) → list[ovo.core.database.models.Labeling]

Get all labelings for a specific design using a single optimized query.

get_available_labels_for_design_ids(design_ids: list[str]) → list[str]

Get unique labels available for the given design IDs.

get_available_labels_for_pool_ids(pool_ids: list[str], **design_filters) → list[str]

Get unique labels available for designs in the given pool IDs.

Args: pool_ids: List of pool IDs to get labels for **design_filters: Additional filters to apply to Design model (e.g., accepted=True)

Returns: Sorted list of unique label strings available for designs in the specified pools