qonto-assistant/src/qonto_assistant/app.py

176 lines
6 KiB
Python
Raw Normal View History

from __future__ import annotations
from collections.abc import Callable
from contextlib import suppress
import logging
from uuid import uuid4
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.concurrency import run_in_threadpool
from qonto_assistant import __version__
from qonto_assistant.audit import AuditLogger
from qonto_assistant.auth import actor_claims_from_request
from qonto_assistant.config import Settings
from qonto_assistant.credentials import build_credential_provider
from qonto_assistant.errors import QontoAssistantError, UpstreamError
from qonto_assistant.policy import PolicyEngine
from qonto_assistant.qonto_client import FixtureQontoClient, QontoClient
from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
from qonto_assistant.service import CapabilityService
def create_app(
*,
settings: Settings | None = None,
service: CapabilityService | None = None,
audit_logger: AuditLogger | None = None,
rate_limiter: RateLimiter | None = None,
concurrency_limiter: ConcurrencyLimiter | None = None,
) -> FastAPI:
settings = settings or Settings.from_env()
audit_logger = audit_logger or AuditLogger()
service = service or _build_service(
settings=settings,
audit_logger=audit_logger,
rate_limiter=rate_limiter,
concurrency_limiter=concurrency_limiter,
)
app = FastAPI(title="qonto-assistant", version=__version__)
app.state.settings = settings
app.state.service = service
@app.exception_handler(QontoAssistantError)
def handle_qonto_error(_: Request, exc: QontoAssistantError) -> JSONResponse:
body = {"error_code": exc.error_code, "detail": exc.message}
if isinstance(exc, UpstreamError) and exc.upstream_status is not None:
body["upstream_status"] = exc.upstream_status
return JSONResponse(status_code=exc.status_code, content=body)
@app.get("/v1/health")
async def health() -> dict[str, str]:
return {
"status": "ok",
"service": settings.service_name,
"version": __version__,
"policy_file": str(settings.policy_file),
}
@app.get("/v1/accounts")
async def get_accounts(request: Request) -> JSONResponse:
claims = actor_claims_from_request(request, settings)
request_id = _request_id(request)
payload = await run_in_threadpool(service.get_accounts, claims=claims, request_id=request_id)
return JSONResponse(content=payload, headers={"X-Request-ID": request_id})
@app.get("/v1/transactions")
async def get_transactions(
request: Request,
account_slug: str | None = None,
page: int = 1,
page_size: int = 50,
window_days: int = 31,
status: str | None = "completed",
side: str | None = None,
) -> JSONResponse:
claims = actor_claims_from_request(request, settings)
request_id = _request_id(request)
payload = await run_in_threadpool(
service.list_transactions,
claims=claims,
request_id=request_id,
account_slug=account_slug,
page=page,
page_size=page_size,
window_days=window_days,
status=status,
side=side,
)
return JSONResponse(content=payload, headers={"X-Request-ID": request_id})
@app.get("/v1/snapshot")
async def get_snapshot(
request: Request,
window_days: int = 31,
page_size: int = 50,
) -> JSONResponse:
claims = actor_claims_from_request(request, settings)
request_id = _request_id(request)
payload = await run_in_threadpool(
service.get_snapshot,
claims=claims,
request_id=request_id,
window_days=window_days,
page_size=page_size,
)
return JSONResponse(content=payload, headers={"X-Request-ID": request_id})
@app.on_event("shutdown")
def shutdown_event() -> None:
with suppress(Exception):
service.client.close()
return app
def _request_id(request: Request) -> str:
existing = getattr(request.state, "request_id", None)
if existing:
return existing
request.state.request_id = request.headers.get("x-request-id", str(uuid4()))
return request.state.request_id
def _build_service(
*,
settings: Settings,
audit_logger: AuditLogger,
rate_limiter: RateLimiter | None,
concurrency_limiter: ConcurrencyLimiter | None,
) -> CapabilityService:
policy = PolicyEngine.from_file(
settings.policy_file,
required_scope=settings.required_scope,
enforce_scope=settings.enforce_scope,
)
if settings.qonto_fixture_dir is not None:
client = FixtureQontoClient(fixture_dir=settings.qonto_fixture_dir)
else:
credential_provider = build_credential_provider(settings)
client = QontoClient(
base_url=settings.qonto_base_url,
organization_path=settings.qonto_organization_path,
transactions_path=settings.qonto_transactions_path,
auth_mode=settings.qonto_auth_mode,
timeout_seconds=settings.qonto_timeout_seconds,
max_retries=settings.qonto_max_retries,
credential_provider=credential_provider,
)
return CapabilityService(
client=client,
policy=policy,
audit_logger=audit_logger,
rate_limiter=rate_limiter
or RateLimiter(
limit=settings.rate_limit_requests,
window_seconds=settings.rate_limit_window_seconds,
),
concurrency_limiter=concurrency_limiter
or ConcurrencyLimiter(limit=settings.max_concurrency),
)
def main() -> None:
settings = Settings.from_env()
logging.basicConfig(level=logging.INFO, format="%(message)s")
uvicorn.run(
"qonto_assistant.app:create_app",
factory=True,
host=settings.host,
port=settings.port,
reload=False,
)