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.

Easy Python 3.9+ Aug 9, 2026 Streaming & messaging 13 views 0 copies

Python code

45 lines
Python 3.9+
from 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

stdout
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

  1. Use a list and manually pop from the front when exceeding max size instead of `deque`.
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Streaming & messaging

Related tutorials and quizzes for this topic.