in msticpy/data/drivers/kql_driver.py [0:0]
def connect(self, connection_str: Optional[str] = None, **kwargs): # noqa: MC0001
"""
Connect to data source.
Parameters
----------
connection_str : str
Connect to a data source
Other Parameters
----------------
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' or 'interactive')
List[str]: list of acceptable auth methods from ('msi', 'cli',
'env' or 'interactive')
mp_az_tenant_id: str, optional
Optional parameter specifying a Tenant ID for use by MSTICPy Azure
authentication.
"""
if not self._previous_connection:
print("Connecting...", end=" ")
mp_az_auth = kwargs.pop("mp_az_auth", "default")
mp_az_tenant_id = kwargs.pop("mp_az_tenant_id", None)
if isinstance(connection_str, WorkspaceConfig):
if not mp_az_tenant_id and "tenant_id" in connection_str:
mp_az_tenant_id = connection_str["tenant_id"]
connection_str = connection_str.code_connect_str
if not connection_str:
raise MsticpyKqlConnectionError(
f"A connection string is needed to connect to {self._connect_target}",
title="no connection string",
)
if "kqlmagic_args" in kwargs:
connection_str = connection_str + " " + kwargs["kqlmagic_args"]
# Default to using Azure Auth if possible.
if mp_az_auth and "try_token" not in kwargs:
self._set_az_auth_option(mp_az_auth, mp_az_tenant_id)
self.current_connection = connection_str
kql_err_setting = self._get_kql_option("short_errors")
self._connected = False
try:
self._set_kql_option("short_errors", False)
if self._ip is not None:
try:
kql_exec(connection_str)
if not self._previous_connection:
print("connected")
except KqlError as ex:
self._raise_kql_error(ex)
except KqlEngineError as ex:
self._raise_kql_engine_error(ex)
except AuthenticationError as ex:
self._raise_authn_error(ex)
except Exception as ex: # pylint: disable=broad-except
self._raise_adal_error(ex)
self._connected = True
self._previous_connection = True
self._schema = self._get_schema()
else:
print(f"Could not connect to kql query provider for {connection_str}")
return self._connected
finally:
self._set_kql_option("short_errors", kql_err_setting)