Initial commit

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

{{project_slug}}/utils/observability.py

@@ -0,0 +1,104 @@
+"""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),
+    }

{{project_slug}}/utils/secrets.py

@@ -0,0 +1,49 @@
+"""Basic functionality to work with sensitive data in configs is defined here."""
+
+import os
+import re
+from typing import Any
+
+import yaml
+
+_env_re = re.compile(r"^!env\((?P<env_var>.+)\)$")
+
+
+class SecretStr(str):
+    """String value which returns "<REDACTED>" on str() and repr() calls.
+
+    If given value matches pattern `^!env\\(.+\\)$` then try to get value from environment variables by the given name.
+
+    To get a value inside one should use `get_secret_value` method.
+    """
+
+    def __new__(cls, other: Any):
+        if isinstance(other, SecretStr):
+            return super().__new__(cls, other.get_secret_value())
+        if isinstance(other, str):
+            if (m := _env_re.match(other)) is not None:
+                env_var = m.group("env_var")
+                if env_var in os.environ:
+                    other = os.environ[env_var]
+                else:
+                    print(
+                        f"CAUTION: secret variable '{other}' looks like a mapping from env,"
+                        f" but no '{env_var} value is found'"
+                    )
+            return super().__new__(cls, other)
+
+    def __str__(self) -> str:
+        return "<REDACTED>"
+
+    def __repr__(self) -> str:
+        return "'<REDACTED!r>'"
+
+    def get_secret_value(self) -> str:
+        return super().__str__()
+
+
+def representSecretStrYAML(dumper: yaml.Dumper, s: SecretStr):
+    return dumper.represent_str(s.get_secret_value())
+
+
+yaml.add_representer(SecretStr, representSecretStrYAML)