A starlette middleware for newrelic agent integration

app.py

from fastapi import FastAPI

from .NewRelicMiddleware import NewRelicMiddleware

app = FastAPI()
app.add_middleware(
    NewRelicMiddleware,
    config_file="/path/to/newrelic.ini",
    environment="test",
    log_level="debug",
)

# add routes and run server

NewRelicMiddleWare.py

import newrelic.agent
from starlette.types import ASGIApp, Receive, Scope, Send


def get_transaction_name(middleware, scope: Scope, *args, **kwargs):
    # Strip off the leading slash
    name = scope.get("path", "/no-path")
    if name == "/":
        return ""
    return name[1:]


def get_headers(middleware, scope: Scope, *args, **kwargs):
    return scope.get("headers", [])


def get_header_values(header_names):
    def retrieve(middleware, scope: Scope, *args, **kwargs):
        headers = scope.get("headers", [])
        # Iterate over all header names and get the first one that has a value
        for header_name in header_names:
            for header in headers:
                if header[0] == header_name:
                    return header[1].decode("utf-8")
        return None

    return retrieve


def get_scope_property(prop_name):
    def retrieve(middleware, scope: Scope, *args, **kwargs):
        return scope.get(prop_name)

    return retrieve


class NewRelicMiddleware:
    def __init__(
        self,
        app: ASGIApp,
        config_file=None,
        environment=None,
        ignore_errors=None,
        log_file=None,
        log_level=None,
    ):
        self.app = app
        newrelic.agent.initialize(
            config_file, environment, ignore_errors, log_file, log_level
        )
        newrelic.agent.register_application()

    @newrelic.agent.web_transaction(
        name=get_transaction_name,
        scheme=get_scope_property("scheme"),
        host=get_header_values([b"x-forwarded-host", b"host"]),
        request_path=get_scope_property("path"),
        request_method=get_scope_property("method"),
        query_string=get_scope_property("query_string"),
        headers=get_headers,
    )
    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        try:
            await self.app(scope, receive, send)
        except Exception as e:
            raise e from None

Middleware implementation inspired by https://discuss.newrelic.com/t/feature-idea-starlette-responder-framework-support/74128/5?u=_arne