Build a Streaming Messaging Helper in Python
Create a simple message stream class that stores recent messages, sends user messages, and retrieves history or latest messages with timestamps.
Python code
45 linesfrom collections import deque
from dataclasses import dataclass
from datetime import datetime
import time
@dataclass
class Message:
user: str
text: str
timestamp: str = ""
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().strftime("%H:%M:%S")
class MessageStream:
def __init__(self, max_size=5):
self._messages = deque(maxlen=max_size)
def send(self, user: str, text: str) -> Message:
message = Message(user=user, text=text)
self._messages.append(message)
return message
def latest(self, count=1):
recent = list(self._messages)[-count:]
return recent[::-1]
def history(self):
return list(self._messages)
def __len__(self):
return len(self._messages)
if __name__ == "__main__":
stream = MessageStream(max_size=5)
stream.send("alice", "hello there")
time.sleep(0.5)
stream.send("bob", "hi alice!")
stream.send("alice", "how are you?")
print("All messages:", stream.history())
print("Latest 2:", stream.latest(2))
Output
All messages: [Message(user='alice', text='hello there', timestamp='12:34:56'), Message(user='bob', text='hi alice!', timestamp='12:34:57'), Message(user='alice', text='how are you?', timestamp='12:34:57')]
Latest 2: [Message(user='alice', text='how are you?', timestamp='12:34:57'), Message(user='bob', text='hi alice!', timestamp='12:34:57')]
How it works
The Message dataclass automatically adds a timestamp when created, using __post_init__ to format the current time. The MessageStream class uses a deque with a max length to keep only the newest messages, discarding older ones. The send method appends a new message and returns it, while latest returns the most recent messages in reverse order. history returns all stored messages as a list. This pattern models a simple in-memory message queue for real-time applications.
Common mistakes
- Forgetting to set `maxlen` in deque, which would store unlimited messages and break the size limit.
- Returning messages in chronological order from `latest` instead of newest-first.
- Not using `__post_init__` to auto-generate timestamps, leading to empty or manual timestamp handling.
Variations
- Use a list and manually pop from the front when exceeding max size instead of `deque`.
- Add a `clear()` method to reset the stream, or persist messages to a file.
Real-world use cases
- Implementation of a chat room buffer that shows the last few messages to new users.
- In-memory event stream for a live dashboard that keeps recent log entries for display.
- A lightweight message bus for microservices to store and replay recent events in a subscriber.
Sponsored
More from Streaming & messaging
- At Most Once Fire-and-Forget Mock in Python easy
- Batch Consume Process Commit Pattern in Python medium
- Dead Letter Queue Failed Messages List Mock in Python easy
- Dedupe processed message IDs in Python easy
- Event Envelope with Schema Version Field in Python easy
- Event sourcing append store replay in Python easy
Keep learning
Related tutorials and quizzes for this topic.