opentelemetry#

class litestar.contrib.opentelemetry.OpenTelemetryConfig#

Bases: object

Configuration class for the OpenTelemetry middleware.

Consult the [OpenTelemetry ASGI documentation](https://opentelemetry-python-contrib.readthedocs.io/en/latest/instrumentation/asgi/asgi.html) for more info about the configuration options.

scope_span_details_extractor() tuple[str, dict[Any, str]]#

Callback which should return a string and a tuple, representing the desired default span name and a dictionary with any additional span attributes to set.

server_request_hook_handler: OpenTelemetryHookHandler | None = None#

Optional callback which is called with the server span and ASGI scope object for every incoming request.

client_request_hook_handler: OpenTelemetryHookHandler | None = None#

Optional callback which is called with the internal span and an ASGI scope which is sent as a dictionary for when the method receive is called.

client_response_hook_handler: OpenTelemetryHookHandler | None = None#

Optional callback which is called with the internal span and an ASGI event which is sent as a dictionary for when the method send is called.

meter_provider: MeterProvider | None = None#

Optional meter provider to use.

If omitted the current globally configured one is used.

tracer_provider: TracerProvider | None = None#

Optional tracer provider to use.

If omitted the current globally configured one is used.

__init__(scope_span_details_extractor: Callable[[Scope], tuple[str, dict[str, Any]]] = <function get_route_details_from_scope>, server_request_hook_handler: OpenTelemetryHookHandler | None = None, client_request_hook_handler: OpenTelemetryHookHandler | None = None, client_response_hook_handler: OpenTelemetryHookHandler | None = None, meter_provider: MeterProvider | None = None, tracer_provider: TracerProvider | None = None, meter: Meter | None = None, exclude: str | list[str] | None = None, exclude_opt_key: str | None = None, exclude_urls_env_key: str = 'LITESTAR', scopes: Scopes | None = None, middleware_class: type[OpenTelemetryInstrumentationMiddleware] = <class 'litestar.contrib.opentelemetry.middleware.OpenTelemetryInstrumentationMiddleware'>) None#
meter: Meter | None = None#

Optional meter to use.

If omitted the provided meter provider or the global one will be used.

exclude: str | list[str] | None = None#

A pattern or list of patterns to skip in the Allowed Hosts middleware.

exclude_opt_key: str | None = None#

An identifier to use on routes to disable hosts check for a particular route.

exclude_urls_env_key: str = 'LITESTAR'#

Key to use when checking whether a list of excluded urls is passed via ENV.

OpenTelemetry supports excluding urls by passing an env in the format ‘{exclude_urls_env_key}_EXCLUDED_URLS’. With the default being LITESTAR_EXCLUDED_URLS.

scopes: Scopes | None = None#

ASGI scopes processed by the middleware, if None both http and websocket will be processed.

middleware_class#

alias of OpenTelemetryInstrumentationMiddleware

property middleware: DefineMiddleware#

Create an instance of DefineMiddleware that wraps with.

[OpenTelemetry InstrumentationMiddleware][litestar.contrib.opentelemetry.OpenTelemetryInstrumentationMiddleware] or a subclass of this middleware.

Returns:

An instance of DefineMiddleware.

class litestar.contrib.opentelemetry.OpenTelemetryInstrumentationMiddleware#

Bases: AbstractMiddleware

OpenTelemetry Middleware.

__init__(app: ASGIApp, config: OpenTelemetryConfig) None#

Middleware that adds OpenTelemetry instrumentation to the application.

Parameters:
class litestar.contrib.opentelemetry.OpenTelemetryPlugin#

Bases: InitPluginProtocol

OpenTelemetry Plugin.

__init__(config: OpenTelemetryConfig | None = None) None#
on_app_init(app_config: AppConfig) AppConfig#

Receive the AppConfig instance after on_app_init hooks have been called.

Examples

from litestar import Litestar, get
from litestar.di import Provide
from litestar.plugins import InitPluginProtocol


def get_name() -> str:
    return "world"


@get("/my-path")
def my_route_handler(name: str) -> dict[str, str]:
    return {"hello": name}


class MyPlugin(InitPluginProtocol):
    def on_app_init(self, app_config: AppConfig) -> AppConfig:
        app_config.dependencies["name"] = Provide(get_name)
        app_config.route_handlers.append(my_route_handler)
        return app_config


app = Litestar(plugins=[MyPlugin()])
Parameters:

app_config – The AppConfig instance.

Returns:

The app config object.

litestar.contrib.opentelemetry.config.OpenTelemetryHookHandler#

alias of Callable[[Span, dict], None]