mirror of
https://github.com/Balshgit/gpt_chat_bot.git
synced 2025-09-11 22:30:41 +03:00
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
import asyncio
|
|
import sys
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.responses import UJSONResponse
|
|
from loguru import logger
|
|
|
|
from app.core.bot import BotApplication, BotQueue
|
|
from app.routers import api_router
|
|
from settings.config import Settings, get_settings
|
|
|
|
logger.remove()
|
|
logger.add(
|
|
sink=sys.stdout,
|
|
colorize=True,
|
|
level='DEBUG',
|
|
format="<cyan>{time:DD.MM.YYYY HH:mm:ss}</cyan> | <level>{level}</level> | <magenta>{message}</magenta>",
|
|
)
|
|
|
|
|
|
class Application:
|
|
def __init__(self, settings: Settings, bot_app: BotApplication) -> None:
|
|
self.app = FastAPI(
|
|
title='Health check bot',
|
|
description='Bot which check all services are working',
|
|
version='0.0.3',
|
|
docs_url=f'{settings.WEBHOOK_PATH}/docs',
|
|
redoc_url=f'{settings.WEBHOOK_PATH}/redocs',
|
|
openapi_url=f'{settings.WEBHOOK_PATH}/api/openapi.json',
|
|
default_response_class=UJSONResponse,
|
|
)
|
|
self.app.state.settings = settings
|
|
self.app.state.queue = BotQueue(bot_app=bot_app.application)
|
|
self.bot_app = bot_app
|
|
self.settings = settings
|
|
|
|
self.app.include_router(api_router)
|
|
self.configure_hooks()
|
|
|
|
@property
|
|
def fastapi_app(self) -> FastAPI:
|
|
return self.app
|
|
|
|
def configure_hooks(self) -> None:
|
|
if self.bot_app.start_with_webhook:
|
|
self.app.add_event_handler("startup", self.bot_app.polling)
|
|
else:
|
|
self.app.add_event_handler("startup", self._on_start_up)
|
|
|
|
self.app.add_event_handler("shutdown", self._on_shutdown)
|
|
|
|
async def _on_start_up(self) -> None:
|
|
await self.bot_app.set_webhook()
|
|
loop = asyncio.get_event_loop()
|
|
loop.create_task(self.app.state.queue.get_updates_from_queue())
|
|
|
|
async def _on_shutdown(self) -> None:
|
|
await asyncio.gather(self.bot_app.delete_webhook(), self.bot_app.shutdown())
|
|
|
|
|
|
def create_app(settings: Settings | None = None) -> FastAPI:
|
|
settings = settings or get_settings()
|
|
bot_app = BotApplication(settings=settings, start_with_webhook=settings.START_WITH_WEBHOOK)
|
|
|
|
return Application(settings=settings, bot_app=bot_app).fastapi_app
|
|
|
|
|
|
def main() -> None:
|
|
import uvicorn
|
|
|
|
app = create_app()
|
|
|
|
"""Entrypoint of the application."""
|
|
uvicorn.run(
|
|
"app.main:create_app",
|
|
workers=app.state.settings.WORKERS_COUNT,
|
|
host=app.state.settings.APP_HOST,
|
|
port=app.state.settings.APP_PORT,
|
|
# reload=app.state.settings.RELOAD, # noqa: E800 remove reload for debug
|
|
factory=True,
|
|
)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|