msticpy.data.drivers.odata_driver module
OData Driver class.
- class msticpy.data.drivers.odata_driver.OData(*, debug=False, max_threads=4, **kwargs)
Bases:
DriverBaseParent class to retrieve date from an oauth based API.
Instantiate OData driver and optionally connect.
- Parameters:
connect (bool, optional) – Set true if you want to connect to the provider at initialization
debug (bool)
max_threads (int)
- CONFIG_NAME: ClassVar[str] = ''
- aad_token: str | None
- add_query_filter(name, query_filter)
Add an expression to the query attach filter.
- Parameters:
name (str)
query_filter (str | Iterable)
- api_root: str | None
- api_ver: str | None
- connect(connection_str=None, *, delegated_auth=False, instance=None, **kwargs)
Connect to oauth data source.
- Parameters:
connection_str (Optional[str], optional) – Connect to a data source
instance (Optional[str], optional) – Optional name of configuration instance - this is added as a prefix to the driver configuration key name when searching for configuration in the msticpyconfig.yaml
delegated_auth (bool) – Boolean to indicate if delegated auth should be used.
self (Self)
- Return type:
None
Notes
Connection string fields: tenant_id client_id client_secret apiRoot apiVersion
- 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)
Return value or KeyError from driver properties.
- Parameters:
name (str)
- Return type:
Any
- 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.
- msal_auth: MSALDelegatedAuth | None
- oauth_url: str | None
- abstractmethod query(query, query_source=None, **kwargs)
Execute query string and return DataFrame of results.
- Parameters:
query (str) – The query to execute
query_source (QuerySource) – The query definition object
self (Self)
- Returns:
A DataFrame (if successful) or the underlying provider result if an error.
- Return type:
pd.DataFrame | str | None
- property query_attach_spec: dict[str, set[str]]
Parameters that determine whether a query is relevant for the driver.
- query_usable(query_source)
Return True if query should be exposed for this driver.
- Parameters:
query_source (QuerySource)
- Return type:
bool
- query_with_results(query, **kwargs)
Execute query string and return DataFrame of results.
- Parameters:
query (str) – The kql query to execute
self (Self)
- Returns:
A DataFrame (if successful) and Kql ResultSet.
- Return type:
Tuple[pd.DataFrame, results.ResultSet]
- req_body: dict[str, str | None] | None
- req_headers: dict[str, Any]
- request_uri: str | None
- property schema: dict[str, dict]
Return current data schema of connection.
- Returns:
Data schema of current connection.
- Return type:
Dict[str, Dict]
- scopes: list[str] | None
- 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, value)
Set an item in driver properties.
- Parameters:
name (str)
value (Any)