autonomy.analyse.logs.db

Database schemas and helpers

validate_sql_identifier

def validate_sql_identifier(identifier: str) -> str

Validate that an identifier is safe for use in SQL queries.

Arguments:

• identifier: The identifier to validate

Raises:

• ValueError: If the identifier is not safe

Returns:

The validated identifier

AgentLogsDB Objects

class AgentLogsDB()

Logs DB

__init__

def __init__(agent: str, file: Path) -> None

Initialize object.

select

def select(start_time: Optional[datetime] = None,
           end_time: Optional[datetime] = None,
           log_level: Optional[str] = None,
           period: Optional[int] = None,
           round_name: Optional[str] = None,
           behaviour_name: Optional[str] = None) -> List[LogRow]

Build select query.

execution_path

def execution_path() -> List[Tuple[int, str, str]]

Extraction FSM execution path

cursor

@property
def cursor() -> sqlite3.Cursor

Creates and returns a database cursor.

exists

def exists() -> bool

Check if table already exists.

delete

def delete() -> "AgentLogsDB"

Delete table

create

def create(reset: bool = False) -> "AgentLogsDB"

Create agent table

insert_many

def insert_many(logs: Iterator[LogRow]) -> "AgentLogsDB"

Insert a record