[Python] Adding user info to azure application insights with FastAPI

I have code below and I would like user info like email so I can find in logs who did what on the authenticated endpoints. My plan is to do it through middleware. Is there any way to get user info in the middleware method dispatch, so I somehow can access it anywhere also outside the endpoints so I can apply it to my logging?

Main.py:

import os
from typing import Union
from fastapi import FastAPI, Security,Request
from starlette.middleware.base import BaseHTTPMiddleware
import uvicorn
from utils.azure.fast_api_auth import FastAPIAuth,FastAPIAuthOptions
from routers.router1 import router1_router
from routers.port import port_router
from enums import fast_api_metadata
from settings import settings

from data.context.dbx_context import DataBricksOptions
from data.repos.dbx_repo import DbxRepo
from settings import settings

import logging
from utils.azure.azure_logging import azure_logger
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

fast_api_options = FastAPIAuthOptions(
title="DNA PORT API",
tenant_id=settings.TENANT_ID,
openapi_tags=fast_api_metadata,
client_id=settings.fastapi_settings.APP_CLIENT_ID)

fast_api_auth = FastAPIAuth(fast_api_options)
app = fast_api_auth.get_fast_api_app()

# oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
try:
# Get the authorization token
if "Authorization" in request.headers:
token =Security(fast_api_auth.azure_scheme)
# You can access user info like:
# token.name - Full name
# token.preferred_username - Email
# token.oid - Object ID
# token.roles - User roles
request.state.user = token
response = await call_next(request)
return response
except Exception as e:
logging.error(f"Error in middleware: {str(e)}")
raise e

FastAPIInstrumentor().instrument_app(app)
app.add_middleware(AuthMiddleware)

app.include_router(
port_router,
prefix="/api/port",
dependencies=[Security(fast_api_auth.azure_scheme)],

tags=["port"]
)

@app.get("/log", dependencies=[Security(fast_api_auth.azure_scheme)])
async def log():

azure_logger.info("Segato5", extra={"custom_dimension": "Kam_value","test1": "val1"})
azure_logger.warning("Segato5", extra={"custom_dimension": "Kam_value","test1": "val1"})
azure_logger.error("Segato5", extra={"custom_dimension": "Kam_value","test1": "val1"})
azure_logger.debug("Segato5", extra={"custom_dimension": "Kam_value","test1": "val1"})
azure_logger.metric("segato2", 8)

return {"whoIsTheBest": "!!"}

@app.get("/", dependencies=[Security(fast_api_auth.azure_scheme)])
def read_root():
return {"Hello": "World"}

@app.get("/settings", dependencies=[Security(fast_api_auth.azure_scheme)])
def get_settings():
return settings

@app.get("/me", dependencies=[Security(fast_api_auth.azure_scheme)])
async def get_me(token: str = Security(fast_api_auth.azure_scheme)):
return {
"name": token.name,
"email": token.preferred_username,
"object_id": token.oid,
"roles": token.roles
}

if __name__ == '__main__':
uvicorn.run(
'main:app',
host="0.0.0.0",
port=8000,
reload=True,



log_level="info" )

fast_api_auth.py:enter code here

from ast import Str
from fastapi_azure_auth import SingleTenantAzureAuthorizationCodeBearer
from contextlib import asynccontextmanager
from fastapi import FastAPI
from typing import AsyncGenerator
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.cors import CORSMiddleware

class FastAPIAuthOptions:
def __init__(self,title: str, tenant_id: str,openapi_tags, client_id: str):
self.openapi_tags = openapi_tags
self.client_id = client_id
self.title = title
self.tenant_id = tenant_id

class FastAPIAuth:
def __init__(self, options: FastAPIAuthOptions):
self.options = options
self.azure_scheme = SingleTenantAzureAuthorizationCodeBearer(
app_client_id=options.client_id,
tenant_id=self.options.tenant_id,
scopes={
f'api://{self.options.client_id}/user_impersonation': 'Access API as user'
},
)

async def load_openid_config(self):
await self.azure_scheme.openid_config.load_config()


def get_fast_api_app(self):
app = FastAPI(
title=self.options.title,
openapi_url=f"/api/v1/openapi.json",
openapi_tags=self.options.openapi_tags,
swagger_ui_oauth2_redirect_url='/oauth2-redirect',
swagger_ui_init_oauth={
'usePkceWithAuthorizationCodeGrant': True,
'clientId': self.options.client_id,
'scopes': [f'api://{self.options.client_id}/user_impersonation'],
},
lifespan=self.lifespan
)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["*"] # Configure this based on your environment
)

app.add_middleware(
CORSMiddleware,
allow_origins=[str(origin) for origin in []],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)

return app

@asynccontextmanager
async def lifespan(self,app: FastAPI) -> AsyncGenerator[None, None]:
try:
await self.load_openid_config()
except Exception as e:
print(f"Failed to load Azure OpenID configuration: {str(e)}")
yield

Continue reading...