Durable topics

RedisBackend

RedisBackend(
    url: str = DEFAULT_URL,
    *,
    prefix: str = "aether:",
    maxlen: int | None = DEFAULT_MAXLEN,
)

Connection and stream handling for durable topics.

Source code in python/aether/_redis.py

def __init__(
    self,
    url: str = DEFAULT_URL,
    *,
    prefix: str = "aether:",
    maxlen: int | None = DEFAULT_MAXLEN,
) -> None:
    if not HAVE_REDIS:
        raise RuntimeError(
            "durable topics need the redis package: pip install 'aether[redis]'"
        )
    self.url = url
    self.prefix = prefix
    self.maxlen = maxlen
    # Identifies this process so the tail can skip what it published
    # itself, which local subscribers already received directly.
    self.node_id = f"{os.getpid()}-{uuid.uuid4().hex[:8]}".encode()
    self._clients: dict[Any, Any] = {}
    self._lock = threading.Lock()

url instance-attribute

url = url

prefix instance-attribute

prefix = prefix

maxlen instance-attribute

maxlen = maxlen

node_id instance-attribute

node_id = f'{os.getpid()}-{uuid.uuid4().hex[:8]}'.encode()

key

key(topic: str) -> str

Source code in python/aether/_redis.py

def key(self, topic: str) -> str:
    return f"{self.prefix}{topic}"

client

client()

A client bound to the calling event loop.

redis-py's async connections belong to the loop that opened them, and Aether runs several worker loops, so each gets its own pool rather than sharing one that would break the moment a second loop touched it.

Source code in python/aether/_redis.py

def client(self):
    """A client bound to the calling event loop.

    redis-py's async connections belong to the loop that opened them, and
    Aether runs several worker loops, so each gets its own pool rather than
    sharing one that would break the moment a second loop touched it.
    """
    loop = asyncio.get_running_loop()
    with self._lock:
        existing = self._clients.get(loop)
        if existing is None:
            existing = aioredis.Redis.from_url(
                self.url,
                decode_responses=False,
                retry=Retry(ExponentialBackoff(cap=1.0, base=0.05), CONNECT_RETRIES),
                retry_on_error=[RedisConnectionError, RedisTimeoutError],
            )
            self._clients[loop] = existing
        return existing

ensure_group async

ensure_group(
    topic: str, group: str, start: str = "0"
) -> None

Create the group if it does not exist. start of "0" means a new group sees everything still in the stream; "$" means only new messages.

Source code in python/aether/_redis.py

async def ensure_group(self, topic: str, group: str, start: str = "0") -> None:
    """Create the group if it does not exist. `start` of "0" means a new
    group sees everything still in the stream; "$" means only new messages.
    """
    try:
        await self.client().xgroup_create(
            self.key(topic), group, id=start, mkstream=True
        )
    except ResponseError as exc:
        if "BUSYGROUP" not in str(exc):
            raise

publish async

publish(topic: str, value: Any) -> str

Source code in python/aether/_redis.py

async def publish(self, topic: str, value: Any) -> str:
    fields = {b"d": _encode(value), b"n": self.node_id}
    kwargs: dict[str, Any] = {}
    if self.maxlen:
        kwargs = {"maxlen": self.maxlen, "approximate": True}
    entry_id = await self.client().xadd(self.key(topic), fields, **kwargs)
    return entry_id.decode() if isinstance(entry_id, bytes) else entry_id

tail async

tail(
    topic: str, start: str = "$", *, skip_own: bool = True
)

Yield (id, value) for messages appended after start.

Runs forever; cancel the task to stop it.

Source code in python/aether/_redis.py

async def tail(self, topic: str, start: str = "$", *, skip_own: bool = True):
    """Yield (id, value) for messages appended after `start`.

    Runs forever; cancel the task to stop it.
    """
    key = self.key(topic)
    last = start
    client = self.client()
    while True:
        result = await client.xread({key: last}, block=BLOCK_MS, count=64)
        for _stream, entries in result or ():
            for entry_id, fields in entries:
                last = entry_id
                if skip_own and fields.get(b"n") == self.node_id:
                    continue
                ident = entry_id.decode() if isinstance(entry_id, bytes) else entry_id
                yield ident, _decode(fields.get(b"d"))

history async

history(
    topic: str, count: int = 100, start: str = "-"
) -> list

Recent messages, oldest first. For replay and for tests.

Source code in python/aether/_redis.py

async def history(self, topic: str, count: int = 100, start: str = "-") -> list:
    """Recent messages, oldest first. For replay and for tests."""
    entries = await self.client().xrange(self.key(topic), min=start, count=count)
    return [
        (
            (eid.decode() if isinstance(eid, bytes) else eid),
            _decode(fields.get(b"d")),
        )
        for eid, fields in entries
    ]

pending async

pending(topic: str, group: str) -> int

Messages this group has been given but not had acked.

A group-level question, so it needs no consumer. Creating one just to ask would add a member to the group that never reads anything.

Source code in python/aether/_redis.py

async def pending(self, topic: str, group: str) -> int:
    """Messages this group has been given but not had acked.

    A group-level question, so it needs no consumer. Creating one just to
    ask would add a member to the group that never reads anything.
    """
    try:
        info = await self.client().xpending(self.key(topic), group)
    except ResponseError:
        return 0  # the group does not exist yet
    return int(info["pending"]) if info else 0

length async

length(topic: str) -> int

Source code in python/aether/_redis.py

async def length(self, topic: str) -> int:
    return await self.client().xlen(self.key(topic))

trim async

trim(topic: str, maxlen: int) -> int

Source code in python/aether/_redis.py

async def trim(self, topic: str, maxlen: int) -> int:
    return await self.client().xtrim(self.key(topic), maxlen=maxlen, approximate=False)

close async

close() -> None

Source code in python/aether/_redis.py

async def close(self) -> None:
    with self._lock:
        clients = list(self._clients.values())
        self._clients.clear()
    for client in clients:
        try:
            await client.aclose()
        except Exception:  # noqa: BLE001 - shutting down anyway
            pass

Consumer

Consumer(
    backend: RedisBackend,
    topic: str,
    group: str,
    name: str,
    *,
    count: int = 32,
    block_ms: int = BLOCK_MS,
    claim_after_ms: int | None = 60000,
    model: Any = None,
)

A member of a consumer group. Async-iterable over Message.

Each message goes to exactly one member of the group, and stays pending until acknowledged, which is what makes delivery at-least-once rather than at-most-once.

Source code in python/aether/_redis.py

def __init__(
    self,
    backend: "RedisBackend",
    topic: str,
    group: str,
    name: str,
    *,
    count: int = 32,
    block_ms: int = BLOCK_MS,
    claim_after_ms: int | None = 60_000,
    model: Any = None,
) -> None:
    self.backend = backend
    self.topic = topic
    self.group = group
    self.name = name
    self.count = count
    self.block_ms = block_ms
    self.claim_after_ms = claim_after_ms
    self.model = model
    self._buffer: deque[Message] = deque()
    # First pass re-reads anything this consumer was holding when it died.
    self._recovering = True
    self._closed = False

backend instance-attribute

backend = backend

topic instance-attribute

topic = topic

group instance-attribute

group = group

name instance-attribute

name = name

count instance-attribute

count = count

block_ms instance-attribute

block_ms = block_ms

claim_after_ms instance-attribute

claim_after_ms = claim_after_ms

model instance-attribute

model = model

start async

start() -> Consumer

Source code in python/aether/_redis.py

async def start(self) -> "Consumer":
    await self.backend.ensure_group(self.topic, self.group)
    return self

ack async

ack(message_id: str) -> None

Source code in python/aether/_redis.py

async def ack(self, message_id: str) -> None:
    client = self.backend.client()
    await client.xack(self.backend.key(self.topic), self.group, message_id)

pending async

pending() -> int

How many messages this group has delivered but not had acked.

Source code in python/aether/_redis.py

async def pending(self) -> int:
    """How many messages this group has delivered but not had acked."""
    client = self.backend.client()
    info = await client.xpending(self.backend.key(self.topic), self.group)
    return int(info["pending"]) if info else 0

close

close() -> None

Source code in python/aether/_redis.py

def close(self) -> None:
    self._closed = True

Message dataclass

Message(
    id: str, data: Any, _consumer: Consumer | None = None
)

One message from a consumer group, awaiting acknowledgement.

id instance-attribute

id: str

data instance-attribute

data: Any

ack async

ack() -> None

Mark it done. Until this is called the message stays pending and will be redelivered if this consumer dies.

Source code in python/aether/_redis.py

async def ack(self) -> None:
    """Mark it done. Until this is called the message stays pending and
    will be redelivered if this consumer dies."""
    if self._consumer is not None:
        await self._consumer.ack(self.id)