Aleksei Sokol
5dd68b7114
Some checks failed
Run linters on applied template / Python 3.13 lint and build (push) Failing after 34s
This is a FastAPI backend microservice template used with `copier` utility. Features of applied template are: - Configuration file processing logic - Metrics and tracing (both optional) configuration available - Debug endpoints - Database migration commands, prepared Alembic environment - Database usage example in ping_db endpoint - gitea sanity check pipeline
115 lines
3.6 KiB
Django/Jinja
115 lines
3.6 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 HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
from opentelemetry import trace
|
|
from starlette import status
|
|
|
|
from {{project_slug}}.schemas import PingResponse
|
|
from {{project_slug}}.utils.observability import get_span_headers
|
|
|
|
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_span_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})
|