developer.overheid.nl

Ontwikkelaarsportaal van de Nederlandse overheid

Ga naar hoofdinhoud

Python voorbeeldimplementatie met Flask

Een voorbeeld hoe Logboek Dataverwerkingen toe te passen met Flask in Python.

Nieuw project​

Maak een nieuw project aan:

uv init --bare ldv-example-python
cd ldv-example-python

Voeg dependency's toe​

# Flask
uv add "flask >=3.1.3, <4.0"

# OpenTelemetry
uv add "opentelemetry-sdk >=1.44.0, <2.0"
uv add "opentelemetry-exporter-otlp >=1.44.0, <2.0"

Logboek​

OpenTelemetry​

In Logboek wordt een OpenTelemetry Tracer geconfigureerd die alleen voor het registreren van verwerkingen gebruikt wordt.

Dit definieert ook een Flask-extentie.

logboek/flask.py
from flask import Config, Flask, current_app
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.semconv.attributes.service_attributes import SERVICE_NAME, SERVICE_VERSION
from opentelemetry.trace import Tracer

from logboek import ProcessingOperator


class Logboek:
    def __init__(self, app: Flask | None = None) -> None:
        self._processing_operator: ProcessingOperator

        if app is not None:
            self.init_app(app)

    def init_app(self, app: Flask) -> None:
        app.extensions["logboek"] = self

        config: Config = app.config.setdefault("LOGBOEK", {})

        endpoint: str | None = config.setdefault("endpoint")
        if not endpoint:
            raise RuntimeError("Logboek endpoint must be set")

        endpoint_insecure: bool = config.setdefault("endpoint_insecure", False)

        tracer = self._create_tracer(
            endpoint,
            endpoint_insecure,
            app.config.setdefault("APP_NAME", ""),
            app.config.setdefault("APP_VERSION", ""),
        )

        self._processing_operator = ProcessingOperator(tracer)

    def _create_tracer(self, endpoint: str, endpoint_insecure: bool, app_name: str, app_version: str) -> Tracer:
        processor = SimpleSpanProcessor(OTLPSpanExporter(endpoint, endpoint_insecure))

        resource: Resource = Resource.create({SERVICE_NAME: app_name, SERVICE_VERSION: app_version})
        provider = TracerProvider(resource=resource)
        provider.add_span_processor(processor)

        return provider.get_tracer("logboek-dataverwerkingen", "0.1.0")

    @property
    def processing_operator(self) -> ProcessingOperator:
        return self._processing_operator


def processing_operator() -> ProcessingOperator:
    return current_app.extensions["logboek"].processing_operator

Registreren van verwerkingen​

Met een ProcessingOperator kunnen we eenvoudig verwerkingen starten en direct de juiste attributen toevoegen. Met een Processing kan het "data subject" gezet worden.

logboek/__init__.py
from collections.abc import Generator, Mapping
from contextlib import _GeneratorContextManager, contextmanager

from opentelemetry.context import Context
from opentelemetry.trace import StatusCode, Tracer, use_span
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.trace.span import Span

from logboek.attributes import DATA_SUBJECT_ID, DATA_SUBJECT_ID_TYPE, PROCESSING_ACTIVITY_ID


class Processing:
    def __init__(self, span: Span) -> None:
        self._span: Span = span

    def set_data_subject(self, subject_id: str, sibject_id_type: str) -> None:
        self._span.set_attributes(
            {
                DATA_SUBJECT_ID: subject_id,
                DATA_SUBJECT_ID_TYPE: sibject_id_type,
            }
        )

    def successful(self) -> None:
        self._span.set_status(StatusCode.OK)

    def failed(self, reason: str) -> None:
        self._span.set_status(StatusCode.ERROR, reason)


class ProcessingOperator:
    def __init__(self, tracer: Tracer) -> None:
        self._tracer: Tracer = tracer

    def start_proccessing_from_foreign_operation(
        self, name: str, activity_id: str, headers: Mapping[str, str], end_on_exit: bool = True
    ) -> _GeneratorContextManager[Processing]:
        context = TraceContextTextMapPropagator().extract(headers)

        return self._start_proccessing(name, activity_id, context, end_on_exit)

    def start_proccessing(
        self, name: str, activity_id: str, end_on_exit: bool = True
    ) -> _GeneratorContextManager[Processing]:
        return self._start_proccessing(name, activity_id, None, end_on_exit)

    @contextmanager
    def _start_proccessing(
        self, name: str, activity_id: str, context: Context | None, end_on_exit: bool
    ) -> Generator[Processing]:
        attributes = {
            PROCESSING_ACTIVITY_ID: activity_id,
        }

        span = self._tracer.start_span(name, attributes=attributes, context=context)

        with use_span(span, end_on_exit):
            yield Processing(span)

Decorator​

Met een decorator kunnen we eenvoudig verwerkingen registreren.

logboek/decorators.py
import functools
from collections.abc import Callable
from typing import ParamSpec, TypeVar

from flask import request

from .flask import processing_operator

P = ParamSpec("P")
RT = TypeVar("RT")


def logboek_processing(
    name: str, activity_id: str, from_foreign: bool = False
) -> Callable[[Callable[P, RT]], Callable[P, RT | None]]:
    def decorator(func: Callable[P, RT]) -> Callable[P, RT | None]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> RT | None:
            op = processing_operator()

            if from_foreign:
                cm = op.start_proccessing_from_foreign_operation(
                    name, activity_id, request.headers, end_on_exit=True
                )
            else:
                cm = op.start_proccessing(name, activity_id, end_on_exit=True)

            with cm as processing:
                try:
                    return_value = func(*args, **kwargs)
                    processing.successful()
                    return return_value
                except Exception as e:
                    processing.failed(str(e))

            return None

        return wrapper

    return decorator

Applicatie​

Logboek configureren​

Voeg de Logboek-extentie toe en configureer de endpoint-URL.

app/__init__.py
from flask import Flask

from logboek.flask import Logboek

from .persons import persons

app = Flask(__name__)
app.config.from_mapping(
    APP_NAME="ldv-example-python",
    APP_VERSION="0.1.0",
    LOGBOEK={
        "endpoint": "127.0.0.1:4317",
        "endpoint_insecure": True,
    },
)

logboek = Logboek()
logboek.init_app(app)

app.register_blueprint(persons)

Verwerkingen definiëren​

Alles komt nu samen in dit Flask blueprint.

De gemarkeerde regels laten zien hoe de logboek_processing-decorator, en de ProcessingOperator-class en Processing-class gebruikt kunnen worden.

app/persons.py
from time import sleep

from flask import Blueprint

from logboek.decorators import logboek_processing
from logboek.flask import processing_operator

persons = Blueprint("persons", __name__, url_prefix="/persons")


@persons.route("/<bsn>/is-minimal-18-years-old")
@logboek_processing("check_18_or_over", "urn:ldv:activity:1407", from_foreign=True)
def is_minimal_18_years_old(bsn: str) -> dict[str, bool]:
    is_18_or_over = _person_is_18_or_over(bsn)

    return {"is_18_or_over": is_18_or_over}


def _person_is_18_or_over(bsn: str) -> bool:
    with processing_operator().start_proccessing("fetch_date_of_birth", "urn:ldv:activity:1550") as processing:
        processing.set_data_subject(bsn, "BSN")

        #  Simuleer het ophalen van de geboortedatum
        sleep(42 / 1000)

        processing.successful()

    return True

Applicatie starten​

Start nu de applicatie:

.venv/bin/flask run

Roep de API aan:

curl http://127.0.0.1:5000/persons/999990342/is-minimal-18-years-old

Logboek inzien​

Als je de OpenTelemetry viewer gebruikt kun je de twee verwerkingen in het logboek zien.