73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
"""This module implements API"""
|
|
|
|
import asyncio
|
|
import traceback
|
|
import time
|
|
import html
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from mab import MatrixBot
|
|
import uvicorn
|
|
|
|
from datatypes import AppConfig
|
|
from database import Database
|
|
|
|
import web_routes
|
|
import web_middleware
|
|
|
|
class Web:
|
|
#
|
|
# PRIVATE
|
|
#
|
|
async def _cb_exception(self, request: Request, exc: Exception):
|
|
traceback.print_exception(exc)
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"detail": "Internal Server Error"}
|
|
)
|
|
|
|
#
|
|
# PUBLIC
|
|
#
|
|
def __init__(self, app_config: AppConfig, bot: MatrixBot, db: Database):
|
|
self._api = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
|
|
self._api.state.database = db
|
|
self._api.state.matrix = bot
|
|
self._api.state.app_config = app_config
|
|
|
|
self._api.include_router(web_routes.router)
|
|
self._api.add_middleware(web_middleware.AuthMiddleware, ["/api"])
|
|
self._api.add_middleware(web_middleware.BanCheckMiddleware, ["/api"])
|
|
self._api.add_middleware(web_middleware.RealIpResolver)
|
|
self._api.add_exception_handler(Exception, self._cb_exception)
|
|
|
|
self._server_config = uvicorn.Config(
|
|
self._api,
|
|
host=app_config.web_ip,
|
|
port=app_config.web_port
|
|
)
|
|
self._server: uvicorn.Server | None = None
|
|
self._server_task: asyncio.Task | None = None
|
|
|
|
async def start(self) -> None:
|
|
"""Start the API."""
|
|
if self._server is not None or self._server_task is not None:
|
|
raise RuntimeError("The server is already started")
|
|
self._server = uvicorn.Server(self._server_config)
|
|
self._server_task = asyncio.create_task(self._server.serve())
|
|
|
|
async def stop(self) -> None:
|
|
"""Stop the API server."""
|
|
if self._server is None or self._server_task is None:
|
|
raise RuntimeError("The server is not started yet")
|
|
self._server_task.cancel()
|
|
try:
|
|
await self._server_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except:
|
|
traceback.print_exc()
|
|
self._server_task = None
|
|
self._server = None |