DBAnalyzer

DBAnalyzer

class duetector.analyzer.db.DBAnalyzer(config: dict[str, Any] | None = None, *args, **kwargs)[source]

Bases: Analyzer

A analyzer using database.

We design this analyzer to be a top module, so it can be used as a standalone tools.

In this analyzer, we use SessionManager to manage database session.

Config scope is db_analyzer. db_analyzer.db is the scope for SessionManager.

Example

from duetector.analyzer.db import DBAnalyzer
from duetector.analyzer.models import Tracking as AT
from duetector.collectors.models import Tracking as CT


collector_id = "db_analyzer_tests_collector"
c_tracking = CT(
    tracer="t",
)
db_analyzer = DBAnalyzer()
m = db_analyzer.sm.get_tracking_model(c_tracking.tracer, collector_id)
with db_analyzer.sm.begin() as session:
    session.add(m(**c_tracking.model_dump(exclude=["tracer"])))
    session.commit()

a_tracking = AT(
    tracer=c_tracking.tracer,
)
assert a_tracking in db_analyzer.query()
assert a_tracking in db_analyzer.query(tracer=a_tracking.tracer)
assert a_tracking in db_analyzer.query(collector_id=collector_id)
assert a_tracking in db_analyzer.query(
        tracer=a_tracking.tracer, collector_id=collector_id
)
assert not db_analyzer.query(tracer="not-exist")
assert not db_analyzer.query(collector_id="not-exist")

Note

Currently, it will NOT be configured by DBcollector’s config, as we design it to be a standalone model.

_convert_row_to_tracking(columns: list[str], row: Any, tracer: str) → Tracking[source]

Convert a row to a tracking record.

Parameters:

columns (List[str]) – Columns.
row (Any) – Row.
tracer (str) – Tracer’s name.

Returns:

A tracking record.

Return type:

duetector.analyzer.models.Tracking

_table_brief(table_name: str, start_datetime: datetime | None = None, end_datetime: datetime | None = None, inspect: bool = True, inspect_type: bool = False, distinct: bool = False) → Brief[source]

Get a brief of a table.

Parameters:: table_name (str) – Table’s name.
Returns:: A brief of this table.
Return type:: Brief

async brief(tracers: list[str] | None = None, collector_ids: list[str] | None = None, start_datetime: datetime | None = None, end_datetime: datetime | None = None, with_details: bool = True, distinct: bool = False, inspect_type: bool = False) → AnalyzerBrief[source]

Get a brief of this analyzer.

Parameters:

tracers (Optional[List[str]], optional) – Tracers. Defaults to None, all tracers will be queried. If a specific tracer is not found, it will be ignored.
collector_ids (Optional[List[str]], optional) – Collector ids. Defaults to None, all collector ids will be queried. If a specific collector id is not found, it will be ignored.
start_datetime (Optional[datetime], optional) – Start time. Defaults to None.
end_datetime (Optional[datetime], optional) – End time. Defaults to None.
with_details (bool, optional) – With details. Defaults to True.
distinct (bool, optional) – Distinct. Defaults to False.

Returns:

A brief of this analyzer.

Return type:

AnalyzerBrief

default_config = {'db': {'engine': {'url': 'sqlite:///duetector-dbcollector.sqlite3'}, 'table_prefix': 'duetector_tracking'}, 'disabled': False}: Default config for DBAnalyzer.

get_all_collector_ids() → list[str][source]

Get all collector id from database.

Returns:: List of collector id.
Return type:: List[str]

get_all_tracers() → list[str][source]

Get all tracers from database.

Returns:: List of tracer’s name.
Return type:: List[str]

async query(tracers: list[str] | None = None, collector_ids: list[str] | None = None, start_datetime: datetime | None = None, end_datetime: datetime | None = None, start: int = 0, limit: int = 20, columns: list[str] | None = None, where: dict[str, Any] | None = None, distinct: bool = False, order_by_asc: list[str] | None = None, order_by_desc: list[str] | None = None) → list[Tracking][source]

Query all tracking records from database.

Parameters:

tracers (Optional[List[str]], optional) – Tracer’s name. Defaults to None, all tracers will be queried.
collector_ids (Optional[List[str]], optional) – Collector id. Defaults to None, all collector id will be queried.
start_datetime (Optional[datetime], optional) – Start time. Defaults to None.
end_datetime (Optional[datetime], optional) – End time. Defaults to None.
start (int, optional) – Start index. Defaults to 0.
limit (int, optional) – Limit of records. Defaults to 20. 0 means no limit.
columns (Optional[List[str]], optional) – Columns to query. Defaults to None, all columns will be queried.
where (Optional[Dict[str, Any]], optional) – Where clause. Defaults to None.
distinct (bool, optional) – Distinct. Defaults to False.
order_by_asc (Optional[List[str]], optional) – Order by asc. Defaults to None.
order_by_desc (Optional[List[str]], optional) – Order by desc. Defaults to None.

Returns:

List of tracking records.

Return type:

List[duetector.analyzer.models.Tracking]

duetector.analyzer.db.init_analyzer(config)[source]