Single-entity commands for the shell-oriented DP3 CLI.
handle_get
handle_get(client, args) -> int
Get full entity data.
Source code in dp3/bin/shcmd/entity/instance.py
| def handle_get(client, args) -> int:
"""Get full entity data."""
return print_response_json(
client.request("GET", f"/entity/{args.etype}/{args.eid}", params=common_time_params(args))
)
|
handle_master
handle_master(client, args) -> int
Get the master record for one entity.
Source code in dp3/bin/shcmd/entity/instance.py
| def handle_master(client, args) -> int:
"""Get the master record for one entity."""
return print_response_json(
client.request(
"GET", f"/entity/{args.etype}/{args.eid}/master", params=common_time_params(args)
)
)
|
handle_snapshots
handle_snapshots(client, args) -> int
Get snapshots for one entity.
Source code in dp3/bin/shcmd/entity/instance.py
| def handle_snapshots(client, args) -> int:
"""Get snapshots for one entity."""
path = f"/entity/{args.etype}/{args.eid}/snapshots"
if args.format == "ndjson":
return stream_json_pages(client, path, common_time_params(args), args.skip, args.limit)
return print_response_json(client.request("GET", path, params=build_time_page_params(args)))
|
handle_ttl
handle_ttl(client, args) -> int
Extend entity TTLs.
Source code in dp3/bin/shcmd/entity/instance.py
| def handle_ttl(client, args) -> int:
"""Extend entity TTLs."""
body = read_json_value(args.body_json)
return print_response_json(
client.request("POST", f"/entity/{args.etype}/{args.eid}/ttl", json_body=body)
)
|
handle_delete
handle_delete(client, args) -> int
Delete entity data.
Source code in dp3/bin/shcmd/entity/instance.py
| def handle_delete(client, args) -> int:
"""Delete entity data."""
return print_response_json(client.request("DELETE", f"/entity/{args.etype}/{args.eid}"))
|
add_id_parser
add_id_parser(commands, etype: str) -> None
Register the id entity selector and nested single-entity commands.
Source code in dp3/bin/shcmd/entity/instance.py
| def add_id_parser(commands, etype: str) -> None:
"""Register the `id` entity selector and nested single-entity commands."""
id_parser = commands.add_parser(
"id",
help="Inspect or modify one entity by id.",
description=f"Inspect or modify one entity of type '{etype}' by id.",
)
id_parser.set_defaults(etype=etype)
eid_action = id_parser.add_argument("eid", metavar="EID", help="Entity id.")
eid_action.completer = complete_entity_id_placeholder
_add_instance_commands(id_parser, etype)
|
complete_entity_id_placeholder
complete_entity_id_placeholder(prefix: str, **_kwargs) -> dict[str, str]
Suggest the entity-id placeholder in completion output.
Source code in dp3/bin/shcmd/entity/instance.py
| def complete_entity_id_placeholder(prefix: str, **_kwargs) -> dict[str, str]:
"""Suggest the entity-id placeholder in completion output."""
if ENTITY_ID_PLACEHOLDER.startswith(prefix):
return {ENTITY_ID_PLACEHOLDER: "Enter an entity id to inspect one entity."}
return {}
|