46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
from fastapi.routing import APIRoute
|
|
|
|
from app.core.crud import CRUDBase
|
|
from app.log import logger
|
|
from app.models.admin import Api
|
|
from app.schemas.apis import ApiCreate, ApiUpdate
|
|
|
|
|
|
class ApiController(CRUDBase[Api, ApiCreate, ApiUpdate]):
|
|
def __init__(self):
|
|
super().__init__(model=Api)
|
|
|
|
async def refresh_api(self):
|
|
from app import app
|
|
|
|
# 删除废弃API数据
|
|
all_api_list = []
|
|
for route in app.routes:
|
|
# 只更新有鉴权的API
|
|
if isinstance(route, APIRoute) and len(route.dependencies) > 0:
|
|
all_api_list.append((list(route.methods)[0], route.path_format))
|
|
delete_api = []
|
|
for api in await Api.all():
|
|
if (api.method, api.path) not in all_api_list:
|
|
delete_api.append((api.method, api.path))
|
|
for item in delete_api:
|
|
method, path = item
|
|
logger.debug(f"API Deleted {method} {path}")
|
|
await Api.filter(method=method, path=path).delete()
|
|
|
|
for route in app.routes:
|
|
if isinstance(route, APIRoute) and len(route.dependencies) > 0:
|
|
method = list(route.methods)[0]
|
|
path = route.path_format
|
|
summary = route.summary
|
|
tags = list(route.tags)[0] if route.tags else "default"
|
|
api_obj = await Api.filter(method=method, path=path).first()
|
|
if api_obj:
|
|
await api_obj.update_from_dict(dict(method=method, path=path, summary=summary, tags=tags)).save()
|
|
else:
|
|
logger.debug(f"API Created {method} {path}")
|
|
await Api.create(**dict(method=method, path=path, summary=summary, tags=tags))
|
|
|
|
|
|
api_controller = ApiController()
|