Patterns for including Telemetry

[adam-moss]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the Typer documentation, with the integrated search.
• I already searched in Google "How to X in Typer" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to Typer but to Click.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import typer
from rich.console import Console

app: typer.Typer = typer.Typer(
    no_args_is_help=True,
    rich_markup_mode="rich",
)

stdout: Console = Console()
stderr: Console = Console(stderr=True)


@app.callback()
def _callback() -> None:
    pass


@app.command()
def do_something() -> None:
  ...


if __name__ == "__main__":
    app()

Description

I am wondering what the recommended / better practice patterns are for including telemetry within a typer based cli?

For example, if I had a decorator like:

def log_execution_time(func: Callable[..., Any]) -> Callable[..., Any]:
    """Log execution time and process time."""

    def wrapper(  # noqa: ANN202
        *args: tuple,
        **kwargs,  # noqa: ANN003
    ):
        """
        Execute the decorated function and log its execution and process time.

        Args:
            *args: Variable length argument list.
            **kwargs: Arbitrary keyword arguments.

        Returns:
            Any: The return value of the decorated function.
        """
        start_execution_time: float = time.monotonic()
        start_process_time: float = time.process_time()

        result: Any = func(*args, **kwargs)

        execution_time: float = time.monotonic() - start_execution_time
        process_time: float = time.process_time() - start_cpu_time

        logger.debug(
            f"{func.__name__} took {execution_time:.6f} seconds to execute and {process_time:.6f} seconds of process time",
        )

        return result

    return wrapper

How could like be applied to a method in a typer cli?

I've tried

@log_execute_time
@app.command()
def do_something() -> None:
  ...

This doesn't fail but equally doesn't log anything. Doing it the opposite way:

@app.command()
@log_execute_time
def do_something() -> None:
  ...

Raises error RuntimeError: Type not yet supported: <class 'tuple'>. If I remove the tuple type annotation I instead get the error No such command 'do_something'

Operating System

macOS

Operating System Details

Monterey 12.6.3

Typer Version

0.7.0

Python Version

3.11.2

Additional Context

I'm currently testing it by using the logging module but ultimately it'll use OpenTelemetry.

[menzenski]

@adam-moss did you ever get this working? I'm in the same boat, wanting to instrument all CLI commands (also for OTEL).

[adam-moss]

Sadly not, I had to park it for now.

[linsomniac]

Have you considered using functools.wraps in your decorator? https://docs.python.org/3/library/functools.html#functools.wraps

[linsomniac]

I've had a chance to test this and I concur that putting the decorator above the @app.command() seems to have the effect that the wrapper function is never called. If I change the wrapper to the following, I am able to put the wrapper below the @app.command() and have it successfully run:

from functools import wraps

def test_wrap(func: Callable[..., Any]) -> Callable[..., Any]:
    @wraps(func)
    def wrapper(*args: tuple, **kwargs):
        print("Starting wrapper")
        return func(*args, **kwargs)

    return wrapper

[lachaib]

I concur on using functools.wraps because Typer is based on signatures when generating the command-line parser, and when it gets the wrapper it'll just see *args and **kwargs in the signature.

However, I may suggest a different approach in case you have many commands you need to decorate: you may subclass Typer and override the __call__ method.
You can simply do it with your decorator

class TyperWithTelemetry(typer.Typer):
     @log_execution_time
     def __call__(self, *args, **kwargs):
          return super().__call__(*args, **kwargs)