Skip to main content

Event-driven architecture library for Python that extends Celery with event publishing/subscribing patterns

Project description

CelerySalt

A schema-driven event API on top of Celery: publish/subscribe (broadcast) and RPC with Pydantic validation and a schema registry.

License: MIT Python 3.10+

Features

  • Pydantic schemas: type-checked event payloads with validation.
  • Broadcast: fire-and-forget pub/sub (one event to many subscribers).
  • RPC: request/response with response and error schema validation.
  • Schema registry: schema registration and lookup by topic/version.
  • Versioning: topic stays stable; versions are metadata.
  • Django integration: optional helpers for wiring queues and module discovery.
  • Protocol compatibility: interoperates with tchu-tchu (tchu_events exchange and _tchu_meta).

Quick Start

Installation

pip install celery-salt

Broadcast Example

from celery_salt import event, subscribe

# Define event schema
@event("user.signup.completed")
class UserSignupCompleted:
    user_id: int
    email: str
    company_id: int
    signup_source: str = "web"

# Publish event
UserSignupCompleted.publish(
    user_id=123,
    email="alice@example.com",
    company_id=1,
    signup_source="web"
)

# Subscribe to event
@subscribe("user.signup.completed")
def send_welcome_email(data: UserSignupCompleted):
    print(f"Sending welcome email to {data.email}")

RPC Example

from celery_salt import event, subscribe, RPCError

# Define RPC request/response schemas
@event("rpc.calculator.add", mode="rpc")
class CalculatorAddRequest:
    a: float
    b: float

@event.response("rpc.calculator.add")
class CalculatorAddResponse:
    result: float
    operation: str = "add"

@event.error("rpc.calculator.add")
class CalculatorAddError:
    error_code: str
    error_message: str

# Handler
@subscribe("rpc.calculator.add")
def handle_add(data: CalculatorAddRequest) -> CalculatorAddResponse:
    return CalculatorAddResponse(result=data.a + data.b, operation="add")

# Client call
response = CalculatorAddRequest.call(a=10, b=5, timeout=10)
print(f"Result: {response.result}")  # 15.0

Architecture

Publisher → RabbitMQ Exchange (tchu_events) → Subscribers
  • Exchange: tchu_events (topic exchange, protocol compatible)
  • Routing: Topic-based with wildcard support (user.*, #)
  • Serialization: JSON with Pydantic validation
  • Result Backend: Redis (required for RPC)

Documentation

Requirements

  • Python 3.10+
  • Celery 5.3+
  • RabbitMQ (message broker)
  • Redis (optional, required for RPC)

Examples

See the examples directory for complete working examples:

Run examples with Docker Compose:

cd examples
docker-compose up -d  # Starts RabbitMQ and Redis

Key Concepts

Event Schemas

Schemas are defined using Pydantic models and registered at import time:

@event("user.created")
class UserCreated:
    user_id: int
    email: str
    created_at: datetime

Publishing Events

# Broadcast (fire-and-forget)
UserCreated.publish(user_id=123, email="user@example.com", created_at=datetime.now())

# RPC (synchronous)
response = CalculatorAddRequest.call(a=10, b=5, timeout=10)

Subscribing to Events

@subscribe("user.created")
def handle_user_created(data: UserCreated):
    # Process event
    pass

RPC Response Validation

@event.response("rpc.calculator.add")
class CalculatorAddResponse:
    result: float

@event.error("rpc.calculator.add")
class CalculatorAddError:
    error_code: str
    error_message: str

Protocol Compatibility

CelerySalt maintains protocol compatibility with tchu-tchu:

  • Same exchange name: tchu_events
  • Same message format: _tchu_meta field
  • Same routing key conventions

This allows gradual migration: apps using celery-salt can communicate with apps still using tchu-tchu.

Development

git clone https://github.com/sigularusrex/celery-salt.git
cd celery-salt

# tests
python -m pytest

License

MIT License - see LICENSE file for details.

Contributing

Please open an issue or pull request on GitHub.

Links

  • GitHub: https://github.com/sigularusrex/celery-salt
  • Issues: https://github.com/sigularusrex/celery-salt/issues

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

celery_salt-1.4.3.tar.gz (40.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

celery_salt-1.4.3-py3-none-any.whl (52.8 kB view details)

Uploaded Python 3

File details

Details for the file celery_salt-1.4.3.tar.gz.

File metadata

  • Download URL: celery_salt-1.4.3.tar.gz
  • Upload date:
  • Size: 40.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.3.1 CPython/3.11.3 Darwin/24.5.0

File hashes

Hashes for celery_salt-1.4.3.tar.gz
Algorithm Hash digest
SHA256 8ddc63ba71b3a034e1e93fc1ad2e4afc0baa1619179f38d6acd81537b20450db
MD5 d83e6387fd55d6660e0a3fd02cf81e73
BLAKE2b-256 fb57db3f916270be2515812d05535f7e413b6d54a843f11afe26284909ba9585

See more details on using hashes here.

File details

Details for the file celery_salt-1.4.3-py3-none-any.whl.

File metadata

  • Download URL: celery_salt-1.4.3-py3-none-any.whl
  • Upload date:
  • Size: 52.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/2.3.1 CPython/3.11.3 Darwin/24.5.0

File hashes

Hashes for celery_salt-1.4.3-py3-none-any.whl
Algorithm Hash digest
SHA256 426af0bd51cd4fd9e5fc0ee949af3b181c3d85445abf7fc9d6cbd0fc43f673e5
MD5 2bced3fffd1663c84d312432e2acace4
BLAKE2b-256 811fe53abeceefd35d30db3ac3e15b38da6ead6fd5423993ea3a967d618502b5

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page