|
# retoor <retoor@molodetz.nl>
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
from typing import Literal
|
|
|
|
from fastapi import APIRouter, FastAPI
|
|
from fastapi.exceptions import HTTPException
|
|
|
|
from devplacepy_services.base.config import PROFILES, port_profile
|
|
from devplacepy_services.base.errors import error_response, sanitize_detail
|
|
from devplacepy_services.base.http import shutdown_pool, startup_pool
|
|
from devplacepy_services.base.middleware import (
|
|
InternalAuthMiddleware,
|
|
InternalCallCounterMiddleware,
|
|
RequestStatsMiddleware,
|
|
SanitizeErrorsMiddleware,
|
|
)
|
|
|
|
|
|
class BaseMicroservice:
|
|
name: str = ""
|
|
title: str = ""
|
|
default_port: int = 0
|
|
workers: int | Literal["auto"] = 1
|
|
stateful: bool = True
|
|
depends_on: list[str] = []
|
|
managed_services: list = []
|
|
use_background: bool = False
|
|
run_supervisor: bool = False
|
|
|
|
def __init__(self) -> None:
|
|
self.started_at = time.monotonic()
|
|
self.version = self._resolve_version()
|
|
|
|
def _resolve_version(self) -> str:
|
|
env_version = os.environ.get("DEVPLACE_VERSION", "").strip()
|
|
if env_version:
|
|
return env_version
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--short", "HEAD"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=2,
|
|
check=False,
|
|
)
|
|
if result.returncode == 0:
|
|
return result.stdout.strip() or "unknown"
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
pass
|
|
return "unknown"
|
|
|
|
def resolved_port(self) -> int:
|
|
if self.default_port:
|
|
return self.default_port
|
|
profile = port_profile()
|
|
return PROFILES[profile][self.name]
|
|
|
|
def apply_base_middleware(self, app: FastAPI, *, internal_auth: bool = False) -> None:
|
|
app.add_middleware(SanitizeErrorsMiddleware)
|
|
if internal_auth:
|
|
app.add_middleware(InternalAuthMiddleware)
|
|
app.add_middleware(InternalCallCounterMiddleware)
|
|
app.add_middleware(RequestStatsMiddleware)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(self, app: FastAPI):
|
|
app.state.service_name = self.name
|
|
self.started_at = time.monotonic()
|
|
await startup_pool()
|
|
background = None
|
|
if self.use_background:
|
|
from devplacepy.services.background import background
|
|
|
|
await background.start()
|
|
background = background
|
|
if self.managed_services:
|
|
from devplacepy.services.manager import service_manager
|
|
|
|
for svc in self.managed_services:
|
|
service_manager.register(svc)
|
|
service_manager.set_lock_owner(True)
|
|
if self.run_supervisor:
|
|
import asyncio
|
|
|
|
asyncio.get_running_loop().call_soon(service_manager.supervise)
|
|
yield
|
|
if self.managed_services:
|
|
from devplacepy.services.manager import service_manager
|
|
|
|
await service_manager.shutdown_all()
|
|
if background is not None:
|
|
await background.stop()
|
|
await shutdown_pool()
|
|
|
|
def create_app(self) -> FastAPI:
|
|
app = FastAPI(title=self.title, lifespan=self.lifespan)
|
|
app.state.service_name = self.name
|
|
|
|
@app.exception_handler(HTTPException)
|
|
async def http_exception_handler(_request, exc: HTTPException):
|
|
if isinstance(exc.detail, dict) and "error" in exc.detail and "code" in exc.detail:
|
|
return error_response(exc.status_code, exc.detail["error"], exc.detail["code"])
|
|
message, code = sanitize_detail(exc.detail, exc.status_code)
|
|
return error_response(exc.status_code, message, code)
|
|
|
|
return app
|
|
|
|
def build_app(self) -> FastAPI:
|
|
raise NotImplementedError
|
|
|
|
|
|
def build_standard_app(
|
|
service: BaseMicroservice,
|
|
*,
|
|
routers: list[tuple[APIRouter, str] | tuple[APIRouter]] | None = None,
|
|
internal_auth: bool = False,
|
|
) -> FastAPI:
|
|
from devplacepy_services.base.health import health_router
|
|
|
|
app = service.create_app()
|
|
service.apply_base_middleware(app, internal_auth=internal_auth)
|
|
app.include_router(health_router(service))
|
|
for entry in routers or []:
|
|
if len(entry) == 2:
|
|
app.include_router(entry[0], prefix=entry[1])
|
|
else:
|
|
app.include_router(entry[0])
|
|
return app |