Jak dodać nagłówek HTTP do żądania w Django?

Przejdź do sekcji TLDR celem uzyskania szybkiej odpowiedzi na pytanie w tytule artykułu.

Zapisz się na newsletter, jeśli nie chcesz przegapić kolejnych publikacji.

Studium przypadku
Rozwiązanie z użyciem Django middleware
PoC
TLDR
Przydatne materiały

Studium przypadku

W środowisku rozproszonym, opartym o mikro-serwisy, istotną rolę odgrywa instrumentacja (ang. instrumentation), czyli dodanie fragmentów kodu, które umożliwiają dokładne monitorowanie działania aplikacji oraz mierzenie jej wydajności. Dzięki temu jesteśmy w stanie skutecznie zidentyfikować usługę, która może stanowić wąskie gardło całego systemu.

W jednym z projektów Django, nad którym swego czasu pracowałem, zaimplementowano instrumentację z wykorzystaniem narzędzi OpenTelemetry. Do propagacji kontekstu użyto propagatora B3, zgodnego ze standardem OpenTelemetry, który informacje o kontekście przekazuje poprzez nagłówki HTTP, o nazwie rozpoczynającej się prefiksem X-B3-.

W pewnym momencie zaszła potrzeba uruchomienia projektu na chmurze Google (Google Cloud Platform) i okazało się, że nasza usługa chmurowa Google używa trochę innego nagłówka do przekazywania informacji o kontekście, a mianowicie: X-Cloud-Trace-Context.

Bez wdawania się w szczegóły decyzyjne: nie chcieliśmy na tym etapie zmieniać sposobu propagacji w usłudze, więc musieliśmy odczytać wartość trace_id z nagłówka X-Cloud-Trace-Context i umieścić ją w innym nagłówku X-B3-TraceId na poziomie żądania (HTTP request), zanim jeszcze zostanie obsłużone przez kod widoków.

Rozwiązanie z użyciem Django middleware

Żeby zmodyfikować samo żądanie HTTP, modyfikując nagłówki, najlepiej posłużyć się mechanizmem middleware. W dosłownym tłumaczeniu jest to oprogramowanie pośredniczące i możemy je sobie wyobrazić jako system wtyczek, przez które przechodzą żądania (requests) oraz odpowiedzi (responses) HTTP podczas standardowej komunikacji:

*Uproszczony schemat obsługi żądań i odpowiedzi HTTP*.

W uproszczeniu: kod middleware ma dostęp do danych żądania (HttpRequest), zanim zostanie ono obsłużone przez metodę widoku, a także do danych odpowiedzi (HttpResponse) zwróconej przez widok, zanim trafi ona do użytkownika.

Django pozwala na definiowanie własnych klas/funkcji middleware, które później należy zarejestrować w ustawieniu settings.MIDDLEWARE.

Warto zwrócić uwagę, że kolejność rejestracji komponentów middleware ma znaczenie. Na powyższym schemacie, żądanie zostanie obsłużone w następującej kolejności: Middleware 1 -> Middleware 2 -> Middleware 3. Odpowiedź natomiast przejdzie przez warstwę pośredniczącą w odwrotnej kolejności: Middleware 3 -> Middleware 2 -> Middleware 1.

Podsumowując, komponent middleware jest odpowiednim miejscem na realizację naszego założenia, czyli przepisania wartości odczytanej z nagłówka X-Cloud-Trace-Context i umieszczeniu jej w innym nagłówku żądania HTTP.

PoC

Stwórzmy możliwie najprostszy projekt Django i dodajmy do niego aplikację o nazwie telemetry. Jeśli nigdy wcześniej nie pracowałeś(-aś) z Django, polecam zapoznać się z oficjalnym wprowadzeniem.
Zdefiniujmy prostą metodę widoku (dostępną pod ścieżką /telemetry), której zadaniem jest zwrócenie nagłówków żądania HTTP w formacie JSON:

from django.http import JsonResponse

def index(request):
    return JsonResponse(dict(request.headers))

# ...
urlpatterns = [
    path("telemetry/", include("telemetry.urls")),
    # ...
]

# ...
urlpatterns = [
    path("", views.index, name="index"),
]

Po uruchomieniu aplikacji z użyciem lokalnego serwera deweloperskiego (python manage.py runserver), możemy wysłać żądanie z ustawionym nagłówkiem X-B3-Traceid i ujrzeć przykładową odpowiedź (przejrzyście sformatowaną dzięki narzędziu jq):

$ curl -s -H "X-B3-TraceId: 463ac35c9f6413ad48485a3953bb6124" http://localhost:8000/telemetry/ | jq

{
  "Content-Length": "",
  "Content-Type": "text/plain",
  "Host": "localhost:8000",
  "User-Agent": "curl/7.68.0",
  "Accept": "*/*",
  "X-B3-Traceid": "463ac35c9f6413ad48485a3953bb6124"
}

Zgodnie z podejściem TDD (Test-Driven Development), zacznijmy od napisania automatycznych testów funkcjonalnych. Rozważmy 2 proste scenariusze:
- Fragment wartości nagłówka X-Cloud-Trace-Context jest przeniesiony do X-B3-Traceid, jeśli takowego nie ma w żądaniu.
- Jeśli nagłówek X-B3-Traceid jest już zdefiniowany w żądaniu, to zignoruj nagłówek X-Cloud-Trace-Context.

import json

from django.test import Client, TestCase


class TestTelemetryMiddleware(TestCase):
    X_CLOUD_TRACE_CONTEXT_HEADER_NAME = "X-Cloud-Trace-Context"
    X_B3_TRACEID_HEADER_NAME = "X-B3-Traceid"

    def setUp(self):
        self.client = Client()

    def test_copies_trace_id_from_X_CLOUD_TRACE_CONTEXT_to_X_B3_TRACEID_request_header(
        self,
    ):
        # Arrange
        valid_trace_id = "4bf92f3577b34da6a3ce929d0e0e4736"
        headers = {
            self.X_CLOUD_TRACE_CONTEXT_HEADER_NAME: self.__build_trace_context(
                valid_trace_id
            )
        }

        # Act
        response_data = self.__send_get_request("/telemetry/", headers)

        # Assert
        assert response_data[self.X_B3_TRACEID_HEADER_NAME] == valid_trace_id

    def test_ignores_X_CLOUD_TRACE_CONTEXT_if_X_B3_TRACEID_already_exists(self):
        # Arrange
        valid_trace_id = "4bf92f3577b34da6a3ce929d0e0e4736"
        valid_b3_trace_id = "f5196e6103f4ad68e25576e59a7c5699"
        headers = {
            self.X_CLOUD_TRACE_CONTEXT_HEADER_NAME: self.__build_trace_context(
                valid_trace_id
            ),
            self.X_B3_TRACEID_HEADER_NAME: valid_b3_trace_id,
        }

        # Act
        response_data = self.__send_get_request("/telemetry/", headers)

        # Assert
        assert response_data[self.X_B3_TRACEID_HEADER_NAME] == valid_b3_trace_id

    def __build_trace_context(
        self, trace_id: str, span_id: str = "17725314949316355921", sampled: bool = True
    ) -> str:
        return f"{trace_id}/{span_id};o={int(sampled)}"

    def __send_get_request(self, path: str, headers: dict) -> dict:
        # Make headers compatible with WSGI environment variables.
        meta_headers = {
            f"HTTP_{name.replace('-', '_')}": value for name, value in headers.items()
        }
        response = self.client.get(path, **meta_headers)
        return json.loads(response.content.decode("utf-8"))

Implementacja klasy TelemetryMiddleware, realizującej pożądaną przez nas funkcjonalność, wymaga prawidłowego sparsowania zawartości nagłówka X-Cloud-Trace-Context. Co prawda nie jest to trudne zadanie, ale zamiast wyważać otwarte drzwi, możemy podejrzeć, jak to zostało zrobione w udostępnionej przez Google bibliotece open source: opentelemetry-operations-python.

import re

from django.http import HttpRequest

# Make sure the regular expresson is compiled only once.
X_CLOUD_TRACE_CONTEXT_RE = re.compile(
    r"(?P<trace_id>[0-9a-f]{32})\/(?P<span_id>[\d]{1,20})(;o=(?P<trace_flags>\d+))?"
)


class TelemetryMiddleware:
    X_CLOUD_TRACE_CONTEXT_HEADER_NAME = "X-Cloud-Trace-Context"
    X_B3_TRACEID_HEADER_NAME = "X-B3-Traceid"
    X_B3_TRACEID_META_HEADER_NAME = f"HTTP_{X_B3_TRACEID_HEADER_NAME.replace('-', '_')}"

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request: HttpRequest):
        if self.X_B3_TRACEID_HEADER_NAME not in request.headers:
            trace_id = self.__extract_trace_id_from_context_header(request)

            request.META[self.X_B3_TRACEID_META_HEADER_NAME] = trace_id
            # Enforce the "HttpRequest.headers" cached property to be refreshed.
            request.__dict__.pop("headers", None)

        response = self.get_response(request)

        # Code to be executed for each request/response after
        # the view is called.

        return response

    def __extract_trace_id_from_context_header(self, request: HttpRequest) -> str:
        trace_context_header = request.headers.get(
            self.X_CLOUD_TRACE_CONTEXT_HEADER_NAME
        )

        if trace_context_header is None:
            return ""

        match = re.fullmatch(X_CLOUD_TRACE_CONTEXT_RE, trace_context_header)

        if match is None:
            return ""

        return match.group("trace_id")

Warto zwrócić uwagę na fragment kodu request.__dict__.pop("headers", None), który powoduje, że wartość HttpRequest.headers (cached property) zostanie odświeżona przy kolejnej próbie odczytu. Jest to konieczne, żeby właściwość zawierała wprowadzone przez nas zmiany do request.META.
Ostatnim krokiem jest zarejestrowanie klasy middleware w ustawieniach:

# ...
MIDDLEWARE = [
    "telemetry.middleware.TelemetryMiddleware",
    # ...
]
# ...

Po uruchomieniu testów (python manage.py test), wszystkie powinny zakończyć się sukcesem.

TLDR

Podsumowując, jeśli chcemy zmodyfikować nagłówki żądania HTTP przed obsłużeniem przez kod widoków Django:

Definiujemy klasę (lub metodę) middleware, w której modyfikujemy nagłówki żądania:

class MyMiddleware:
	def __init__(self, get_response):
	        self.get_response = get_response

	def __call__(self, request):
        # Kod wykonywany dla każdego żądania, zanim trafi do widoku.

		# Dodajemy nowy nagłówek w odpowiednim formacie:
		# HTTP_MY_HEADER oznacza nagłówek "My-Header".
		request.META["HTTP_MY_HEADER"] = "header_value"
		
		# Wymuszamy odświeżenie cached property HttpReqest.headers.
		request.__dict__.pop("headers", None)

        response = self.get_response(request)

        # Kod wykonywany dla każdej odpowiedzi, zwróconej przez widok.

        return response

Rejestrujemy komponent w ustawieniach settings.py:

MIDDLEWARE = [
	"path.to.MyMiddleware",
	# ...
]

Przydatne materiały

Unit Testing Django Middleware – jak przetestować sam komponent middleware przy użyciu testów jednostkowych.
StackOverflow: Django overwrite header value in request object – odrobinę szersze wyjaśnienie dotyczące odświeżenia właściwości HttpRequest.headers.

Jak dodać nagłówek HTTP do żądania w Django?

Studium przypadku

Rozwiązanie z użyciem Django middleware

PoC

TLDR

Przydatne materiały

O autorze

Łukasz Mieczkowski

Dodaj komentarz

Anuluj odpowiedź

Łukasz Mieczkowski

Kontakt

Studium przypadku

Rozwiązanie z użyciem Django middleware

PoC

TLDR

Przydatne materiały

O autorze

Łukasz Mieczkowski

Dodaj komentarz

Anuluj odpowiedź

Więcej

Łukasz Mieczkowski

Kontakt