42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
from typing import List
|
|
|
|
from app.core.crud import CRUDBase
|
|
from app.models.admin import Api, Menu, Datasource
|
|
from app.schemas.codegen import DatasourceCreate, DatasourceUpdate, DatasourceInfo
|
|
from app.utils.db import DatabaseInfo
|
|
|
|
class DatasourceController(CRUDBase[Datasource, DatasourceCreate, DatasourceUpdate]):
|
|
def __init__(self):
|
|
super().__init__(model=Datasource)
|
|
|
|
async def load_tables(self, name: str) -> list[DatasourceInfo]:
|
|
# datasource_obj = await self.model.filter(name=name).first()
|
|
# if not datasource_obj:
|
|
# raise HTTPException(status_code=400, detail="数据源不存在")
|
|
|
|
with DatabaseInfo(
|
|
host="lt.330770.xyz",
|
|
port=3307,
|
|
user="root",
|
|
password="rap_sky",
|
|
database="rpa"
|
|
) as db:
|
|
tables = db.get_all_tables()
|
|
|
|
return tables
|
|
|
|
|
|
async def update_datasources(self, datasource: Datasource, menu_ids: List[int], api_infos: List[dict]) -> None:
|
|
await datasource.menus.clear()
|
|
for menu_id in menu_ids:
|
|
menu_obj = await Menu.filter(id=menu_id).first()
|
|
await datasource.menus.add(menu_obj)
|
|
|
|
await datasource.apis.clear()
|
|
for item in api_infos:
|
|
api_obj = await Api.filter(path=item.get("path"), method=item.get("method")).first()
|
|
await datasource.apis.add(api_obj)
|
|
|
|
|
|
datasource_controller = DatasourceController()
|