msticpy.data.drivers.driver_base module

Data driver base class.

class msticpy.data.drivers.driver_base.DriverBase(**kwargs)

Bases: ABC

Base class for data providers.

Initialize new instance.

add_query_filter(name: str, query_filter: str | Iterable)

Add an expression to the query attach filter.

abstract connect(connection_str: str | None = None, **kwargs)

Connect to data source.

Parameters:

connection_str (Optional[str]) – Connect to a data source

property connected: bool

Return true if at least one connection has been made.

Returns:

True if a successful connection has been made.

Return type:

bool

Notes

This does not guarantee that the last data source connection was successful. It is a best effort to track whether the provider has made at least one successful authentication.

property driver_queries: Iterable[Dict[str, Any]]

Return queries retrieved from the service after connecting.

Returns:

List of Dictionary of query_name, query_text. Name of container to add queries to.

Return type:

List[Dict[str, str]]

get_driver_property(name: str) → Any

Return value or KeyError from driver properties.

static get_http_timeout(**kwargs)

Get http timeout from settings or kwargs.

property instance: str | None

Return instance name, if one is set.

Returns:

The name of driver instance or None if the driver does not support multiple instances

Return type:

Optional[str]

property loaded: bool

Return true if the provider is loaded.

Returns:

True if the provider is loaded.

Return type:

bool

Notes

This is not relevant for some providers.

abstract query(query: str, query_source: QuerySource | None = None, **kwargs) → DataFrame | Any

Execute query string and return DataFrame of results.

Parameters:

• query (str) – The query to execute

• query_source (QuerySource) – The query definition object

• kwargs – Are passed to the underlying provider query method, if supported.

Returns:

A DataFrame (if successful) or the underlying provider result if an error.

Return type:

Union[pd.DataFrame, Any]

property query_attach_spec: Dict[str, Set[str]]

Parameters that determine whether a query is relevant for the driver.

query_usable(query_source: QuerySource) → bool

Return True if query should be exposed for this driver.

abstract query_with_results(query: str, **kwargs) → Tuple[DataFrame, Any]

Execute query string and return DataFrame plus native results.

Parameters:

query (str) – The query to execute

Returns:

A DataFrame and native results.

Return type:

Tuple[pd.DataFrame,Any]

property schema: Dict[str, Dict]

Return current data schema of connection.

Returns:

Data schema of current connection.

Return type:

Dict[str, Dict]

property service_queries: Tuple[Dict[str, str], str]

Return queries retrieved from the service after connecting.

Returns:

Dictionary of query_name, query_text. Name of container to add queries to.

Return type:

Tuple[Dict[str, str], str]

set_driver_property(name: str, value: Any)

Set an item in driver properties.

class msticpy.data.drivers.driver_base.DriverProps

Bases: object

Defined driver properties.

EFFECTIVE_ENV = 'effective_environment'

FILTER_ON_CONNECT = 'filter_queries_on_connect'

FORMATTERS = 'formatters'

HAS_DRIVER_QUERIES = 'has_driver_queries'

MAX_PARALLEL = 'max_parallel'

PROPERTY_TYPES: Dict[str, Any] = {'effective_environment': (<class 'str'>, <enum 'DataEnvironment'>), 'filter_queries_on_connect': <class 'bool'>, 'formatters': <class 'dict'>, 'has_driver_queries': <class 'bool'>, 'max_parallel': <class 'int'>, 'public_attribs': <class 'dict'>, 'supports_async': <class 'bool'>, 'supports_threading': <class 'bool'>, 'use_query_paths': <class 'bool'>}

PUBLIC_ATTRS = 'public_attribs'

SUPPORTS_ASYNC = 'supports_async'

SUPPORTS_THREADING = 'supports_threading'

USE_QUERY_PATHS = 'use_query_paths'

classmethod defaults()

Return default values for driver properties.

classmethod valid_type(property_name: str, value: Any) → bool

Return expected property type.