Skip to content

utils

utils

Finder service utility functions.

Pure helpers for relay URL extraction from Nostr event data and API responses.

Classes

Functions

extract_urls_from_response 

extract_urls_from_response(
    data: Any, expression: str = "[*]"
) -> list[str]

Extract URL strings from a JSON API response via a JMESPath expression.

Applies expression to the parsed JSON data and filters the result to only string values. Validation (scheme, host, etc.) is left to the caller.

Parameters:

  • data (Any) –

    Parsed JSON response (any type).

  • expression (str, default: '[*]' ) –

    JMESPath expression that should evaluate to a list of strings. Defaults to [*] (identity on a flat list).

Returns:

  • list[str] –

    List of extracted URL strings (may contain duplicates or invalid

  • list[str] –

    values -- the caller decides how to validate them).

Source code in src/bigbrotr/services/finder/utils.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def extract_urls_from_response(data: Any, expression: str = "[*]") -> list[str]:
    """Extract URL strings from a JSON API response via a JMESPath expression.

    Applies *expression* to the parsed JSON *data* and filters the result
    to only string values.  Validation (scheme, host, etc.) is left to the
    caller.

    Args:
        data: Parsed JSON response (any type).
        expression: JMESPath expression that should evaluate to a list of
            strings.  Defaults to ``[*]`` (identity on a flat list).

    Returns:
        List of extracted URL strings (may contain duplicates or invalid
        values -- the caller decides how to validate them).
    """
    result = jmespath.search(expression, data)
    if not isinstance(result, list):
        return []
    return [item for item in result if isinstance(item, str)]

extract_relays_from_rows 

extract_relays_from_rows(
    rows: list[dict[str, Any]],
) -> dict[str, Relay]

Extract and deduplicate relay URLs from event tagvalues.

Parses every value in each row's tagvalues array via parse_relay_url. Values that parse as valid relay URLs become candidates; all others (hex IDs, pubkeys, hashtags, etc.) are silently discarded.

Parameters:

  • rows (list[dict[str, Any]]) –

    Event rows with tagvalues key (from fetch_event_tagvalues).

Returns:

  • dict[str, Relay] –

    Mapping of normalized relay URL to

  • dict[str, Relay] –

    Relay for deduplication.

Source code in src/bigbrotr/services/finder/utils.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def extract_relays_from_rows(rows: list[dict[str, Any]]) -> dict[str, Relay]:
    """Extract and deduplicate relay URLs from event tagvalues.

    Parses every value in each row's ``tagvalues`` array via
    ``parse_relay_url``. Values that parse as valid relay URLs become
    candidates; all others (hex IDs, pubkeys, hashtags, etc.) are
    silently discarded.

    Args:
        rows: Event rows with ``tagvalues`` key (from
            ``fetch_event_tagvalues``).

    Returns:
        Mapping of normalized relay URL to
        [Relay][bigbrotr.models.relay.Relay] for deduplication.
    """
    relays: dict[str, Relay] = {}

    for row in rows:
        tagvalues = row.get("tagvalues")
        if not tagvalues:
            continue
        for val in tagvalues:
            validated = parse_relay_url(val)
            if validated:
                relays[validated.url] = validated

    return relays