from fastapi import APIRouter, Request, Query, Path, Body from typing import Any, List, Dict, Union from app.services.tjnetwork import ( Any, ChangeSet, get_control, get_control_schema, get_rule, get_rule_schema, set_control, set_rule, ) router = APIRouter() @router.get("/getcontrolschema/", summary="获取控制架构", description="获取网络中控制对象的架构定义") async def fastapi_get_control_schema(network: str = Query(..., description="管网名称(或数据库名称)")) -> dict[str, dict[str, Any]]: """获取控制架构。 返回指定网络中控制对象的属性架构定义。 """ return get_control_schema(network) @router.get("/getcontrolproperties/", summary="获取控制属性", description="获取指定网络中的控制属性信息") async def fastapi_get_control_properties(network: str = Query(..., description="管网名称(或数据库名称)")) -> dict[str, Any]: """获取控制属性。 返回指定网络中的控制对象属性信息。 """ return get_control(network) @router.post("/setcontrolproperties/", response_model=None, summary="设置控制属性", description="更新指定网络中的控制属性") async def fastapi_set_control_properties( network: str = Query(..., description="管网名称(或数据库名称)"), req: Request = None ) -> ChangeSet: """设置控制属性。 更新指定网络中的控制属性值。 """ props = await req.json() return set_control(network, ChangeSet(props)) @router.get("/getruleschema/", summary="获取规则架构", description="获取网络中规则对象的架构定义") async def fastapi_get_rule_schema(network: str = Query(..., description="管网名称(或数据库名称)")) -> dict[str, dict[str, Any]]: """获取规则架构。 返回指定网络中规则对象的属性架构定义。 """ return get_rule_schema(network) @router.get("/getruleproperties/", summary="获取规则属性", description="获取指定网络中的规则属性信息") async def fastapi_get_rule_properties(network: str = Query(..., description="管网名称(或数据库名称)")) -> dict[str, Any]: """获取规则属性。 返回指定网络中的规则对象属性信息。 """ return get_rule(network) @router.post("/setruleproperties/", response_model=None, summary="设置规则属性", description="更新指定网络中的规则属性") async def fastapi_set_rule_properties( network: str = Query(..., description="管网名称(或数据库名称)"), req: Request = None ) -> ChangeSet: """设置规则属性。 更新指定网络中的规则属性值。 """ props = await req.json() return set_rule(network, ChangeSet(props))