python-fastapi-mm-zhcs-yj-api/main.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from fastapi import FastAPI, Request #, Response,HTTPException
from fastapi.responses import PlainTextResponse, JSONResponse, RedirectResponse, FileResponse
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.staticfiles import StaticFiles
from routers import home, api ,prod_api
from exceptions import *
from pymysql.err import OperationalError
import os
import platform
from config import settings
from extensions import logger
from jobs import register_jobs
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from common.websocketManager import manager
from utils.redis_util import *

app = FastAPI(docs_url=None, redoc_url=None)
# app.include_router(home.router)
app.include_router(api.router, prefix="/api")
app.include_router(prod_api.router, prefix="")

app.add_middleware(SessionMiddleware, secret_key='MM-ZHCS-YJ-API', max_age=36000 * 24)
app.add_middleware(CORSMiddleware,allow_origins=['*'],allow_credentials=False,allow_methods=['*'],allow_headers=['*'])

app.mount('/static', StaticFiles(directory='static'), name='static')

@app.middleware("http")
async def process_authorization(request: Request, call_next):
    response = await call_next(request)
    return response
    
@app.get("/", response_class=PlainTextResponse)
async def main():
    return "应急后端接口9988"

@app.get("/favicon.ico", response_class=FileResponse)
async def favicon():
    return FileResponse("./static/favicon.ico")


@app.on_event('startup')
def app_startup():
    lock_key = "MM-ZHCS-YJ-API-app_startup"
    if redis_lock(lock_key):
        logger.info('---------------------------------------')
        logger.info('--------------'+ settings.ENVIRONMENT +'---------------')
        logger.info('---------------------------------------')
        logger.info("server started.")

        try:
            from common.enc import mpfun
            print(mpfun.enc_data('test2'))
        except Exception as e:
            print(str(e))

        sys = platform.system()
        if sys == "Windows":
            from warnings import filterwarnings
            from pytz_deprecation_shim import PytzUsageWarning
            filterwarnings('ignore', category=PytzUsageWarning)

        executors = {
            'default': ThreadPoolExecutor(20),
            'processpool': ProcessPoolExecutor(5)
        }

        job_defaults = {
            'coalesce': True,
            'max_instance': 1
        }
        
        scheduler = AsyncIOScheduler(timezone='Asia/Shanghai', executors=executors, job_defaults=job_defaults)
        register_jobs(scheduler)
        scheduler.start()

        redis_unlock(lock_key)

# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
#     return JSONResponse(status_code=200, content={ "errcode": 100060, "errmsg": "参数错误" }, )

@app.exception_handler(TokenException)
async def exception_handler(request: Request, exc: TokenException):
    return JSONResponse(
        status_code=200,
        content={"code": 401, "msg":"Token过期"}
        # content={"errcode": "4001", "errmsg":"验证不通过"}
    )
@app.exception_handler(RoleException)
async def role_exception_handler(request: Request, exc: RoleException):
    # 这样界面才有文字提示
    return JSONResponse(
        status_code=200,
        content={"code": 403, "msg":"表示权限不足"}
    )
    # return JSONResponse(
    #     status_code=403,  # 使用 403 Forbidden 表示权限不足
    #     content={"errcode": exc.errcode, "errmsg": exc.errmsg}
    # )
@app.exception_handler(OperationalError)
async def exception_handler(request: Request, exc: OperationalError):
    return JSONResponse(
        status_code=412,
        content={"code": "412", "msg":f"接口对应sql异常，请检查{exc}"}
    )


# @app.exception_handler(AttributeError)
# async def exception_handler(request: Request, exc: TokenException):
#     return JSONResponse(
#         status_code=200,
#         content={"errcode": "500", "errmsg":"Internal Server Error"}
#     )


@app.exception_handler(AuthException)
async def exception_handler(request: Request, exc: TokenException):
    return RedirectResponse("/")
    
@app.exception_handler(AlertException)
async def exception_handler(request: Request, exc: AlertException):
    return JSONResponse(
        status_code=200,
        content={"ret": 1, "msg": exc.msg}
    )


    
if __name__ == '__main__':
    import uvicorn
    sys = platform.system()
    if sys == "Windows":
        os.system("title 茂名应急数据中台 DEV")
    
    uvicorn.run(app='main:app', host='0.0.0.0',port=9988, reload=True, debug=True)
    # uvicorn.run(app='main:app', host='0.0.0.0', port=9988)