Metadata-Version: 2.4
Name: DefStream
Version: 0.1.22
Summary: Synchronous message processing with pluggable broker adapters
Home-page: https://github.com/yong5eon/DefStream
Author: Heo Yongseon
Project-URL: Source, https://github.com/yong5eon/DefStream
Project-URL: Issues, https://github.com/yong5eon/DefStream/issues
Project-URL: Releases, https://github.com/yong5eon/DefStream/releases
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: rabbitmq
Requires-Dist: pika>=1.3.2; extra == "rabbitmq"
Provides-Extra: redis
Requires-Dist: redis>=5.1.1; extra == "redis"
Dynamic: author
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: license-file
Dynamic: project-url
Dynamic: provides-extra
Dynamic: requires-python
Dynamic: summary

# DefStream

Synchronous event/message processing with pluggable broker adapters

DefStream processes broker messages with ordinary Python task methods and can run them in thread or process pools. It requires Python 3.9 or later.

## Installation

Install the client extra for the broker you use. A broker server must be available separately.

```shell
python -m pip install "DefStream[RabbitMQ]"
```

For Redis Streams, install `DefStream[Redis]`. To install both clients, use `DefStream[RabbitMQ,Redis]`. The plain `DefStream` installation includes neither broker client.

## Quick start

The following worker listens to a local RabbitMQ server at `127.0.0.1:5672`. The example credentials are for local development only.

```python
from signal import SIGINT, signal

from DefStream.Broker.Implements.RabbitMQ import Configuration, Connection, Reject, Subscription
from DefStream.Worker import Run, ThreadPool, Worker, WorkerSetup, declare


class Add(
  Run,
  parameters=Subscription(queue='example.add', error='example.add.error'),
  on_error=Reject(),
):
  def run(self, a: int, b: int):
    print(a + b)
    return a + b


class Prepare(WorkerSetup):
  def __call__(self, worker: Worker) -> None:
    for task in worker.tasks:
      declare(task=task, broker=worker.broker)


if __name__ == '__main__':
  worker = Worker(
    Connection(Configuration(host='127.0.0.1', port=5672, username='guest', password='guest')),
    ThreadPool(4),
    setups=Prepare(),
  )
  worker.add(Add)
  signal(SIGINT, lambda *_: worker.stop())
  worker.run()
```

Run the script after starting RabbitMQ. The worker declares its task queue and waits for messages. Store production credentials outside source code.

## License

See the `LICENSE` file included with the source distribution for the license terms.
