Aleksei Sokol
6e9ed87ec4
Some checks failed
Run linters on applied template / Python 3.13 lint and build (push) Failing after 39s
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
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
"""Observability helper functions are defined here."""
|
|
|
|
import logging
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
import structlog
|
|
from opentelemetry import trace
|
|
|
|
LoggingLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
|
|
|
|
|
def configure_logging(
|
|
log_level: LoggingLevel, files: dict[str, LoggingLevel] | None = None, root_logger_level: LoggingLevel = "INFO"
|
|
) -> structlog.stdlib.BoundLogger:
|
|
level_name_mapping = {
|
|
"DEBUG": logging.DEBUG,
|
|
"INFO": logging.INFO,
|
|
"WARNING": logging.WARNING,
|
|
"ERROR": logging.ERROR,
|
|
"CRITICAL": logging.CRITICAL,
|
|
}
|
|
files = files or {}
|
|
|
|
structlog.configure(
|
|
processors=[
|
|
structlog.contextvars.merge_contextvars,
|
|
structlog.stdlib.add_log_level,
|
|
structlog.stdlib.add_logger_name,
|
|
structlog.processors.TimeStamper(fmt="iso"),
|
|
structlog.processors.StackInfoRenderer(),
|
|
structlog.processors.format_exc_info,
|
|
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
|
|
],
|
|
logger_factory=structlog.stdlib.LoggerFactory(),
|
|
wrapper_class=structlog.stdlib.BoundLogger,
|
|
cache_logger_on_first_use=True,
|
|
)
|
|
|
|
logger: structlog.stdlib.BoundLogger = structlog.get_logger("main")
|
|
logger.setLevel(level_name_mapping[log_level])
|
|
|
|
console_handler = logging.StreamHandler(sys.stderr)
|
|
console_handler.setFormatter(
|
|
structlog.stdlib.ProcessorFormatter(processor=structlog.dev.ConsoleRenderer(colors=True))
|
|
)
|
|
|
|
root_logger = logging.getLogger()
|
|
root_logger.addHandler(console_handler)
|
|
|
|
for filename, level in files.items():
|
|
try:
|
|
Path(filename).parent.mkdir(parents=True, exist_ok=True)
|
|
except Exception as exc:
|
|
print(f"Cannot create directory for log file {filename}, application will crash most likely. {exc!r}")
|
|
file_handler = logging.FileHandler(filename=filename, encoding="utf-8")
|
|
file_handler.setFormatter(structlog.stdlib.ProcessorFormatter(processor=structlog.processors.JSONRenderer()))
|
|
file_handler.setLevel(level_name_mapping[level])
|
|
root_logger.addHandler(file_handler)
|
|
|
|
root_logger.setLevel(root_logger_level)
|
|
|
|
return logger
|
|
|
|
|
|
def get_handler_from_path(path: str) -> str:
|
|
parts = path.split("/")
|
|
return "/".join(part if not part.rstrip(".0").isdigit() else "*" for part in parts)
|
|
|
|
|
|
class URLsMapper:
|
|
"""Helper to change URL from given regex pattern to the given static value.
|
|
|
|
For example, with map {"/api/debug/.*": "/api/debug/*"} all requests with URL starting with "/api/debug/"
|
|
will be placed in path "/api/debug/*" in metrics.
|
|
"""
|
|
|
|
def __init__(self, urls_map: dict[str, str]):
|
|
self._map: dict[re.Pattern, str] = {}
|
|
|
|
for pattern, value in urls_map.items():
|
|
self.add(pattern, value)
|
|
|
|
def add(self, pattern: str, mapped_to: str) -> None:
|
|
"""Add entry to the map. If pattern compilation is failed, ValueError is raised."""
|
|
regexp = re.compile(pattern)
|
|
self._map[regexp] = mapped_to
|
|
|
|
def map(self, url: str) -> str:
|
|
"""Check every map entry with `re.match` and return matched value. If not found, return original string."""
|
|
for regexp, mapped_to in self._map.items():
|
|
if regexp.match(url) is not None:
|
|
return mapped_to
|
|
return url
|
|
|
|
|
|
def get_span_headers() -> dict[str, str]:
|
|
ctx = trace.get_current_span().get_span_context()
|
|
return {
|
|
"X-Span-Id": str(ctx.span_id),
|
|
"X-Trace-Id": str(ctx.trace_id),
|
|
}
|