更新 cli 命令,新增 network 其他元素的属性查询

This commit is contained in:
2026-06-05 15:48:53 +08:00
parent 7efaeb41e8
commit 52b8f07abd
3 changed files with 468 additions and 59 deletions
+137 -27
View File
@@ -10,56 +10,42 @@ from .core import CLIError, require_network
from .option_types import ComponentOptionKind
@network_app.command("get-node-properties")
def network_get_node_properties(
@network_app.command("get-junction-properties")
def network_get_junction_properties(
ctx: typer.Context,
node: Annotated[str, typer.Option("--node", help="节点 ID")],
junction: Annotated[str, typer.Option("--junction", help="节点 ID")],
) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取节点属性成功",
method="GET",
path="/getnodeproperties/",
params={"network": require_network(runtime), "node": node},
path="/getjunctionproperties/",
params={"network": require_network(runtime), "junction": junction},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-link-properties")
def network_get_link_properties(
@network_app.command("get-pipe-properties")
def network_get_pipe_properties(
ctx: typer.Context,
link: Annotated[str, typer.Option("--link", help="线 ID")],
pipe: Annotated[str, typer.Option("--pipe", help=" ID")],
) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取管线属性成功",
summary="读取管属性成功",
method="GET",
path="/getlinkproperties/",
params={"network": require_network(runtime), "link": link},
path="/getpipeproperties/",
params={"network": require_network(runtime), "pipe": pipe},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-all-junction-properties")
def network_get_all_junction_properties(ctx: typer.Context) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取全部节点属性成功",
method="GET",
path="/getalljunctionproperties/",
params={"network": require_network(runtime)},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-all-pipe-properties")
def network_get_all_pipe_properties(ctx: typer.Context) -> None:
@network_app.command("get-all-pipes-properties")
def network_get_all_pipes_properties(ctx: typer.Context) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
@@ -72,6 +58,130 @@ def network_get_all_pipe_properties(ctx: typer.Context) -> None:
)
@network_app.command("get-reservoir-properties")
def network_get_reservoir_properties(
ctx: typer.Context,
reservoir: Annotated[str, typer.Option("--reservoir", help="水库 ID")],
) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取水库属性成功",
method="GET",
path="/getreservoirproperties/",
params={"network": require_network(runtime), "reservoir": reservoir},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-all-reservoirs-properties")
def network_get_all_reservoir_properties(ctx: typer.Context) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取全部水库属性成功",
method="GET",
path="/getallreservoirproperties/",
params={"network": require_network(runtime)},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-tank-properties")
def network_get_tank_properties(
ctx: typer.Context,
tank: Annotated[str, typer.Option("--tank", help="水箱 ID")],
) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取水箱属性成功",
method="GET",
path="/gettankproperties/",
params={"network": require_network(runtime), "tank": tank},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-all-tanks-properties")
def network_get_all_tank_properties(ctx: typer.Context) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取全部水箱属性成功",
method="GET",
path="/getalltankproperties/",
params={"network": require_network(runtime)},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-pump-properties")
def network_get_pump_properties(
ctx: typer.Context,
pump: Annotated[str, typer.Option("--pump", help="水泵 ID")],
) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取水泵属性成功",
method="GET",
path="/getpumpproperties/",
params={"network": require_network(runtime), "pump": pump},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-all-pumps-properties")
def network_get_all_pump_properties(ctx: typer.Context) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取全部水泵属性成功",
method="GET",
path="/getallpumpproperties/",
params={"network": require_network(runtime)},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-valve-properties")
def network_get_valve_properties(
ctx: typer.Context,
valve: Annotated[str, typer.Option("--valve", help="阀门 ID")],
) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取阀门属性成功",
method="GET",
path="/getvalveproperties/",
params={"network": require_network(runtime), "valve": valve},
require_auth=True,
require_network_ctx=True,
)
@network_app.command("get-all-valves-properties")
def network_get_all_valve_properties(ctx: typer.Context) -> None:
runtime = runtime_context(ctx)
emit_api(
ctx,
summary="读取全部阀门属性成功",
method="GET",
path="/getallvalveproperties/",
params={"network": require_network(runtime)},
require_auth=True,
require_network_ctx=True,
)
@component_option_app.command("schema")
def component_option_schema(
ctx: typer.Context,