feat(api): add monitoring feature with track_events use case

Author: eoaksnes

api/.openapi.json

@@ -5,13 +5,15 @@ description = "API for Template Fastapi React"
 requires-python = ">=3.14"
 readme = "../README.md"
 dependencies = [
-  "azure-monitor-opentelemetry>=1.6.5",
+  "azure-identity>=1.19.0",
+  "azure-monitor-opentelemetry>=1.8.7",
+  "azure-monitor-opentelemetry-exporter>=1.0.0b49",
   "cachetools>=7.1.1",
   "fastapi[standard]>=0.136.1",
   "httpx>=0.28",
-  "opentelemetry-instrumentation-fastapi>=0.62b1",
-  "pydantic>=2.13.4",
-  "pydantic-settings>=2.14.1",
+  "opentelemetry-instrumentation-fastapi>=0.61b0",
+  "pydantic>=2.13.3",
+  "pydantic-settings>=2.14.0",
   "pyjwt>=2.8.0",
   "pymongo>=4.17.0",
 ]
@@ -24,19 +26,6 @@ dev = [
   "types-cachetools>=7.0.0.20260503",
 ]
 
-[tool.uv]
-# opentelemetry-sdk 1.39 removed `LogData` from `opentelemetry.sdk._logs`,
-# which breaks azure-monitor-opentelemetry-exporter (still on the old API
-# as of 1.0.0b45). Override the transitive resolution to keep us on the
-# last working SDK line until the exporter migrates to LogRecord. Pinning
-# in `dependencies` doesn't work — the instrumentation packages declare
-# matching `semantic-conventions` versions that drag the SDK back up.
-override-dependencies = [
-  "opentelemetry-sdk<1.39",
-  "opentelemetry-api<1.39",
-  "opentelemetry-semantic-conventions<0.60b0",
-]
-
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

api/pyproject.toml

@@ -5,6 +5,7 @@
 from app.common import LocalLoggerMiddleware, responses
 from app.config import config
 from app.features.health_check import router as health_check_router
+from app.features.monitoring import router as monitoring_router
 from app.features.todo import router as todo_router
 from app.features.whoami import router as whoami_router
 
@@ -28,6 +29,7 @@ def create_app() -> FastAPI:
     public_routes.include_router(health_check_router)
 
     authenticated_routes = APIRouter()
+    authenticated_routes.include_router(monitoring_router)
     authenticated_routes.include_router(todo_router)
     authenticated_routes.include_router(whoami_router)
 
@@ -53,7 +55,19 @@ def create_app() -> FastAPI:
         from azure.monitor.opentelemetry import configure_azure_monitor
         from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 
-        configure_azure_monitor(connection_string=config.APPINSIGHTS_CONSTRING, logger_name="API")
+        kwargs: dict[str, object] = {
+            "connection_string": config.APPINSIGHTS_CONSTRING,
+            "logger_name": "API",
+        }
+        if config.has_azure_service_principal:
+            from azure.identity import ClientSecretCredential
+
+            kwargs["credential"] = ClientSecretCredential(
+                tenant_id=config.AZURE_TENANT_ID,
+                client_id=config.OAUTH_CLIENT_ID,
+                client_secret=config.OAUTH_CLIENT_SECRET.get_secret_value(),
+            )
+        configure_azure_monitor(**kwargs)
         FastAPIInstrumentor.instrument_app(app, excluded_urls="healthcheck")
 
     app.include_router(authenticated_routes, dependencies=[Security(auth_with_jwt)])

api/src/app/app.py

@@ -29,6 +29,13 @@ def get_JWK_client() -> jwt.PyJWKClient:
 
 
 def auth_with_jwt(jwt_token: str = Security(oauth2_scheme)) -> User:
+    """Validate the caller's id_token (Authorization header) and return a `User`.
+
+    oauth2-proxy injects the id_token as `Authorization: Bearer ...`. The
+    id_token's `aud` claim is the OIDC client app id (`OAUTH_AUDIENCE`).
+    Roles are read from the `roles` claim, populated by the API app's
+    optional claims with `emit_as_roles` (so group GUIDs land in `roles`).
+    """
     if not config.AUTH_ENABLED:
         return default_user
     if not jwt_token:
@@ -37,8 +44,12 @@ def auth_with_jwt(jwt_token: str = Security(oauth2_scheme)) -> User:
     try:
         payload = jwt.decode(jwt_token, key, algorithms=["RS256"], audience=config.OAUTH_AUDIENCE)
         if config.MICROSOFT_AUTH_PROVIDER in payload["iss"]:
-            # Azure AD uses an oid string to uniquely identify users. Each user has a unique oid value.
-            user = User(user_id=payload["oid"], **payload)
+            user = User(
+                user_id=payload["oid"],
+                full_name=payload.get("name"),
+                email=payload.get("preferred_username") or payload.get("upn"),
+                roles=payload.get("roles", []),
+            )
         else:
             user = User(user_id=payload["sub"], **payload)
     except jwt.exceptions.InvalidTokenError as error:

api/src/app/authentication/authentication.py

@@ -1,4 +1,4 @@
-from pydantic import Field
+from pydantic import Field, SecretStr
 from pydantic_settings import BaseSettings
 
 from app.authentication.models import User
@@ -32,10 +32,16 @@ class Config(BaseSettings):
     OAUTH_TOKEN_ENDPOINT: str = ""
     OAUTH_AUTH_ENDPOINT: str = ""
     OAUTH_CLIENT_ID: str = ""
+    OAUTH_CLIENT_SECRET: SecretStr = SecretStr("")
     OAUTH_AUTH_SCOPE: str = ""
     OAUTH_AUDIENCE: str = ""
+    AZURE_TENANT_ID: str = ""
     MICROSOFT_AUTH_PROVIDER: str = "login.microsoftonline.com"
 
+    @property
+    def has_azure_service_principal(self) -> bool:
+        return bool(self.AZURE_TENANT_ID and self.OAUTH_CLIENT_ID and self.OAUTH_CLIENT_SECRET.get_secret_value())
+
     @property
     def log_level(self) -> str:
         """Returns LOGGER_LEVEL as a (lower case) string."""

api/src/app/config.py

@@ -0,0 +1,3 @@
+from app.features.monitoring.monitoring_feature import router
+
+__all__ = ["router"]

api/src/app/features/monitoring/__init__.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter, Depends
+
+from app.authentication.authentication import auth_with_jwt
+from app.authentication.models import User
+from app.common.exception_handlers import ExceptionHandlingRoute
+from app.features.monitoring.use_cases.track_events import Event, TelemetryResult, track_events_use_case
+
+router = APIRouter(tags=["monitoring"], prefix="/monitoring", route_class=ExceptionHandlingRoute)
+
+
+@router.post("/v2/track", operation_id="track")
+async def track(events: list[Event], user: User = Depends(auth_with_jwt)) -> TelemetryResult | None:
+    return track_events_use_case(events=events, user=user)

api/src/app/features/monitoring/monitoring_feature.py

@@ -0,0 +1,123 @@
+import datetime
+import functools
+from http import HTTPStatus
+from typing import Any, NotRequired, TypedDict
+
+from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
+from azure.monitor.opentelemetry.exporter._connection_string_parser import ConnectionStringParser
+from azure.monitor.opentelemetry.exporter._generated.exporter.models import MonitorBase, TelemetryItem
+
+from app.authentication.models import User
+from app.common.logger import logger
+from app.config import config
+
+
+@functools.lru_cache(maxsize=1)
+def get_trace_exporter() -> AzureMonitorTraceExporter:
+    kwargs: dict[str, Any] = {
+        "connection_string": config.APPINSIGHTS_CONSTRING,
+        "logger_name": "API",
+    }
+    if config.has_azure_service_principal:
+        from azure.identity import ClientSecretCredential
+
+        kwargs["credential"] = ClientSecretCredential(
+            tenant_id=config.AZURE_TENANT_ID,
+            client_id=config.OAUTH_CLIENT_ID,
+            client_secret=config.OAUTH_CLIENT_SECRET.get_secret_value(),
+        )
+    return AzureMonitorTraceExporter(**kwargs)
+
+
+class EventData(TypedDict):
+    ver: int
+    properties: dict[str, Any]
+    name: str
+    url: NotRequired[str]
+    measurements: NotRequired[dict[str, float]]
+
+
+class MetricsData(TypedDict):
+    ver: int
+    metrics: list[dict[str, Any]]
+    properties: dict[str, Any]
+
+
+class ExceptionData(TypedDict):
+    ver: int
+    exceptions: list[dict[str, Any]]
+    properties: dict[str, Any]
+
+
+class EventBase(TypedDict):
+    baseType: str
+    baseData: EventData | MetricsData | ExceptionData
+
+
+class Event(TypedDict):
+    name: str
+    tags: dict[str, str]
+    time: datetime.datetime
+    data: EventBase
+
+
+class TelemetryErrorDetails(TypedDict):
+    index: int
+    statusCode: int
+    message: str
+
+
+class TelemetryResult(TypedDict):
+    items_received: int
+    items_accepted: int
+    errors: list[TelemetryErrorDetails]
+
+
+def track_events_use_case(user: User, events: list[Event]) -> TelemetryResult | None:
+    if connection_string := config.APPINSIGHTS_CONSTRING:
+        instrumentation_key = ConnectionStringParser(connection_string=connection_string).instrumentation_key
+    else:
+        logger.warning("Environment variable 'APPINSIGHTS_CONSTRING' is unset; Unable to send to Azure Monitor")
+        return TelemetryResult(
+            items_accepted=0,
+            items_received=len(events),
+            errors=[
+                TelemetryErrorDetails(
+                    index=idx,
+                    statusCode=HTTPStatus.SERVICE_UNAVAILABLE,
+                    message="No Applications Insight connection string provided",
+                )
+                for idx in range(len(events))
+            ],
+        )
+    telemetry_items = [
+        TelemetryItem(
+            name=event["name"],
+            instrumentation_key=instrumentation_key,
+            tags=event["tags"],
+            time=event["time"],
+            data=MonitorBase(event["data"]),
+        )
+        for event in events
+    ]
+    trace_exporter = get_trace_exporter()
+    try:
+        result = trace_exporter.client.track(telemetry_items)
+        return TelemetryResult(
+            items_received=result.items_received or len(events),
+            items_accepted=result.items_accepted or 0,
+            errors=[
+                TelemetryErrorDetails(
+                    index=error.index or 0,
+                    statusCode=error.status_code or 0,
+                    message=error.message or "",
+                )
+                for error in (result.errors or [])
+            ],
+        )
+    except Exception as e:
+        logger.error(f"Failed to track events: {e}")
+    return None
+
+
+__all__ = ["track_events_use_case", "Event", "TelemetryResult"]

api/src/app/features/monitoring/use_cases/track_events/__init__.py

@@ -24,6 +24,7 @@ class AddTodoResponse(BaseModel):
             "examples": ["vytxeTZskVKR7C7WgdSP3d"],
         }
     )
+    user_id: str
     title: str = Field(
         json_schema_extra={
             "examples": ["Read about clean architecture"],
@@ -33,7 +34,12 @@ class AddTodoResponse(BaseModel):
 
     @classmethod
     def from_entity(cls, todo_item: TodoItem) -> Self:
-        return cls(id=todo_item.id, title=todo_item.title, is_completed=todo_item.is_completed)
+        return cls(
+            id=todo_item.id,
+            user_id=todo_item.user_id,
+            title=todo_item.title,
+            is_completed=todo_item.is_completed,
+        )
 
 
 def add_todo_use_case(

api/src/app/features/todo/use_cases/add_todo.py

@@ -10,12 +10,18 @@
 
 class GetTodoAllResponse(BaseModel):
     id: str
+    user_id: str
     title: str
     is_completed: bool
 
     @classmethod
     def from_entity(cls, todo_item: TodoItem) -> Self:
-        return cls(id=todo_item.id, title=todo_item.title, is_completed=todo_item.is_completed)
+        return cls(
+            id=todo_item.id,
+            user_id=todo_item.user_id,
+            title=todo_item.title,
+            is_completed=todo_item.is_completed,
+        )
 
 
 # Telemetry example: Initialize a span that will be used to log telemetry data