Open source · Python · MIT License

Build microservices
with clean Python

Envoxy is a Python microservice framework that unifies HTTP, ZeroMQ, MQTT, PostgreSQL, Redis, CouchDB, and Celery behind a single, decorator-driven API, so you write business logic, not glue code.

View on GitHub Quick Reference
views.py 
import envoxy
from envoxy import View, on, auth_required, cache, pgsqlc
from .models import Product


@on(endpoint='/api/v1/products/{id:str}', protocols=['http'])
class ProductView(View):

    @cache(ttl=60)
    @auth_required()
    def get(self, request, id, **kwargs):
        with pgsqlc.sa_session('primary') as session:
            product = session.query(Product).filter_by(id=id).first()
            return envoxy.Response(product)

    @auth_required()
    def post(self, request, **kwargs):
        data = request.get_json()
        with pgsqlc.sa_session('primary') as session:
            p = Product(**data)
            session.add(p)
            return envoxy.Response(p, status=201)


# MQTT subscriber: same pattern, different protocol
@on(endpoint='devices/+/telemetry', protocols=['mqtt'], server='broker')
class TelemetryView(View):

    @envoxy.log_event
    def on_event(self, data, **kwargs):
        # data is the decoded MQTT payload
        pgsqlc.sa_session('primary').__enter__()  # store telemetry
Everything you need, nothing you don't
One import surface. Multiple transports. Zero boilerplate.
⚡

@on decorator routing

Declare endpoints and protocols declaratively. HTTP, MQTT, and ZeroMQ routes share the same class-based view pattern. Swap the protocol without touching business logic.

🔌

ZeroMQ inter-service messaging

zmqc provides synchronous request-reply between services over ZeroMQ sockets. No message broker required for in-cluster RPC.

🗄️

SQLAlchemy ORM built-in

EnvoxyBase models, pgsqlc.sa_session() context managers, @pgsqlc.sa_transactional decorator, and Alembic migrations, all wired to your config.

📡

MQTT pub/sub

Subscribe to MQTT topics with on_event handlers decorated with @log_event. The broker address and client lifecycle are managed by the framework.

🔒

Auth decorators

@auth_required() and @auth_anonymous_allowed() sit at the method level. Pluggable backends inject headers into kwargs so handlers stay clean.

⚙️

Response caching

@cache(ttl=N) transparently caches GET responses in Redis. Cache key is derived from the full endpoint path and request params automatically.

🗃️

Multiple data stores

First-class dispatchers for PostgreSQL (pgsqlc), Redis (redisc), and CouchDB (couchdbc), all configured via a single JSON config file.

🌿

Celery task queue

celeryc gives you async task dispatch from any view handler. Background jobs, scheduled tasks, and retries with the same config-driven setup.

📋

Structured logging

Levelled log system (log.system, log.trace, log.verbose, log.error) with colour-coded output and error reference codes built in.

Integrations
Every dispatcher ships inside the package. No plugins, no extras, no version hell.
HTTP (Flask/Gunicorn)
ZeroMQ (zmqc)
MQTT (mqttc)
PostgreSQL (pgsqlc + SQLAlchemy)
Redis (redisc)
CouchDB (couchdbc)
Celery (celeryc)
Alembic migrations
JWT / Auth backends
Response cache (Redis-backed)
Real ORM. No magic strings.
EnvoxyBase extends SQLAlchemy declarative base. Full type safety, Alembic-tracked migrations, raw SQL escape hatch via to_sa_text.
models.py 
from envoxy.db.orm.base import EnvoxyBase
from envoxy.db.orm.sqltypes import (
    String, Integer, JSON
)
from envoxy.db.orm.schema import Column, Index


class Product(EnvoxyBase):
    sku         = Column(String(64),  nullable=False)
    name        = Column(String(255), nullable=False)
    price_cents = Column(Integer,     default=0)
    cdata       = Column(JSON,        nullable=True)

    __table_args__ = (
        Index("idx_products_sku", "sku"),
    )
queries.py 
from envoxy import pgsqlc


# ORM: session context manager
def get_by_sku(server_key, sku):
    with pgsqlc.sa_session(server_key) as s:
        return s.query(Product)\
                 .filter_by(sku=sku).first()


# Transactional decorator
@pgsqlc.sa_transactional('primary')
def find_or_create(session=None, sku=None):
    p = session.query(Product)\
               .filter_by(sku=sku).first()
    if not p:
        p = Product(sku=sku, price_cents=0)
        session.add(p)
    return p