Question: I'm working on implementing the observer pattern in Python using asyncio and async iterators. The goal is to create a “change stream” where tasks can add changes, and other tasks can subscribe to these changes as asynchronous iterators. I'm trying to create an interface similar to broadcast streams in Dart. Here's a simplified version of what I have so far: from asyncio import Condition class ChangeStream: def __init__(self): self._condition = Condition() self._change = None async def add_change(self, change): async with self._condition: self._change = change self._condition.notify_all() async def __aiter__(self): async with self._condition: while True: await self._condition.wait() yield self._change  Is this the best approach to implementing the observer pattern? Are there more efficient or easier-to-reason-about ways to implement this? Edit: The goal here is that observers can see all changes from after they subscribe, but none of the changes from before they subscribe.

Answer: From your description and comment, it appears you're really looking for a Publish-Subscribe pattern instead of an Observer pattern. They are similar, but where Observers are typically known by the subject and get updates directly, in Publish-Subscribe, the publisher publishes to a bus (like your Stream) that tracks consumers that are interested in the data. Here's an example with a Queue-based implementation, and some extra code to show how it all works - of course you could use the class differently, the TaskGroup is just one way of getting some tasks going: from asyncio import Queue, TaskGroup, sleep, run from random import random class ChangeStream: def __init__(self): self._subscribers = [] async def add_change(self, change): for queue in self._subscribers: await queue.put(change) async def __aiter__(self): queue = Queue() self._subscribers.append(queue) try: while True: yield (value := await queue.get()) if value is None: raise GeneratorExit except GeneratorExit: self._subscribers.remove(queue) async def produce_changes(stream: ChangeStream): for i in range(10): await sleep(random() * 3) print(f"Add value: {i}") await stream.add_change(i) print("Done adding values, writing None to stream.") await stream.add_change(None) async def consume_changes(stream: ChangeStream, name: str, start_delay: int): await sleep(start_delay) print(f"{name} starting...") async for value in stream: print(f"{name} received: {value}") if value is None: break async def main(): # create a stream, pass it to a producer task to publish to stream = ChangeStream() async with TaskGroup() as tg: tg.create_task(produce_changes(stream)) tg.create_task(consume_changes(stream, 'consumer 1', 0)) tg.create_task(consume_changes(stream, 'consumer 2', 5)) # start after 5 sec if __name__ == "__main__": run(main())  Output (example, due to randomness): consumer 1 starting... Add value: 0 consumer 1 received: 0 Add value: 1 consumer 1 received: 1 Add value: 2 consumer 1 received: 2 consumer 2 starting... Add value: 3 consumer 1 received: 3 consumer 2 received: 3 Add value: 4 consumer 1 received: 4 consumer 2 received: 4 Add value: 5 consumer 1 received: 5 consumer 2 received: 5 Add value: 6 consumer 1 received: 6 consumer 2 received: 6 Add value: 7 consumer 1 received: 7 consumer 2 received: 7 Add value: 8 consumer 1 received: 8 consumer 2 received: 8 Add value: 9 Done adding values, writing None to stream. consumer 1 received: 9 consumer 1 received: None consumer 2 received: 9 consumer 2 received: None