mirror of
https://github.com/kuitoi/kuitoi-Server.git
synced 2025-08-17 08:15:42 +00:00
106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
import asyncio
|
|
from asyncio import CancelledError
|
|
|
|
import uvicorn
|
|
from fastapi import FastAPI, Request, HTTPException
|
|
from fastapi.exceptions import RequestValidationError
|
|
from starlette import status
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
from starlette.responses import JSONResponse
|
|
|
|
import core.utils
|
|
from . import utils
|
|
|
|
# from .models import SecretKey
|
|
|
|
web_app = FastAPI()
|
|
log = core.utils.get_logger("web")
|
|
|
|
uvserver = None
|
|
data_pool = []
|
|
data_run = [True]
|
|
|
|
|
|
def response(data=None, code=status.HTTP_200_OK, error_code=0, error_message=None):
|
|
if 200 >= code <= 300:
|
|
return JSONResponse(content={"result": data, "error": None}, status_code=code)
|
|
return JSONResponse(
|
|
content={"error": {"code": error_code if error_code else code, "message": f"{error_message}"}, "result": None},
|
|
status_code=code)
|
|
|
|
|
|
@web_app.get("/")
|
|
async def index():
|
|
log.debug("Request IndexPage;")
|
|
return response("Index page")
|
|
|
|
|
|
@web_app.get("/method/{method}")
|
|
async def _method(method, secret_key: str = None):
|
|
# log.debug(f"Request method; kwargs: {kwargs}")
|
|
is_auth = secret_key == config.WebAPI["secret_key"]
|
|
spl = method.split(".")
|
|
if len(spl) != 2:
|
|
raise StarletteHTTPException(405)
|
|
api_class, api_method = spl
|
|
match api_class:
|
|
case "events":
|
|
match api_method, is_auth:
|
|
case "get", False:
|
|
return response(data_pool)
|
|
raise StarletteHTTPException(404)
|
|
|
|
|
|
async def _stop():
|
|
await asyncio.sleep(1)
|
|
if uvserver is not None:
|
|
uvserver.should_exit = True
|
|
data_run[0] = False
|
|
|
|
|
|
@web_app.get("/stop")
|
|
async def stop(secret_key: str):
|
|
log.debug(f"Request stop; secret key: {secret_key}")
|
|
if secret_key == config.WebAPI["secret_key"]:
|
|
log.info("Stopping Web server")
|
|
asyncio.create_task(_stop())
|
|
return response("Web server stopped")
|
|
|
|
|
|
@web_app.exception_handler(HTTPException)
|
|
async def default_exception_handler(request: Request, exc: HTTPException):
|
|
return response(
|
|
code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
error_code=exc.status_code, error_message=f"Internal Server Error: {exc.status_code}"
|
|
)
|
|
|
|
|
|
@web_app.exception_handler(StarletteHTTPException)
|
|
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
|
|
code = exc.status_code
|
|
if code == status.HTTP_405_METHOD_NOT_ALLOWED:
|
|
return response(code=code, error_message="Method Not Allowed")
|
|
if code == status.HTTP_404_NOT_FOUND:
|
|
return response(code=code, error_message="Method not Found")
|
|
return response(code=code, error_message="Unhandled error..")
|
|
|
|
|
|
@web_app.exception_handler(RequestValidationError)
|
|
async def request_validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
code = status.HTTP_422_UNPROCESSABLE_ENTITY
|
|
return response(code=code, error_message="Request Validation Error")
|
|
|
|
|
|
utils.hack_fastapi()
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
uvconfig = uvicorn.Config(web_app,
|
|
host=config.WebAPI["server_ip"],
|
|
port=config.WebAPI["server_port"],
|
|
loop="asyncio")
|
|
uvserver = uvicorn.Server(uvconfig)
|
|
uvserver.run()
|
|
except KeyboardInterrupt or CancelledError:
|
|
pass
|