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
andwebsocket
will be processed.
- middleware_class#
- 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:
app¶ – The
next
ASGI app to call.config¶ – An instance of
OpenTelemetryConfig
- 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()])