Skip to content

Running Traces

tabular_tracing.py handles all tracing capabilities in your model.

It also includes some pre built traces to make life easier. Feel free to build your own.

Example
Monitor Queries
import pytabular as p
import logging as l
model = p.Tabular(CONNECTION_STR)
query_trace = p.QueryMonitor(model)
query_trace.start() # (1)

###

p.logger.setLevel(l.DEBUG) # (2)

###

query_trace.stop()
query_trace.drop() # (3)
  1. You will now start to see query traces on your model get outputed to your console.
  2. If you want to see the FULL query then set logging to DEBUG.
  3. You can drop on your own, or will get dropped on script exit.

BaseTrace

Generates trace to be run on Server.

This is the base class to customize the type of Trace you are looking for. It's recommended to use the out of the box traces built. It's on the roadmap to have an intuitive way to build traces for users.

Source code in pytabular/tabular_tracing.py
class BaseTrace:
    """Generates trace to be run on Server.

    This is the base class to customize the type of Trace you are looking for.
    It's recommended to use the out of the box traces built.
    It's on the roadmap to have an intuitive way to build traces for users.
    """

    def __init__(
        self,
        tabular_class,
        trace_events: List[TraceEvent],
        trace_event_columns: List[TraceColumn],
        handler: Callable,
    ) -> None:
        """This will `build()`, `add()`, and `update()` the trace to model.

        It will also register the dropping on the trace on exiting python.

        Args:
            tabular_class (Tabular): The model you want the trace for.
            trace_events (List[TraceEvent]): The TraceEvents you want have in your trace.
                From Microsoft.AnalysisServices.TraceEventClass.
            trace_event_columns (List[TraceColumn]): The trace event columns you want in your trace.
                From Microsoft.AnalysisServices.TraceColumn.
            handler (Callable): The `handler` is a function that will take in two args.
                The first arg is `source` and it is currently unused.
                The second is arg is `args` and here
                is where you can access the results of the trace.
        """
        logger.debug("Trace Base Class initializing...")
        self.Name = "PyTabular_" + "".join(
            random.SystemRandom().choices(
                [str(x) for x in [y for y in range(0, 10)]], k=10
            )
        )
        self.ID = self.Name.replace("PyTabular_", "")
        self.Trace = Trace(self.Name, self.ID)
        logger.debug(f"Trace {self.Trace.Name} created...")
        self.tabular_class = tabular_class
        self.Event_Categories = self._query_dmv_for_event_categories()

        self.trace_events = trace_events
        self.trace_event_columns = trace_event_columns
        self.handler = handler

        self.build()
        self.add()
        self.update()
        atexit.register(self.drop)

    def build(self) -> bool:
        """Run on init.

        This will take the inputed arguments for the class
        and attempt to build the Trace.

        Returns:
                bool: True if successful
        """
        logger.info(f"Building Trace {self.Name}")
        te = [TraceEvent(trace_event) for trace_event in self.trace_events]
        logger.debug(f"Adding Events to... {self.Trace.Name}")
        [self.Trace.get_Events().Add(t) for t in te]

        def add_column(trace_event, trace_event_column):
            """Adds the column to trace event."""
            try:
                trace_event.Columns.Add(trace_event_column)
            except Exception:
                logger.warning(f"{trace_event} - {trace_event_column} Skipped")
                pass

        logger.debug("Adding Trace Event Columns...")
        [
            add_column(trace_event, trace_event_column)
            for trace_event_column in self.trace_event_columns
            for trace_event in te
            if str(trace_event_column.value__)
            in self.Event_Categories[str(trace_event.EventID.value__)]
        ]

        logger.debug("Adding Handler to Trace...")
        self.handler = TraceEventHandler(self.handler)
        self.Trace.OnEvent += self.handler
        return True

    def add(self) -> int:
        """Runs on init. Adds built trace to the Server.

        Returns:
                int: Return int of placement in Server.Traces.get_Item(int).
        """
        logger.info(f"Adding {self.Name} to {self.tabular_class.Server.Name}")
        return self.tabular_class.Server.Traces.Add(self.Trace)

    def update(self) -> None:
        """Runs on init. Syncs with Server.

        Returns:
            None: Returns None.
                Unless unsuccessful then it will return the error from Server.
        """
        logger.info(f"Updating {self.Name} in {self.tabular_class.Server.Name}")
        if self.tabular_class.Server.Connected is False:
            self.tabular_class.reconnect()

        return self.Trace.Update()

    def start(self) -> None:
        """Call when you want to start the trace.

        Returns:
            None: Returns None.
                Unless unsuccessful then it will return the error from Server.
        """
        logger.info(f"Starting {self.Name} in {self.tabular_class.Server.Name}")
        return self.Trace.Start()

    def stop(self) -> None:
        """Call when you want to stop the trace.

        Returns:
            None: Returns None.
                Unless unsuccessful then it will return the error from Server.
        """
        logger.info(f"Stopping {self.Name} in {self.tabular_class.Server.Name}")
        return self.Trace.Stop()

    def drop(self) -> None:
        """Call when you want to drop the trace.

        Returns:
            None: Returns None. Unless unsuccessful,
                then it will return the error from Server.
        """
        logger.info(f"Dropping {self.Name} in {self.tabular_class.Server.Name}")
        atexit.unregister(self.drop)
        return self.Trace.Drop()

    def _query_dmv_for_event_categories(self):
        """Internal use. Called during the building process of a refresh.

        It is used to locate allowed columns for event categories.
        This is done by executing a `Tabular().Query()`
        on the `DISCOVER_EVENT_CATEGORIES` table in the DMV.
        Then the function will parse the results,
        as it is xml inside of rows.
        """
        event_categories = {}
        events = []
        logger.debug("Querying DMV for columns rules...")
        logger.debug("select * from $SYSTEM.DISCOVER_TRACE_EVENT_CATEGORIES")
        df = self.tabular_class.query(
            "select * from $SYSTEM.DISCOVER_TRACE_EVENT_CATEGORIES"
        )
        for index, row in df.iterrows():
            xml_data = xmltodict.parse(row.Data)
            if isinstance(xml_data["EVENTCATEGORY"]["EVENTLIST"]["EVENT"], list):
                events += [
                    event for event in xml_data["EVENTCATEGORY"]["EVENTLIST"]["EVENT"]
                ]
            else:
                events += [xml_data["EVENTCATEGORY"]["EVENTLIST"]["EVENT"]]
        for event in events:
            event_categories[event["ID"]] = [
                column["ID"] for column in event["EVENTCOLUMNLIST"]["EVENTCOLUMN"]
            ]
        return event_categories

__init__(tabular_class, trace_events, trace_event_columns, handler)

This will build(), add(), and update() the trace to model.

It will also register the dropping on the trace on exiting python.

Parameters:

Name Type Description Default
tabular_class Tabular

The model you want the trace for.

required
trace_events List[TraceEvent]

The TraceEvents you want have in your trace. From Microsoft.AnalysisServices.TraceEventClass.

required
trace_event_columns List[TraceColumn]

The trace event columns you want in your trace. From Microsoft.AnalysisServices.TraceColumn.

required
handler Callable

The handler is a function that will take in two args. The first arg is source and it is currently unused. The second is arg is args and here is where you can access the results of the trace.

required
Source code in pytabular/tabular_tracing.py
def __init__(
    self,
    tabular_class,
    trace_events: List[TraceEvent],
    trace_event_columns: List[TraceColumn],
    handler: Callable,
) -> None:
    """This will `build()`, `add()`, and `update()` the trace to model.

    It will also register the dropping on the trace on exiting python.

    Args:
        tabular_class (Tabular): The model you want the trace for.
        trace_events (List[TraceEvent]): The TraceEvents you want have in your trace.
            From Microsoft.AnalysisServices.TraceEventClass.
        trace_event_columns (List[TraceColumn]): The trace event columns you want in your trace.
            From Microsoft.AnalysisServices.TraceColumn.
        handler (Callable): The `handler` is a function that will take in two args.
            The first arg is `source` and it is currently unused.
            The second is arg is `args` and here
            is where you can access the results of the trace.
    """
    logger.debug("Trace Base Class initializing...")
    self.Name = "PyTabular_" + "".join(
        random.SystemRandom().choices(
            [str(x) for x in [y for y in range(0, 10)]], k=10
        )
    )
    self.ID = self.Name.replace("PyTabular_", "")
    self.Trace = Trace(self.Name, self.ID)
    logger.debug(f"Trace {self.Trace.Name} created...")
    self.tabular_class = tabular_class
    self.Event_Categories = self._query_dmv_for_event_categories()

    self.trace_events = trace_events
    self.trace_event_columns = trace_event_columns
    self.handler = handler

    self.build()
    self.add()
    self.update()
    atexit.register(self.drop)

build()

Run on init.

This will take the inputed arguments for the class and attempt to build the Trace.

Returns:

Name Type Description
bool bool

True if successful

Source code in pytabular/tabular_tracing.py
def build(self) -> bool:
    """Run on init.

    This will take the inputed arguments for the class
    and attempt to build the Trace.

    Returns:
            bool: True if successful
    """
    logger.info(f"Building Trace {self.Name}")
    te = [TraceEvent(trace_event) for trace_event in self.trace_events]
    logger.debug(f"Adding Events to... {self.Trace.Name}")
    [self.Trace.get_Events().Add(t) for t in te]

    def add_column(trace_event, trace_event_column):
        """Adds the column to trace event."""
        try:
            trace_event.Columns.Add(trace_event_column)
        except Exception:
            logger.warning(f"{trace_event} - {trace_event_column} Skipped")
            pass

    logger.debug("Adding Trace Event Columns...")
    [
        add_column(trace_event, trace_event_column)
        for trace_event_column in self.trace_event_columns
        for trace_event in te
        if str(trace_event_column.value__)
        in self.Event_Categories[str(trace_event.EventID.value__)]
    ]

    logger.debug("Adding Handler to Trace...")
    self.handler = TraceEventHandler(self.handler)
    self.Trace.OnEvent += self.handler
    return True

add()

Runs on init. Adds built trace to the Server.

Returns:

Name Type Description
int int

Return int of placement in Server.Traces.get_Item(int).

Source code in pytabular/tabular_tracing.py
def add(self) -> int:
    """Runs on init. Adds built trace to the Server.

    Returns:
            int: Return int of placement in Server.Traces.get_Item(int).
    """
    logger.info(f"Adding {self.Name} to {self.tabular_class.Server.Name}")
    return self.tabular_class.Server.Traces.Add(self.Trace)

update()

Runs on init. Syncs with Server.

Returns:

Name Type Description
None None

Returns None. Unless unsuccessful then it will return the error from Server.

Source code in pytabular/tabular_tracing.py
def update(self) -> None:
    """Runs on init. Syncs with Server.

    Returns:
        None: Returns None.
            Unless unsuccessful then it will return the error from Server.
    """
    logger.info(f"Updating {self.Name} in {self.tabular_class.Server.Name}")
    if self.tabular_class.Server.Connected is False:
        self.tabular_class.reconnect()

    return self.Trace.Update()

start()

Call when you want to start the trace.

Returns:

Name Type Description
None None

Returns None. Unless unsuccessful then it will return the error from Server.

Source code in pytabular/tabular_tracing.py
def start(self) -> None:
    """Call when you want to start the trace.

    Returns:
        None: Returns None.
            Unless unsuccessful then it will return the error from Server.
    """
    logger.info(f"Starting {self.Name} in {self.tabular_class.Server.Name}")
    return self.Trace.Start()

stop()

Call when you want to stop the trace.

Returns:

Name Type Description
None None

Returns None. Unless unsuccessful then it will return the error from Server.

Source code in pytabular/tabular_tracing.py
def stop(self) -> None:
    """Call when you want to stop the trace.

    Returns:
        None: Returns None.
            Unless unsuccessful then it will return the error from Server.
    """
    logger.info(f"Stopping {self.Name} in {self.tabular_class.Server.Name}")
    return self.Trace.Stop()

drop()

Call when you want to drop the trace.

Returns:

Name Type Description
None None

Returns None. Unless unsuccessful, then it will return the error from Server.

Source code in pytabular/tabular_tracing.py
def drop(self) -> None:
    """Call when you want to drop the trace.

    Returns:
        None: Returns None. Unless unsuccessful,
            then it will return the error from Server.
    """
    logger.info(f"Dropping {self.Name} in {self.tabular_class.Server.Name}")
    atexit.unregister(self.drop)
    return self.Trace.Drop()

RefreshTrace

Bases: BaseTrace

Subclass of BaseTrace(). Usefull for monitoring refreshes.

This is the default trace that is run on refreshes. It will output all the various details into logger(). See _refresh_handler() for more details on what gets put into logger().

Source code in pytabular/tabular_tracing.py
class RefreshTrace(BaseTrace):
    """Subclass of `BaseTrace()`. Usefull for monitoring refreshes.

    This is the default trace that is run on refreshes.
    It will output all the various details into `logger()`.
    See `_refresh_handler()` for more details on what gets
    put into `logger()`.
    """

    def __init__(
        self,
        tabular_class,
        trace_events: List[TraceEvent] = [
            TraceEventClass.ProgressReportBegin,
            TraceEventClass.ProgressReportCurrent,
            TraceEventClass.ProgressReportEnd,
            TraceEventClass.ProgressReportError,
        ],
        trace_event_columns: List[TraceColumn] = [
            TraceColumn.EventSubclass,
            TraceColumn.CurrentTime,
            TraceColumn.ObjectName,
            TraceColumn.ObjectPath,
            TraceColumn.DatabaseName,
            TraceColumn.SessionID,
            TraceColumn.TextData,
            TraceColumn.EventClass,
            TraceColumn.ProgressTotal,
        ],
        handler: Callable = _refresh_handler,
    ) -> None:
        """Init will extend through `BaseTrace()`. But pass through specific params.

        Args:
            tabular_class (Tabular): This is your `Tabular()` class.
            trace_events (List[TraceEvent], optional): Defaults to [
                TraceEventClass.ProgressReportBegin,
                TraceEventClass.ProgressReportCurrent, TraceEventClass.ProgressReportEnd,
                TraceEventClass.ProgressReportError, ].
            trace_event_columns (List[TraceColumn], optional): Defaults to
                [ TraceColumn.EventSubclass, TraceColumn.CurrentTime,
                TraceColumn.ObjectName, TraceColumn.ObjectPath, TraceColumn.DatabaseName,
                TraceColumn.SessionID, TraceColumn.TextData, TraceColumn.EventClass,
                TraceColumn.ProgressTotal, ].
            handler (Callable, optional): _description_. Defaults to _refresh_handler.
        """
        super().__init__(tabular_class, trace_events, trace_event_columns, handler)

__init__(tabular_class, trace_events=[TraceEventClass.ProgressReportBegin, TraceEventClass.ProgressReportCurrent, TraceEventClass.ProgressReportEnd, TraceEventClass.ProgressReportError], trace_event_columns=[TraceColumn.EventSubclass, TraceColumn.CurrentTime, TraceColumn.ObjectName, TraceColumn.ObjectPath, TraceColumn.DatabaseName, TraceColumn.SessionID, TraceColumn.TextData, TraceColumn.EventClass, TraceColumn.ProgressTotal], handler=_refresh_handler)

Init will extend through BaseTrace(). But pass through specific params.

Parameters:

Name Type Description Default
tabular_class Tabular

This is your Tabular() class.

required
trace_events List[TraceEvent]

Defaults to [ TraceEventClass.ProgressReportBegin, TraceEventClass.ProgressReportCurrent, TraceEventClass.ProgressReportEnd, TraceEventClass.ProgressReportError, ].

[ProgressReportBegin, ProgressReportCurrent, ProgressReportEnd, ProgressReportError]
trace_event_columns List[TraceColumn]

Defaults to [ TraceColumn.EventSubclass, TraceColumn.CurrentTime, TraceColumn.ObjectName, TraceColumn.ObjectPath, TraceColumn.DatabaseName, TraceColumn.SessionID, TraceColumn.TextData, TraceColumn.EventClass, TraceColumn.ProgressTotal, ].

[EventSubclass, CurrentTime, ObjectName, ObjectPath, DatabaseName, SessionID, TextData, EventClass, ProgressTotal]
handler Callable

description. Defaults to _refresh_handler.

_refresh_handler
Source code in pytabular/tabular_tracing.py
def __init__(
    self,
    tabular_class,
    trace_events: List[TraceEvent] = [
        TraceEventClass.ProgressReportBegin,
        TraceEventClass.ProgressReportCurrent,
        TraceEventClass.ProgressReportEnd,
        TraceEventClass.ProgressReportError,
    ],
    trace_event_columns: List[TraceColumn] = [
        TraceColumn.EventSubclass,
        TraceColumn.CurrentTime,
        TraceColumn.ObjectName,
        TraceColumn.ObjectPath,
        TraceColumn.DatabaseName,
        TraceColumn.SessionID,
        TraceColumn.TextData,
        TraceColumn.EventClass,
        TraceColumn.ProgressTotal,
    ],
    handler: Callable = _refresh_handler,
) -> None:
    """Init will extend through `BaseTrace()`. But pass through specific params.

    Args:
        tabular_class (Tabular): This is your `Tabular()` class.
        trace_events (List[TraceEvent], optional): Defaults to [
            TraceEventClass.ProgressReportBegin,
            TraceEventClass.ProgressReportCurrent, TraceEventClass.ProgressReportEnd,
            TraceEventClass.ProgressReportError, ].
        trace_event_columns (List[TraceColumn], optional): Defaults to
            [ TraceColumn.EventSubclass, TraceColumn.CurrentTime,
            TraceColumn.ObjectName, TraceColumn.ObjectPath, TraceColumn.DatabaseName,
            TraceColumn.SessionID, TraceColumn.TextData, TraceColumn.EventClass,
            TraceColumn.ProgressTotal, ].
        handler (Callable, optional): _description_. Defaults to _refresh_handler.
    """
    super().__init__(tabular_class, trace_events, trace_event_columns, handler)

QueryMonitor

Bases: BaseTrace

Subclass of BaseTrace(). Usefull for monitoring queries.

The default handler for QueryMonitor() shows full query in logger.debug(). So you will need to set your logger to debug() if you would like to see them. Otherwise, will show basic info on who/what is querying.

Source code in pytabular/tabular_tracing.py
class QueryMonitor(BaseTrace):
    """Subclass of `BaseTrace()`. Usefull for monitoring queries.

    The default handler for `QueryMonitor()` shows full query in `logger.debug()`.
    So you will need to set your logger to `debug()` if you would like to see them.
    Otherwise, will show basic info on who/what is querying.
    """

    def __init__(
        self,
        tabular_class,
        trace_events: List[TraceEvent] = [TraceEventClass.QueryEnd],
        trace_event_columns: List[TraceColumn] = [
            TraceColumn.EventSubclass,
            TraceColumn.StartTime,
            TraceColumn.EndTime,
            TraceColumn.Duration,
            TraceColumn.Severity,
            TraceColumn.Error,
            TraceColumn.NTUserName,
            TraceColumn.DatabaseName,
            TraceColumn.ApplicationName,
            TraceColumn.TextData,
        ],
        handler: Callable = _query_monitor_handler,
    ) -> None:
        """Init will extend through to BaseTrace, but pass through specific params.

        Args:
            tabular_class (Tabular): This is your `Tabular()` class.
                All that will need to provided to successfully init.
            trace_events (List[TraceEvent], optional): Defaults to [TraceEventClass.QueryEnd].
            trace_event_columns (List[TraceColumn], optional): Defaults to
                [ TraceColumn.EventSubclass, TraceColumn.StartTime, TraceColumn.EndTime,
                TraceColumn.Duration, TraceColumn.Severity, TraceColumn.Error,
                TraceColumn.NTUserName, TraceColumn.DatabaseName, TraceColumn.ApplicationName,
                TraceColumn.TextData, ].
            handler (Callable, optional): Defaults to `_query_monitor_handler()`.
        """
        super().__init__(tabular_class, trace_events, trace_event_columns, handler)
        logger.info("Query text lives in DEBUG, adjust logging to see query text.")

__init__(tabular_class, trace_events=[TraceEventClass.QueryEnd], trace_event_columns=[TraceColumn.EventSubclass, TraceColumn.StartTime, TraceColumn.EndTime, TraceColumn.Duration, TraceColumn.Severity, TraceColumn.Error, TraceColumn.NTUserName, TraceColumn.DatabaseName, TraceColumn.ApplicationName, TraceColumn.TextData], handler=_query_monitor_handler)

Init will extend through to BaseTrace, but pass through specific params.

Parameters:

Name Type Description Default
tabular_class Tabular

This is your Tabular() class. All that will need to provided to successfully init.

required
trace_events List[TraceEvent]

Defaults to [TraceEventClass.QueryEnd].

[QueryEnd]
trace_event_columns List[TraceColumn]

Defaults to [ TraceColumn.EventSubclass, TraceColumn.StartTime, TraceColumn.EndTime, TraceColumn.Duration, TraceColumn.Severity, TraceColumn.Error, TraceColumn.NTUserName, TraceColumn.DatabaseName, TraceColumn.ApplicationName, TraceColumn.TextData, ].

[EventSubclass, StartTime, EndTime, Duration, Severity, Error, NTUserName, DatabaseName, ApplicationName, TextData]
handler Callable

Defaults to _query_monitor_handler().

_query_monitor_handler
Source code in pytabular/tabular_tracing.py
def __init__(
    self,
    tabular_class,
    trace_events: List[TraceEvent] = [TraceEventClass.QueryEnd],
    trace_event_columns: List[TraceColumn] = [
        TraceColumn.EventSubclass,
        TraceColumn.StartTime,
        TraceColumn.EndTime,
        TraceColumn.Duration,
        TraceColumn.Severity,
        TraceColumn.Error,
        TraceColumn.NTUserName,
        TraceColumn.DatabaseName,
        TraceColumn.ApplicationName,
        TraceColumn.TextData,
    ],
    handler: Callable = _query_monitor_handler,
) -> None:
    """Init will extend through to BaseTrace, but pass through specific params.

    Args:
        tabular_class (Tabular): This is your `Tabular()` class.
            All that will need to provided to successfully init.
        trace_events (List[TraceEvent], optional): Defaults to [TraceEventClass.QueryEnd].
        trace_event_columns (List[TraceColumn], optional): Defaults to
            [ TraceColumn.EventSubclass, TraceColumn.StartTime, TraceColumn.EndTime,
            TraceColumn.Duration, TraceColumn.Severity, TraceColumn.Error,
            TraceColumn.NTUserName, TraceColumn.DatabaseName, TraceColumn.ApplicationName,
            TraceColumn.TextData, ].
        handler (Callable, optional): Defaults to `_query_monitor_handler()`.
    """
    super().__init__(tabular_class, trace_events, trace_event_columns, handler)
    logger.info("Query text lives in DEBUG, adjust logging to see query text.")