Work With Results

Each item yielded by a scan stream is a Python dict containing the same tags that appear in the CLI (target, probe, alias, protocol-specific fields, latency data, etc.). This page shows how to work with those dictionaries directly.

Inspect replies

async def show_basic_fields(stream):
    async for row in stream:
        print(row["target"], row.get("alias"), row.get("latency_ms"))

Use .get() when a field may be absent (for example, some probes do not emit latency_ms).

Collect results for later processing

async def gather_results(stream):
    results = []
    async for row in stream:
        results.append(row)
    return results

Because each row is already a dictionary, you can store the list in memory, hand it to another function, or serialise it with json.dumps.

Derive quick summaries

def count_alias_states(rows):
    summary = {"aliased": 0, "not_aliased": 0, "unknown": 0}
    for row in rows:
        verdict = row.get("alias", "unknown")
        summary.setdefault(verdict, 0)
        summary[verdict] += 1
    return summary

The same pattern works for other columns such as probe, port, or protocol-specific tags.

Preserve ordering with lightweight wrappers

from dataclasses import dataclass

@dataclass
class Reply:
    target: str
    alias: str | None
    tags: dict[str, str]

def to_reply(row: dict[str, str]) -> Reply:
    return Reply(
        target=row["target"],
        alias=row.get("alias"),
        tags=row,
    )

Wrapping rows makes it easy to type-annotate the values the rest of your code expects while still storing the complete tag dictionary for reference.