msticpy.data.drivers.kql_driver module
KQL Driver class.
- class msticpy.data.drivers.kql_driver.KqlDriver(connection_str: Optional[str] = None, **kwargs)
Bases:
DriverBaseKqlDriver class to execute kql queries.
Instantiate KqlDriver and optionally connect.
- Parameters
connection_str (str, optional) – Connection string
debug (bool) – print out additional diagnostic information.
- add_query_filter(name, query_filter)
Add an expression to the query attach filter.
- connect(connection_str: Optional[str] = None, **kwargs)
Connect to data source.
- Parameters
connection_str (Union[str, WorkspaceConfig, None]) – Connection string or WorkspaceConfig for the Sentinel Workspace.
kqlmagic_args (str, optional) – Additional string of parameters to be passed to KqlMagic
mp_az_auth (Union[bool, str, list, None], optional) – Optional parameter directing KqlMagic to use MSTICPy Azure authentication. Values can be: True or “default”: use the settings in msticpyconfig.yaml ‘Azure’ section str: single auth method name (‘msi’, ‘cli’, ‘env’, ‘vscode’, ‘powershell’, ‘cache’ or ‘interactive’) List[str]: list of acceptable auth methods from (‘msi’, ‘cli’, ‘env’, ‘vscode’, ‘powershell’, ‘cache’ or ‘interactive’)
mp_az_tenant_id (str, optional) – Optional parameter specifying a Tenant ID for use by MSTICPy Azure authentication.
workspace (str, optional) – Alternative to supplying a WorkspaceConfig object as the connection_str parameter. Giving a workspace name will fetch the workspace settings from msticpyconfig.yaml.
- 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]]
- static get_http_timeout(**kwargs)
Get http timeout from settings or kwargs.
- property instance: Optional[str]
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.
- query(query: str, query_source: Optional[QuerySource] = None, **kwargs) Union[DataFrame, Any]
Execute query string and return DataFrame of results.
- Parameters
query (str) – The query to execute
query_source (QuerySource) – The query definition object
- Returns
A DataFrame (if successful) or the underlying provider result if an error.
- Return type
Union[pd.DataFrame, results.ResultSet]
- property query_attach_spec: Dict[str, Set[str]]
Parameters that determine whether a query is relevant for the driver.
- query_with_results(query: str, **kwargs) Tuple[DataFrame, Kqlmagic.kql_proxy.KqlResponse]
Execute query string and return DataFrame of results.
- Parameters
query (str) – The kql query to execute
- Returns
A DataFrame (if successfull) and Kql ResultSet.
- Return type
Tuple[pd.DataFrame, results.ResultSet]
- 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]