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.1.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.1-py3-none-any.whl (52.8 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: celery_salt-1.4.1.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.1.tar.gz
Algorithm Hash digest
SHA256 48d9d2088b67e7a58d39d8486792545b171c45efabd4dfc3fd05ef450130bd38
MD5 3f0f9c5b4bb93da51cd46b4e4e4d7542
BLAKE2b-256 3850ef561bfb9b99b53eba9451bcb7a03dfdb11a5de5e24b014d0e7eb6582ce7

See more details on using hashes here.

File details

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

File metadata

  • Download URL: celery_salt-1.4.1-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.1-py3-none-any.whl
Algorithm Hash digest
SHA256 caa9c18f6104b1d6454ceb411d13acf1ad898970e0bc7a6f5b998351b81f3699
MD5 a920713f90097dd15289cd07accf099b
BLAKE2b-256 f89d6c85ded2df6f9869c9b4e866ec041651f76434b537388464005fb26b724e

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