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.3.0.tar.gz (37.7 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.3.0-py3-none-any.whl (49.4 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: celery_salt-1.3.0.tar.gz
  • Upload date:
  • Size: 37.7 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.3.0.tar.gz
Algorithm Hash digest
SHA256 9d6783196ff75a8336c83b38080409c78666a2ff25392b26777d4ed67a86e0b1
MD5 20bbf934e209edb6e91495aa5b822efe
BLAKE2b-256 f9bb89a368a7207b9dd6e844b5effeeff8dfafd7fdbbba66b96931d15ae8797e

See more details on using hashes here.

File details

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

File metadata

  • Download URL: celery_salt-1.3.0-py3-none-any.whl
  • Upload date:
  • Size: 49.4 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.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 5a79ac20566812159abffeecf799f4c2cfbbe12566e7858081bf0d2e57bdcb6f
MD5 8374f0f9897b48404d6fc24307725d74
BLAKE2b-256 313e408ae90c981cb737a6894dca89767279a7b6223bb8b5f4d1f211dd2b734c

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