pymongoose: python bindings for mongoose using with cython.

[shakfu]

Overview

pymongoose is a thin cython wrapper over the native mongoose C library and includes coverage of:

• HTTP/HTTPS: Server and client with TLS support, chunked transfer encoding, SSE
• WebSocket/WSS: Full WebSocket support with text/binary frames over TLS
• MQTT/MQTTS: Publish/subscribe messaging with QoS support
• TCP/UDP: Raw socket support with custom protocols
• DNS: Asynchronous hostname resolution
• SNTP: Network time synchronization
• TLS/SSL: Certificate-based encryption with custom CA support
• Timers: Periodic callbacks with precise timing control
• Flow Control: Backpressure handling and buffer management
• Authentication: HTTP Basic Auth, MQTT credentials
• JSON Parsing: Built-in JSON extraction utilities
• URL Encoding: Safe URL parameter encoding

The wrapper is:

• Compatible with Python 3.9+: Modern Python with type hints
• Comprehensive: 210 tests, 100% pass rate

It has:

• Production Examples: 17 complete examples from Mongoose tutorials
• GIL Optimization: 21 methods release GIL for true parallel execution
• High Performance: 60k+ req/sec (6-37x faster than pure Python frameworks)

A Few Examples

Simple HTTP Server

import signal
from pymongoose import Manager, MG_EV_HTTP_MSG

shutdown_requested = False

def signal_handler(sig, frame):
    global shutdown_requested
    shutdown_requested = True

def handler(conn, event, data):
    if event == MG_EV_HTTP_MSG:
        conn.reply(200, "Hello, World!")

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

mgr = Manager(handler)
mgr.listen("http://0.0.0.0:8000", http=True)

print("Server running on http://localhost:8000. Press Ctrl+C to stop.")
try:
    while not shutdown_requested:
        mgr.poll(100)
    print("Shutting down...")
finally:
    mgr.close()

Serve Static Files

import signal
from pymongoose import Manager, MG_EV_HTTP_MSG

shutdown_requested = False

def signal_handler(sig, frame):
    global shutdown_requested
    shutdown_requested = True

def handler(conn, event, data):
    if event == MG_EV_HTTP_MSG:
        conn.serve_dir(data, root_dir="./public")

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

mgr = Manager(handler)
mgr.listen("http://0.0.0.0:8000", http=True)

try:
    while not shutdown_requested:
        mgr.poll(100)
finally:
    mgr.close()

WebSocket Echo Server

import signal
from pymongoose import Manager, MG_EV_HTTP_MSG, MG_EV_WS_MSG

shutdown_requested = False

def signal_handler(sig, frame):
    global shutdown_requested
    shutdown_requested = True

def handler(conn, event, data):
    if event == MG_EV_HTTP_MSG:
        conn.ws_upgrade(data)  # Upgrade HTTP to WebSocket
    elif event == MG_EV_WS_MSG:
        conn.ws_send(data.text)  # Echo back

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

mgr = Manager(handler)
mgr.listen("http://0.0.0.0:8000", http=True)

try:
    while not shutdown_requested:
        mgr.poll(100)
finally:
    mgr.close()