100 lines
4.1 KiB
Python
100 lines
4.1 KiB
Python
# -*- coding:utf-8 -*-
|
|
"""
|
|
@Author : xuxingchen
|
|
@Contact : xuxingchen@sinochem.com
|
|
@Desc : api后端服务程序启动入口
|
|
"""
|
|
import argparse
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
import uvicorn
|
|
|
|
from config import VERSION, PREFIX_PATH
|
|
from utils import logger, database, misc
|
|
from models import sessions, houses, devices, products, householders, subsystem, parkinglots
|
|
from routers import (login, houses_manage, devices_manage, space_manage, householder_manage,
|
|
access_devices_0, access_devices_1,
|
|
parkinglot_0, parkinglot_1)
|
|
|
|
|
|
class MainServer:
|
|
def __init__(self, app_args):
|
|
self.port = app_args.port
|
|
self.app = None
|
|
self.prepare()
|
|
self.server_main()
|
|
|
|
def prepare(self):
|
|
"""准备处理，一些适配和配置加载工作"""
|
|
# 数据库初始化
|
|
logger.Logger.init("数据库执行初始化校验 ...")
|
|
th = database.get_table_handler()
|
|
sessions.SessionsTable.check(th)
|
|
houses.HousesTable.check(th)
|
|
products.ProductsTable.check(th)
|
|
devices.DevicesTable.check(th)
|
|
subsystem.SubsystemTable.check(th)
|
|
householders.HouseholdersTable.check(th)
|
|
parkinglots.ParkinglotsTable.check(th)
|
|
logger.Logger.init("数据库完成初始化校验 ✅")
|
|
|
|
self.app = FastAPI(
|
|
title="Smart Community Local Platform",
|
|
description="智慧社区本地化平台后端服务",
|
|
version=VERSION,
|
|
docs_url=f"{PREFIX_PATH}/docs" # 设为None则会关闭
|
|
)
|
|
|
|
# 注册路由模块中的路由
|
|
self.app.include_router(login.router, prefix=PREFIX_PATH, tags=["登录界面"])
|
|
self.app.include_router(houses_manage.router, prefix=PREFIX_PATH, tags=["内置信息"])
|
|
self.app.include_router(householder_manage.router, prefix=PREFIX_PATH, tags=["住户管理界面"])
|
|
self.app.include_router(access_devices_0.router, prefix=PREFIX_PATH, tags=["通行设备界面"])
|
|
self.app.include_router(access_devices_1.router, prefix=PREFIX_PATH, tags=["通行授权界面"])
|
|
self.app.include_router(parkinglot_0.router, prefix=PREFIX_PATH, tags=["车场管理界面"])
|
|
self.app.include_router(parkinglot_1.router, prefix=PREFIX_PATH, tags=["车场授权界面"])
|
|
self.app.include_router(devices_manage.router, prefix=PREFIX_PATH, tags=["设备管理界面"])
|
|
self.app.include_router(space_manage.router, prefix=PREFIX_PATH, tags=["通行空间映射界面"])
|
|
|
|
@self.app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
logger.Logger.error(f"{request.url.path} 请求参数错误 {exc.errors()} - {request.client.host}")
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content={"status": False,
|
|
"message": f"请求参数错误，请检查输入"
|
|
f"{'，错误类型：' + exc.errors()[0]['type'] if len(exc.errors()) > 0 else ''}"}
|
|
)
|
|
|
|
@self.app.exception_handler(misc.InvalidException)
|
|
async def invalid_target_exception_handler(request: Request, exc: misc.InvalidException):
|
|
logger.Logger.error(f"{request.url.path} {exc.message} - {request.client.host}")
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={"status": False, "message": f"{exc.message}"}
|
|
)
|
|
|
|
def server_main(self):
|
|
"""主服务程序"""
|
|
logger.Logger.debug(f"当前已开启 DEBUG 模式")
|
|
uvicorn.run(self.app, host="0.0.0.0", port=self.port, access_log=False)
|
|
|
|
|
|
def main(port: str = 9680):
|
|
parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
|
|
parser.add_argument(
|
|
"-p",
|
|
"--port",
|
|
type=int,
|
|
default=port,
|
|
help=f"Port of server, default {logger.new_dc(port)}",
|
|
)
|
|
app_args = parser.parse_args()
|
|
MainServer(app_args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|