nostr_tools.Filter¶

class nostr_tools.Filter[source]¶

Bases: object

Nostr event filter following NIP-01 subscription specification.

This class creates filters for querying and subscribing to events from Nostr relays. Filters support multiple criteria including event IDs, authors, kinds, time ranges, result limits, and tag-based filtering. All filter criteria are optional and use AND logic when combined.

Examples

Filter by event kind:

>>> # Get last 10 text notes
>>> filter = Filter(kinds=[1], limit=10)

Filter by author:

>>> # Get events from specific author
>>> filter = Filter(authors=["abc123..."])

Filter by time range:

>>> import time
>>> # Events from last hour
>>> filter = Filter(
...     kinds=[1],
...     since=int(time.time()) - 3600
... )

Filter by tags:

>>> # Get replies to a specific event
>>> filter = Filter(
...     kinds=[1],
...     e=["original_event_id"]  # Tag filter using kwargs
... )

Complex filter:

>>> # Reactions to my notes in last 24 hours
>>> filter = Filter(
...     kinds=[7],  # Reactions
...     e=["my_note_id1", "my_note_id2"],
...     since=int(time.time()) - 86400,
...     limit=100
... )

Create from dictionary:

>>> filter_data = {
...     "kinds": [1],
...     "authors": ["abc123..."],
...     "limit": 10
... }
>>> filter = Filter.from_dict(filter_data)

Raises:: FilterValidationError – If any attribute value is invalid (e.g., invalid hex strings, negative timestamps, invalid tag names).

Initialize Filter instance with subscription criteria.

This constructor accepts standard filter parameters plus tag filters as keyword arguments. Tag filters use single-letter keys corresponding to tag names (e.g., e for event references, p for pubkey references).

Parameters:

ids (Optional[list[str]]) – List of event IDs to match (64-char hex).
authors (Optional[list[str]]) – List of author pubkeys (64-char hex).
kinds (Optional[list[int]]) – List of event kinds (0-65535).
since (Optional[int]) – Unix timestamp for minimum event age (>= 0).
until (Optional[int]) – Unix timestamp for maximum event age (>= 0).
limit (Optional[int]) – Maximum number of events to return (>= 0).
**tags (list[str]) – Tag filters as keyword arguments. Single-letter keys only (a-z, A-Z). Example: e=[‘event_id’], p=[‘pubkey’], t=[‘hashtag’]

Examples

Basic filter:

>>> filter = Filter(kinds=[1], limit=10)

With tag filters:

>>> filter = Filter(
...     kinds=[1],
...     e=['event_id_to_reply_to'],  # #e tag
...     p=['mentioned_pubkey']       # #p tag
... )

Time-based filter:

>>> filter = Filter(
...     kinds=[1],
...     since=1234567890,
...     until=1234567999
... )

Methods

`__init__`([ids, authors, kinds, since, ...])	Initialize Filter instance with subscription criteria.
`from_dict`(data)	Create Filter from dictionary representation.
`from_subscription_filter`(data)	Create Filter from subscription filter dictionary.
`to_dict`()	Convert Filter to dictionary representation.
`validate`()	Validate the Filter instance.

Attributes

`authors`	List of author public keys (64-char lowercase hex strings).
`ids`	List of event IDs to match (64-char lowercase hex strings).
`is_valid`	Check if the Filter is valid without raising exceptions.
`kinds`	0=metadata, 1=text note, 3=contacts, 7=reaction
`limit`	Maximum number of events to return.
`since`	Unix timestamp (seconds).
`subscription_filter`	Build the subscription filter dictionary for relay communication.
`until`	Unix timestamp (seconds).
`tags`	{"e": ["event_id1", "event_id2"], "p": ["pubkey1"]}

__init__(ids=None, authors=None, kinds=None, since=None, until=None, limit=None, **tags)[source]¶

Initialize Filter instance with subscription criteria.

Parameters:

ids (Optional[list[str]]) – List of event IDs to match (64-char hex).
authors (Optional[list[str]]) – List of author pubkeys (64-char hex).
kinds (Optional[list[int]]) – List of event kinds (0-65535).
since (Optional[int]) – Unix timestamp for minimum event age (>= 0).
until (Optional[int]) – Unix timestamp for maximum event age (>= 0).
limit (Optional[int]) – Maximum number of events to return (>= 0).
**tags (list[str]) – Tag filters as keyword arguments. Single-letter keys only (a-z, A-Z). Example: e=[‘event_id’], p=[‘pubkey’], t=[‘hashtag’]

Return type:

None

Examples

Basic filter:

>>> filter = Filter(kinds=[1], limit=10)

With tag filters:

>>> filter = Filter(
...     kinds=[1],
...     e=['event_id_to_reply_to'],  # #e tag
...     p=['mentioned_pubkey']       # #p tag
... )

Time-based filter:

>>> filter = Filter(
...     kinds=[1],
...     since=1234567890,
...     until=1234567999
... )

ids: list[str] | None = None¶: List of event IDs to match (64-char lowercase hex strings). Events matching any ID in the list will be returned.

authors: list[str] | None = None¶: List of author public keys (64-char lowercase hex strings). Events from any author in the list will be returned.

kinds: list[int] | None = None¶

0=metadata, 1=text note, 3=contacts, 7=reaction

Type:: List of event kinds to match (0-65535). Events matching any kind in the list will be returned. Common kinds

since: int | None = None¶: Unix timestamp (seconds). Only events created at or after this time will be returned. Must be >= 0.

until: int | None = None¶: Unix timestamp (seconds). Only events created at or before this time will be returned. Must be >= 0.

limit: int | None = None¶: Maximum number of events to return. Must be >= 0. Relays may impose their own limits.

tags: dict[str, list[str]] | None¶

{“e”: [“event_id1”, “event_id2”], “p”: [“pubkey1”]}

Type:: Tag-based filters. Keys are single alphabetic characters (a-z, A-Z), values are lists of strings. Example

__post_init__()[source]¶

Validate and normalize filter after initialization.

This method is automatically called after the dataclass is created. It normalizes empty collections to None, converts hex strings to lowercase, removes invalid tag filters, and validates all filter criteria.

Raises:: FilterValidationError – If validation fails after normalization.
Return type:: None

validate()[source]¶

Validate the Filter instance.

Raises:: FilterValidationError – If any attribute has an invalid value
Return type:: None

property subscription_filter: dict[str, Any]¶

Build the subscription filter dictionary for relay communication.

Converts the Filter instance into a dictionary format suitable for sending to Nostr relays in REQ messages. Tag filters are converted to the #<tag_name> format required by the protocol.

Returns:

Dictionary suitable for Nostr subscription filtering.: Only includes non-None filter criteria. Tag filters are prefixed with # (e.g., {“#e”: [“event_id”], “#p”: [“pubkey”]}).

Return type:

dict[str, Any]

Examples

Basic subscription filter:

>>> filter = Filter(kinds=[1], limit=10)
>>> print(filter.subscription_filter)
{"kinds": [1], "limit": 10}

With tag filters:

>>> filter = Filter(kinds=[1], e=["event_id"], p=["pubkey"])
>>> print(filter.subscription_filter)
{"kinds": [1], "#e": ["event_id"], "#p": ["pubkey"]}

Send to relay:

>>> sub_filter = filter.subscription_filter
>>> message = ["REQ", subscription_id, sub_filter]
>>> await ws.send_str(json.dumps(message))

property is_valid: bool¶

Check if the Filter is valid without raising exceptions.

This property attempts validation and returns True if successful, False otherwise. Unlike validate(), this method does not raise exceptions, making it safe for conditional checks.

Returns:

True if the filter passes all validation checks,: False if validation fails for any reason.

Return type:

bool

Examples

>>> filter = Filter(kinds=[1], limit=10)
>>> if filter.is_valid:
...     events = await fetch_events(client, filter)
... else:
...     print("Invalid filter configuration")

>>> # Check before using
>>> filter = Filter(authors=["invalid"])
>>> if not filter.is_valid:
...     print("Filter validation failed")

classmethod from_subscription_filter(data)[source]¶

Create Filter from subscription filter dictionary.

This method converts a subscription filter dictionary (typically from a REQ message) into a Filter object. It handles the conversion of #-prefixed tag filters back to tag dictionary format.

Parameters:

data (dict[str, Any]) – Dictionary containing subscription filter data. Can include standard filter fields and #-prefixed tag filters like {“#e”: [“event_id”], “#p”: [“pubkey”]}.

Returns:

An instance of Filter with all criteria properly converted.

Return type:

Filter

Raises:

TypeError – If data is not a dictionary.
FilterValidationError – If data contains invalid filter criteria.

Examples

Parse REQ message filter:

>>> req_filter = {
...     "kinds": [1],
...     "#e": ["abc123..."],
...     "#p": ["def456..."],
...     "limit": 10
... }
>>> filter = Filter.from_subscription_filter(req_filter)
>>> print(filter.tags)
{'e': ['abc123...'], 'p': ['def456...']}

Convert relay subscription:

>>> relay_filter = {"kinds": [1, 3], "#t": ["nostr"]}
>>> filter = Filter.from_subscription_filter(relay_filter)
>>> # Use filter object with standard methods

classmethod from_dict(data)[source]¶

Create Filter from dictionary representation.

This method creates a Filter instance from a dictionary with standard Filter format (not subscription format). Tag filters should be provided in the “tags” dictionary field without # prefixes.

Parameters:

data (dict[str, Any]) – Dictionary containing filter data with keys: - ids (Optional[list[str]]): Event IDs - authors (Optional[list[str]]): Author public keys - kinds (Optional[list[int]]): Event kinds - since (Optional[int]): Minimum timestamp - until (Optional[int]): Maximum timestamp - limit (Optional[int]): Maximum number of events - tags (Optional[dict]): Tag filters without # prefix

Returns:

An instance of Filter created from the dictionary.

Return type:

Filter

Raises:

TypeError – If data is not a dictionary.
FilterValidationError – If filter data is invalid.

Examples

Create from stored configuration:

>>> config = {
...     "kinds": [1],
...     "limit": 10,
...     "tags": {"e": ["event_id"], "p": ["pubkey"]}
... }
>>> filter = Filter.from_dict(config)

Parse from JSON:

>>> import json
>>> json_str = '{"kinds": [1], "authors": ["abc123..."], "limit": 50}'
>>> filter_dict = json.loads(json_str)
>>> filter = Filter.from_dict(filter_dict)

Load from database:

>>> filter_data = db.filters.find_one({"name": "my_filter"})
>>> filter = Filter.from_dict(filter_data)

to_dict()[source]¶

Convert Filter to dictionary representation.

This method serializes the Filter instance into a dictionary format suitable for JSON encoding, storage, or transmission. The output uses standard Filter format with tag filters in the “tags” dictionary field (not subscription format with # prefixes).

Returns:

Dictionary representation of Filter with keys:

ids (Optional[list[str]]): Event IDs or None
authors (Optional[list[str]]): Author public keys or None
kinds (Optional[list[int]]): Event kinds or None
since (Optional[int]): Minimum timestamp or None
until (Optional[int]): Maximum timestamp or None
limit (Optional[int]): Maximum number of events or None
tags (Optional[dict]): Tag filters or None

Return type:

dict[str, Any]

Examples

Serialize to JSON:

>>> filter = Filter(kinds=[1], limit=10, e=["event_id"])
>>> filter_dict = filter.to_dict()
>>> import json
>>> json_str = json.dumps(filter_dict)

Store in database:

>>> filter = Filter(kinds=[1], authors=["abc123..."])
>>> filter_data = filter.to_dict()
>>> db.filters.insert_one({"name": "my_filter", **filter_data})

Create configuration:

>>> filters_config = {
...     "notes": Filter(kinds=[1], limit=100).to_dict(),
...     "metadata": Filter(kinds=[0]).to_dict()
... }