Разрабатываем приложение на Python, которое слушает очередь (принимает сообщения и выполняет работу над ним). Требования devopsов: наличие healthcheck у приложения. Healthcheck действительный, то есть отдаём значение 200, если приложение и инфраструктура доступны.
Healthcheck планируем делать на FastAPI/aiohttp в виде одного ендпоинта с отдачей статуса.
Вопрос: насколько верно делать class Application(Object) с консьюмерами плюс добавить внутрь приложения запуск FastAPI instance с запуском через uvicorn?
Не будет ли проблем с event loop при переключении между обработкой сообщений RabbitMQ и FastAPI проверкой healthcheck?
Нечто подобное:
class HTTPServerManager:
settings: Settings
api_server: Optional[Server] = None
additional_router: Optional[APIRouter] = None
live_probe_service: LiveProbeService = None
LOGGING_EXCLUDE_ENDPOINT = ['/metrics', '/docs', '/health/live', '/openapi.json', ]
def __post_init__(self):
self.port = self.settings.app_http_server_port
self.http_handler = self.settings.app_http_server_handler
self.dev_mode = self.settings.dev_mode
self.loop = asyncio.get_event_loop()
self.server_application = ServerApplication(self.settings, self.live_probe_service)
self.init_api_server()
async def run_startup(self) -> None:
self.init_api_server()
async def run_shutdown(self) -> None:
if self.api_server is not None and self.api_server.started:
await self.api_server.shutdown()
async def serve(self) -> None:
await self.api_server.serve()
def init_api_server(self) -> None:
if self.additional_router:
self.server_application.include_router(self.additional_router)
config = Config(
app=self.server_application,
loop=self.http_handler,
host='0.0.0.0',
port=self.port,
log_config=get_uvicorn_config(HTTPServerManager.LOGGING_EXCLUDE_ENDPOINT),
access_log=True,
)
self.api_server = Server(config)
Базовое приложение:
class Application(BaseApplication):
def __init__(self) -> None:
self.container_manager = ContainerManager(self.container_cls)
self.container = self.container_manager.get_container()
...
self.loop = asyncio.get_event_loop()
self.http_transport = self.container.get(HTTPServerManager)
self.live_probe_service = self.container.get(LiveProbeService)
def run(self) -> None:
try:
self.logger.info(f'Application {self.app_name} is starting')
self.loop.run_until_complete(self.container_manager.run_startup())
self.logger.info('Application container created')
gather_tasks = asyncio.gather(
self.queue_transport.run(),
self.http_transport.serve(),
)
self.loop.run_until_complete(gather_tasks)
self.loop.run_forever()
except:
...
Средний
Простой
