Aleksei Sokol
53f14a8624
All checks were successful
Run linters on applied template / Python 3.13 lint and build (push) Successful in 1m40s
Changes: - put ObservabilityMiddleware before ExceptionHandlerMiddleware to avoid repetative code - add application startup and last metrics update metrics along with CPU usage metric and threads count - move host and port to new uvicorn section at config along with new reload and forwarded_allow_ips - add request_id and remove trace_id/span_id generation if tracing is disabled - move logging logic from utils to observability - pass trace_id/span_id in HEX form
118 lines
3.9 KiB
Django/Jinja
118 lines
3.9 KiB
Django/Jinja
"""Debug and testing endpoints are defined here. They should be included in router only in debug-mode."""
|
|
|
|
import asyncio
|
|
from typing import Literal
|
|
|
|
import aiohttp
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from opentelemetry import trace
|
|
from starlette import status
|
|
|
|
from {{project_slug}}.dependencies import auth_dep
|
|
from {{project_slug}}.observability.utils import get_tracing_headers
|
|
from {{project_slug}}.schemas import PingResponse
|
|
|
|
from .routers import debug_errors_router
|
|
|
|
_tracer = trace.get_tracer(__name__)
|
|
|
|
|
|
class DebugException(Exception):
|
|
pass
|
|
|
|
|
|
class DebugExceptionWithParams(Exception):
|
|
def __init__(self, status_code: int, message: str):
|
|
self.status_code = status_code
|
|
self.message = message
|
|
|
|
|
|
@debug_errors_router.get(
|
|
"/status_code",
|
|
response_model=PingResponse,
|
|
status_code=status.HTTP_200_OK,
|
|
)
|
|
async def get_status_code(status_code: int, as_exception: bool = False):
|
|
"""Return given status code. If `as_exception` is set to True, return as HTTPException."""
|
|
if as_exception:
|
|
raise HTTPException(
|
|
status_code=status_code, detail=f"debugging with status code {status_code} as http_exception"
|
|
)
|
|
return JSONResponse(PingResponse().model_dump(), status_code=status_code)
|
|
|
|
|
|
@debug_errors_router.get(
|
|
"/exception",
|
|
response_model=PingResponse,
|
|
status_code=status.HTTP_200_OK,
|
|
)
|
|
async def get_exception(error_type: Literal["RuntimeError", "DebugException", "DebugExceptionWithParams"]):
|
|
"""Raise exception: `RuntimeError`, `DebugException` or `DebugExceptionWithParams` depending on given parameter."""
|
|
if error_type == "DebugException":
|
|
raise DebugException()
|
|
if error_type == "DebugExceptionWithParams":
|
|
raise DebugExceptionWithParams(522, "Message goes here")
|
|
raise RuntimeError("That's how runtime errors look like")
|
|
|
|
|
|
@debug_errors_router.get(
|
|
"/tracing_check",
|
|
status_code=status.HTTP_200_OK,
|
|
)
|
|
async def tracing_check():
|
|
"""Add event to span, and sleep 2 seconds inside an inner span."""
|
|
span = trace.get_current_span()
|
|
|
|
span.add_event("successful log entry", attributes={"parameters": "go here"})
|
|
with _tracer.start_as_current_span("long operation") as inner_span:
|
|
inner_span.add_event("going to sleep")
|
|
await asyncio.sleep(2)
|
|
inner_span.add_event("woke up")
|
|
|
|
return JSONResponse({"finished": "ok"})
|
|
|
|
|
|
@debug_errors_router.get(
|
|
"/inner_get_tracing",
|
|
status_code=status.HTTP_200_OK,
|
|
)
|
|
async def inner_get(host: str = "http://127.0.0.1:8080"):
|
|
"""Perform GET request with span proxying to get more complicated trace."""
|
|
|
|
def perform_get(session: aiohttp.ClientSession) -> asyncio.Task:
|
|
return asyncio.create_task(
|
|
session.get(
|
|
"/api/debug/tracing_check",
|
|
headers=get_tracing_headers(),
|
|
)
|
|
)
|
|
|
|
with _tracer.start_as_current_span("inner request"):
|
|
inner_results = []
|
|
async with aiohttp.ClientSession(host) as session:
|
|
requests_futures = [perform_get(session) for _ in range(2)]
|
|
results: list[aiohttp.ClientResponse] = await asyncio.gather(*requests_futures)
|
|
results.append(await perform_get(session))
|
|
|
|
for response in results:
|
|
inner_results.append(
|
|
{
|
|
"inner_response": await response.json(),
|
|
"headers": dict(response.headers.items()),
|
|
"status_code": response.status,
|
|
}
|
|
)
|
|
|
|
return JSONResponse({"inner_results": inner_results})
|
|
|
|
|
|
@debug_errors_router.get(
|
|
"/authentication_info",
|
|
status_code=status.HTTP_200_OK,
|
|
)
|
|
async def authentication_info(auth: auth_dep.AuthenticationData = Depends(auth_dep.from_request)):
|
|
"""Check authentication data from `Authorization` and `X-API-Key` headers."""
|
|
|
|
return JSONResponse({"auth_data": {"x-api-key": auth.api_key, "jwt_payload": auth.jwt_payload}})
|