from typing import List, Any from fastapi import APIRouter, Request, HTTPException, Query, Body from app.services.tjnetwork import ( ChangeSet, get_all_extension_data_keys, get_all_extension_data, get_extension_data, set_extension_data ) router = APIRouter() @router.get( "/getallextensiondatakeys/", summary="获取所有扩展数据键", description="获取指定网络的所有扩展数据的键列表" ) async def get_all_extension_data_keys_endpoint( network: str = Query(..., description="管网名称(或数据库名称)") ) -> list[str]: """ 获取所有扩展数据键。 返回指定网络中所有可用的扩展数据键。 Args: network: 管网名称(或数据库名称) Returns: 扩展数据键列表 """ return get_all_extension_data_keys(network) @router.get( "/getallextensiondata/", summary="获取所有扩展数据", description="获取指定网络的所有扩展数据" ) async def get_all_extension_data_endpoint( network: str = Query(..., description="管网名称(或数据库名称)") ) -> dict[str, Any]: """ 获取所有扩展数据。 返回指定网络的所有扩展数据及其值。 Args: network: 管网名称(或数据库名称) Returns: 扩展数据字典 """ return get_all_extension_data(network) @router.get( "/getextensiondata/", summary="获取指定扩展数据", description="获取指定网络中指定键的扩展数据值" ) async def get_extension_data_endpoint( network: str = Query(..., description="管网名称(或数据库名称)"), key: str = Query(..., description="扩展数据键") ) -> str | None: """ 获取指定扩展数据。 返回指定网络中指定键对应的扩展数据值。 Args: network: 管网名称(或数据库名称) key: 扩展数据键 Returns: 扩展数据值,如果不存在返回None """ return get_extension_data(network, key) @router.post( "/setextensiondata/", response_model=None, summary="设置扩展数据", description="设置指定网络中的扩展数据" ) async def set_extension_data_endpoint( network: str = Query(..., description="管网名称(或数据库名称)"), req: Request = None ) -> ChangeSet: """ 设置扩展数据。 在指定网络中设置扩展数据,并返回变更集信息。 Args: network: 管网名称(或数据库名称) req: 包含扩展数据的请求体 Returns: 变更集信息 """ props = await req.json() print(props) cs = set_extension_data(network, ChangeSet(props)) print(cs.operations[0]) return cs