simAGV/main.py

# encoding: utf-8
import asyncio
import os
import time

import logging
from tortoise import Tortoise
import uvicorn
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from tortoise.exceptions import DoesNotExist, DBConnectionError

from conf import setting, register_mysql
from logic import (
    agv,
    warehouse
)
from agvtask.agvtasks import start_scheduler

app = FastAPI(debug=setting.DEBUG_MODE)
app.mount('/static', StaticFiles(directory="static"), name='static')


def enable_debug_sql():
    import logging, sys

    fmt = logging.Formatter(
        fmt="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )
    sh = logging.StreamHandler(sys.stdout)
    sh.setLevel(logging.DEBUG)
    sh.setFormatter(fmt)

    # will print debug sql
    logger_db_client = logging.getLogger("tortoise.db_client")
    logger_db_client.setLevel(logging.DEBUG)


if setting.SQL_DEBUG_MODE:
    enable_debug_sql()

# 注册连接数据库
register_mysql(app)
Tortoise.init_models(["models.agv"], "models")


@app.on_event("startup")
async def startup():
    await start_scheduler()

# 执行时间中间件
@app.middleware('http')
async def process_time_header(request: Request, next):
    start_time = time.time()
    response = await next(request)
    process_time = time.time() - start_time
    response.headers['X-Process-Time'] = str(process_time)
    return response


@app.on_event("startup")
async def startup_log_event():
    logger = logging.getLogger("uvicorn.access")
    access_error = logging.getLogger("uvicorn.error")
    handler = logging.handlers.TimedRotatingFileHandler('logs/server.log', when='midnight', backupCount=30)
    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
    logger.addHandler(handler)
    logger.setLevel(logging.WARNING)
    access_error.addHandler(handler)


# 跨域中间件
app.add_middleware(
    CORSMiddleware, allow_origins=setting.YANEI_CORS_ORIGINS, allow_credentials=True, allow_methods=['*'], allow_headers=['*']
)

app.include_router(agv.router)
app.include_router(warehouse.router)

if __name__ == '__main__':
    uvicorn.run(app='main:app', host='0.0.0.0', port=setting.PORT, reload=setting.DEBUG_DEV_MODE)