Building Real-time Web Applications with PynneX and FastAPI

[nexconnectio]

I've created three examples demonstrating how to build real-time web applications with FastAPI, using Python worker threads and event-driven patterns. Rather than fully replacing established solutions like Celery or Redis, this approach aims to offer a lighter alternative for scenarios where distributed task queues may be overkill. No locks, no manual concurrency headaches — just emitters in the worker and listeners on the main thread or other workers.

Why PynneX?

While there are several solutions for handling concurrent tasks in Python, each comes with its own trade-offs:

• Celery: Powerful for distributed tasks but might be overkill for simpler scenarios
• Redis: Great as an in-memory data store, though it adds external dependencies
• RxPY: Comprehensive reactive programming but has a steeper learning curve
• asyncio.Queue: Basic but needs manual implementation of high-level patterns
• Qt's Signals & Slots: xcellent pattern but tied to GUI frameworks

PynneX takes the proven emitter-listener pattern and makes it seamlessly work with asyncio for general Python applications:

• Lightweight: No external dependencies beyond Python stdlib
• Focused: Designed specifically for thread-safe communication between threads
• Simple: Clean and intuitive through declarative event handling
• Flexible: Not tied to any UI framework or architecture

For simpler scenarios where you just need clean thread communication without distributed task queues, PynneX provides a lightweight alternative.

🍓 1. Berry Checker (Basic)

A minimal example showing the core concepts:

• Worker thread for background processing
• WebSocket real-time updates
• Event-driven task handling

Demo | View Code

📱 2. QR Code Generator (Intermediate)

Building on the basic concepts and adding:

• Real-time image generation
• Base64 image encoding/decoding
• Clean Controller-Worker pattern

Thread safety comes for free: the worker generates QR codes and emits them, the main thread listens and updates the UI. No manual synchronization needed.

Demo | View Code

📈 3. Stock Monitor (Advanced)

A full-featured example showcasing:

• Multiple worker threads
• Interactive data grid (ag-Grid)
• Real-time charts (eCharts)
• Price alert system
• Clean architecture

Demo | View Code

Quick Start

Clone repository

git clone https://github.com/nexconnectio/pynnex.git
cd pynnex

Install dependencies

pip install fastapi python-socketio uvicorn

Run any example

python examples/fastapi_socketio_simple.py
python examples/fastapi_socketio_qr.py
python examples/fastapi_socketio_stock_monitor.py

Then open http://localhost:8000 in your browser.

Key Features

• Python worker threads for background processing
• WebSocket for real-time updates
• Event-driven architecture with emitter-listener pattern
• Clean separation of concerns
• No complex dependencies

Technical Details

PynneX provides a lightweight layer for:

1. emitter-listener pattern for event handling
2. Worker thread management
3. Thread-safe task queuing

Built with:

• FastAPI for the web framework
• SocketIO for WebSocket communication
• Python's built-in threading and asyncio

Learn More

• PynneX Documentation
• GitHub Repository
• PyPI Package

The examples above demonstrate how to build real-time web applications with clean thread communication patterns, without the complexity of traditional task queue systems.