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.0.tar.gz (37.8 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.0-py3-none-any.whl (49.5 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: celery_salt-1.4.0.tar.gz
  • Upload date:
  • Size: 37.8 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.0.tar.gz
Algorithm Hash digest
SHA256 ed404463ffa11038002ef9b720f0efe3e984e3b1dd801c64656d469f6b68ad4b
MD5 0be56ebad269f93f8d7fbd6894b37d16
BLAKE2b-256 4671ccbbe8e20ca1e6ccea8d7366ac502cfbd389a401b25f950647051ce063a0

See more details on using hashes here.

File details

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

File metadata

  • Download URL: celery_salt-1.4.0-py3-none-any.whl
  • Upload date:
  • Size: 49.5 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.0-py3-none-any.whl
Algorithm Hash digest
SHA256 2c3b98cafe59f46ecbb30ebfaf25c20ea103d1f55ba6c2a2d31bfb4d83d51f5e
MD5 8cab11760fc7e6e8c84d36fd3ebcbf7a
BLAKE2b-256 f8076fe8845c366c0ffa5b3d9419974a952efc4209df4147f1b8a5e10bbf014a

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