Skip to content

dp3.bin.shcmd.entity.instance

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 {}