Database utilities

class duetector.db.SessionManager(config: Dict[str, Any] | None = None, *args, **kwargs)

A wrapper for sqlalchemy session

Special config:

table_prefix (str): prefix for all table names engine (Dict[str, Any]): config for sqlalchemy.create_engine

Example:

from duetector.db import SessionManager
from duetector.collectors.models import Tracking
sessionmanager = SessionManager()
t = Tracking(
    tracer="t",
)
m = sessionmanager.get_tracking_model(t.tracer, "id")

with sessionmanager.begin() as session:
    session.add(m(**t.model_dump(exclude=["tracer"])))
    session.commit()

assert sessionmanager.inspect_all_tables() == [
    sessionmanager.get_table_names("t", "id")
]
assert sessionmanager.inspect_all_tables("not-exist") == []

begin() → Generator[Session, None, None]

Get a sqlalchemy session.

Example:

with session_manager.begin() as session:
    session.add(...)

config_scope: str | None = 'db'

property debug

default_config = {'engine': {'url': 'sqlite:///:memory:'}, 'table_prefix': 'duetector_tracking'}

property engine

A sqlalchemy engine

property engine_config: Dict[str, Any]

Config for sqlalchemy.create_engine

get_all_models() → Dict[str, type]

get_table_names(tracer: str = 'unknown', collector_id: str = '') → str

get_tracking_model(tracer: str = 'unknown', collector_id: str = '') → TrackingInterface

Get a sqlalchemy model for tracking, each tracer will create a table in database.

Parameters:

• tracer (str) – name of tracer

• collector_id (str) – id of collector

Returns:

a sqlalchemy model for tracking

Return type:

type

inspect_all_collector_ids() → List[str]

inspect_all_tables(tracer: str | None = None, collector_id: str | None = None) → str

inspect_all_tracers() → List[str]

property sessionmaker

A sessionmaker for sqlalchemy session

table_name_to_collector_id(table_name: str) → str

table_name_to_tracer(table_name: str) → str

property table_prefix

Prefix for all table names

class duetector.db.TrackingInterface

A interface for tracking.

classmethod inspect_fields(value_as_type: bool = False) → Dict[str, Any]

to_analyzer_tracking() → Tracking

Convert to analyzer’s tracking model.

to_collector_tracking() → Tracking

Convert to collector’s tracking model.

class duetector.db.TrackingMixin

A mixin for sqlalchemy model to track process

comm: Mapped[str | None]

cwd: Mapped[str | None]

dt: Mapped[datetime | None]

extended: Mapped[Dict[str, Any]] = <sqlalchemy.orm.properties.MappedColumn object>

fname: Mapped[str | None]

gid: Mapped[int | None]

id: Mapped[int] = <sqlalchemy.orm.properties.MappedColumn object>

pid: Mapped[int | None]

uid: Mapped[int | None]