Skip to content

utils

utils

DVM service utility functions.

Pure helpers for NIP-90 event parsing and event builder construction. All functions are stateless — they build EventBuilder instances or parse event data without touching the network or database.

Classes

Functions

parse_job_params 

parse_job_params(event: Any) -> dict[str, Any]

Extract NIP-90 parameters from event tags.

Reads ["param", key, value] tags and an optional ["bid", amount] tag from the event. Invalid bid values are silently ignored.

Parameters:

  • event (Any) –

    A nostr_sdk.Event (or mock with .tags().to_vec()).

Returns:

  • dict[str, Any] –

    Dict mapping parameter names to their string values, plus an

  • dict[str, Any] –

    optional "bid" key with an integer value.

Source code in src/bigbrotr/services/dvm/utils.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def parse_job_params(event: Any) -> dict[str, Any]:
    """Extract NIP-90 parameters from event tags.

    Reads ``["param", key, value]`` tags and an optional ``["bid", amount]``
    tag from the event.  Invalid bid values are silently ignored.

    Args:
        event: A ``nostr_sdk.Event`` (or mock with ``.tags().to_vec()``).

    Returns:
        Dict mapping parameter names to their string values, plus an
        optional ``"bid"`` key with an integer value.
    """
    params: dict[str, Any] = {}
    for tag in event.tags().to_vec():
        values = tag.as_vec()
        if len(values) >= _MIN_PARAM_TAG_LEN and values[0] == "param":
            params[values[1]] = values[2]
        elif len(values) >= _MIN_TAG_LEN and values[0] == "bid":
            with contextlib.suppress(ValueError):
                params["bid"] = int(values[1])
    return params

parse_query_filters 

parse_query_filters(
    filter_str: str,
) -> dict[str, str] | None

Parse a comma-separated filter string into a dict.

Format: "column=value,column=op:value" where op is one of =, >, <, >=, <=, ILIKE.

Parameters:

  • filter_str (str) –

    Raw filter string from a NIP-90 param tag.

Returns:

  • dict[str, str] | None –

    Dict of column→value pairs, or None if empty or unparsable.

Source code in src/bigbrotr/services/dvm/utils.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def parse_query_filters(filter_str: str) -> dict[str, str] | None:
    """Parse a comma-separated filter string into a dict.

    Format: ``"column=value,column=op:value"`` where ``op`` is one of
    ``=``, ``>``, ``<``, ``>=``, ``<=``, ``ILIKE``.

    Args:
        filter_str: Raw filter string from a NIP-90 ``param`` tag.

    Returns:
        Dict of column→value pairs, or ``None`` if empty or unparsable.
    """
    if not filter_str:
        return None
    filters: dict[str, str] = {}
    for raw_part in filter_str.split(","):
        part = raw_part.strip()
        if "=" not in part:
            continue
        key, _, value = part.partition("=")
        filters[key.strip()] = value.strip()
    return filters or None

build_result_event 

build_result_event(
    request_kind: int,
    request_event_id: str,
    customer_pubkey: str,
    result: QueryResult,
    price: int,
) -> EventBuilder

Build a NIP-90 job result event (request kind + 1000).

Parameters:

  • request_kind (int) –

    The NIP-90 request kind (e.g. 5050).

  • request_event_id (str) –

    Hex ID of the request event.

  • customer_pubkey (str) –

    Hex pubkey of the requesting user.

  • result (QueryResult) –

    Query result to serialize as JSON content.

  • price (int) –

    Millisat price (included as amount tag if > 0).

Returns:

  • EventBuilder –

    Unsigned EventBuilder ready for signing and sending.

Source code in src/bigbrotr/services/dvm/utils.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def build_result_event(
    request_kind: int,
    request_event_id: str,
    customer_pubkey: str,
    result: QueryResult,
    price: int,
) -> EventBuilder:
    """Build a NIP-90 job result event (request kind + 1000).

    Args:
        request_kind: The NIP-90 request kind (e.g. 5050).
        request_event_id: Hex ID of the request event.
        customer_pubkey: Hex pubkey of the requesting user.
        result: Query result to serialize as JSON content.
        price: Millisat price (included as ``amount`` tag if > 0).

    Returns:
        Unsigned ``EventBuilder`` ready for signing and sending.
    """
    result_kind = request_kind + 1000
    content = json.dumps(
        {
            "data": result.rows,
            "meta": {
                "total": result.total,
                "limit": result.limit,
                "offset": result.offset,
            },
        },
        default=str,
    )

    tags = [
        Tag.parse(["e", request_event_id]),
        Tag.parse(["p", customer_pubkey]),
        Tag.parse(
            [
                "request",
                json.dumps({"id": request_event_id, "kind": request_kind}),
            ]
        ),
    ]
    if price > 0:
        tags.append(Tag.parse(["amount", str(price)]))

    return EventBuilder(Kind(result_kind), content).tags(tags)

build_error_event 

build_error_event(
    request_event_id: str,
    customer_pubkey: str,
    error_message: str,
) -> EventBuilder

Build a NIP-90 error feedback event (kind 7000).

Parameters:

  • request_event_id (str) –

    Hex ID of the request event.

  • customer_pubkey (str) –

    Hex pubkey of the requesting user.

  • error_message (str) –

    Human-readable error description.

Returns:

  • EventBuilder –

    Unsigned EventBuilder ready for signing and sending.

Source code in src/bigbrotr/services/dvm/utils.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def build_error_event(
    request_event_id: str,
    customer_pubkey: str,
    error_message: str,
) -> EventBuilder:
    """Build a NIP-90 error feedback event (kind 7000).

    Args:
        request_event_id: Hex ID of the request event.
        customer_pubkey: Hex pubkey of the requesting user.
        error_message: Human-readable error description.

    Returns:
        Unsigned ``EventBuilder`` ready for signing and sending.
    """
    tags = [
        Tag.parse(["status", "error", error_message]),
        Tag.parse(["e", request_event_id]),
        Tag.parse(["p", customer_pubkey]),
    ]
    return EventBuilder(Kind(7000), "").tags(tags)

build_payment_required_event 

build_payment_required_event(
    request_event_id: str, customer_pubkey: str, price: int
) -> EventBuilder

Build a NIP-90 payment-required feedback event (kind 7000).

Parameters:

  • request_event_id (str) –

    Hex ID of the request event.

  • customer_pubkey (str) –

    Hex pubkey of the requesting user.

  • price (int) –

    Required payment in millisats.

Returns:

  • EventBuilder –

    Unsigned EventBuilder ready for signing and sending.

Source code in src/bigbrotr/services/dvm/utils.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def build_payment_required_event(
    request_event_id: str,
    customer_pubkey: str,
    price: int,
) -> EventBuilder:
    """Build a NIP-90 payment-required feedback event (kind 7000).

    Args:
        request_event_id: Hex ID of the request event.
        customer_pubkey: Hex pubkey of the requesting user.
        price: Required payment in millisats.

    Returns:
        Unsigned ``EventBuilder`` ready for signing and sending.
    """
    tags = [
        Tag.parse(["status", "payment-required", f"This query costs {price} millisats"]),
        Tag.parse(["e", request_event_id]),
        Tag.parse(["p", customer_pubkey]),
        Tag.parse(["amount", str(price)]),
    ]
    return EventBuilder(Kind(7000), "").tags(tags)

build_announcement_event 

build_announcement_event(
    d_tag: str,
    kind: int,
    name: str,
    about: str,
    tables: list[str],
) -> EventBuilder

Build a NIP-89 handler announcement event (kind 31990).

Parameters:

  • d_tag (str) –

    Unique handler identifier for the replaceable event.

  • kind (int) –

    NIP-90 request kind this handler supports.

  • name (str) –

    Human-readable handler name.

  • about (str) –

    Handler description.

  • tables (list[str]) –

    List of enabled table names.

Returns:

  • EventBuilder –

    Unsigned EventBuilder ready for signing and sending.

Source code in src/bigbrotr/services/dvm/utils.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def build_announcement_event(
    d_tag: str,
    kind: int,
    name: str,
    about: str,
    tables: list[str],
) -> EventBuilder:
    """Build a NIP-89 handler announcement event (kind 31990).

    Args:
        d_tag: Unique handler identifier for the replaceable event.
        kind: NIP-90 request kind this handler supports.
        name: Human-readable handler name.
        about: Handler description.
        tables: List of enabled table names.

    Returns:
        Unsigned ``EventBuilder`` ready for signing and sending.
    """
    tags = [
        Tag.parse(["d", d_tag]),
        Tag.parse(["k", str(kind)]),
    ]
    content = json.dumps({"name": name, "about": about, "tables": tables})
    return EventBuilder(Kind(31990), content).tags(tags)