src/initializer.py - master - MonkaS/logistic-erp-back

src/initializer.py
80 строк3 KB
MonkaS
first_commit
25 май 2026, 23:38
25 май 2026, 23:38bc05d42
from contextlib import asynccontextmanager
import logging

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.middleware.cors import CORSMiddleware

from api.router import API_ROUTER
from core.exception_handlers import (
    app_exception_handler,
    http_exception_handler,
    request_validation_exception_handler,
    unknown_exception_handler, object_does_not_exist_exception_handler,
)
from core.exceptions import AppException, ObjectDoesNotExistException
from core.middlewares import LoggingMiddleware
from repositories.db.manager import db_manager
from settings import settings
import redis.asyncio as redis

logger = logging.getLogger(__name__)


def add_routers(app: FastAPI):
    app.include_router(API_ROUTER, prefix=settings.APP_URL_PREFIX)


def add_exception_handlers(app: FastAPI):
    app.add_exception_handler(AppException, app_exception_handler)
    app.add_exception_handler(StarletteHTTPException, http_exception_handler)
    app.add_exception_handler(RequestValidationError, request_validation_exception_handler)
    app.add_exception_handler(ObjectDoesNotExistException, object_does_not_exist_exception_handler)
    app.add_exception_handler(Exception, unknown_exception_handler)


def add_middlewares(app: FastAPI):
    domain_regex = f'https?://.*\\.{settings.CORS_DOMAIN.replace(".", "\\.")}' if settings.CORS_DOMAIN is not None else None
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.CORS_ALLOWED_HOSTS,
        allow_origin_regex=domain_regex,
        allow_credentials=True,
        allow_methods=['*'],
        allow_headers=['*'],
    )
    app.add_middleware(LoggingMiddleware)


@asynccontextmanager
async def initialize(app: FastAPI):
    db_manager.init(settings.create_dsn())
    redis_client = None
    try:
        redis_client = redis.from_url(settings.create_redis_url(), decode_responses=True)
        await redis_client.ping()
    except Exception:
        if redis_client is not None:
            await redis_client.close()
        logger.warning('Redis is unavailable, continuing without it', exc_info=True)
        redis_client = None
    app.state.redis = redis_client

    yield

    redis_client = getattr(app.state, 'redis', None)
    if redis_client is not None:
        await redis_client.close()
    await db_manager.close()


def create_app() -> FastAPI:
    app = FastAPI(
        lifespan=initialize,
        root_path='/api',
    )
    add_middlewares(app)
    add_exception_handlers(app)
    add_routers(app)
    return app