# Taskiq - FastStream
<p align="center">
<a href="https://github.com/taskiq-python/taskiq-faststream/actions/workflows/test.yml" target="_blank">
<img src="https://github.com/taskiq-python/taskiq-faststream/actions/workflows/test.yml/badge.svg" alt="Tests status"/>
</a>
<a href="https://pypi.org/project/taskiq-faststream/" target="_blank">
<img src="https://img.shields.io/pypi/v/taskiq-faststream?label=pypi%20package" alt="Package version">
</a>
<a href="https://pepy.tech/project/taskiq-faststream" target="_blank">
<img src="https://static.pepy.tech/personalized-badge/taskiq-faststream?period=month&units=international_system&left_color=grey&right_color=blue" alt="downloads"/>
</a>
<a href="https://pypi.org/project/taskiq-faststream" target="_blank">
<img src="https://img.shields.io/pypi/pyversions/taskiq-faststream.svg" alt="Supported Python versions">
</a>
<a href="https://github.com/taskiq-python/taskiq-faststream/blob/master/LICENSE" target="_blank">
<img alt="GitHub" src="https://img.shields.io/github/license/taskiq-python/taskiq-faststream?color=%23007ec6">
</a>
</p>
---
The current package is just a wrapper for [**FastStream**](https://faststream.airt.ai/0.2/?utm_source=github&utm_medium=acquisition&utm_campaign=measure) objects to make them compatible with [**Taskiq**](https://taskiq-python.github.io/) library.
The main goal of it - provide **FastStream** with a great **Taskiq** tasks [scheduling](https://taskiq-python.github.io/guide/scheduling-tasks.html) feature.
## Installation
If you already have **FastStream** project to interact with your Message Broker, you can add scheduling to it by installing just a **taskiq-faststream**
```bash
pip install taskiq-faststream
```
If you starting with a clear project, you can specify **taskiq-faststream** broker by the following distributions:
```bash
pip install taskiq-faststream[rabbit]
# or
pip install taskiq-faststream[kafka]
# or
pip install taskiq-faststream[nats]
```
## Usage
The package gives you two classes: `AppWrapper` and `BrokerWrapper`
These are just containers for the related **FastStream** objects to make them **taskiq**-compatible
To create scheduling tasks for your broker, just wrap it to `BrokerWrapper` and use it like a regular **taskiq** Broker.
```python
# regular FastStream code
from faststream.nats import NatsBroker
broker = NatsBroker()
@broker.subscriber("test-subject")
async def handler(msg: str):
print(msg)
# taskiq-faststream scheduling
from taskiq.schedule_sources import LabelScheduleSource
from taskiq_faststream import BrokerWrapper, StreamScheduler
# wrap FastStream object
taskiq_broker = BrokerWrapper(broker)
# create periodic task
taskiq_broker.task(
message="Hi!",
# If you are using RabbitBroker, then you need to replace subject with queue.
# If you are using KafkaBroker, then you need to replace subject with topic.
subject="test-subject",
schedule=[{
"cron": "* * * * *",
}],
)
# create scheduler object
scheduler = StreamScheduler(
broker=taskiq_broker,
sources=[LabelScheduleSource(taskiq_broker)],
)
```
To run the scheduler, just use the following command
```bash
taskiq scheduler module:scheduler
```
Also, you can wrap your **FastStream** application the same way (allows to use lifespan events and AsyncAPI documentation):
```python
# regular FastStream code
from faststream import FastStream
from faststream.nats import NatsBroker
broker = NatsBroker()
app = FastStream(broker)
@broker.subscriber("test-subject")
async def handler(msg: str):
print(msg)
# wrap FastStream object
from taskiq_faststream import AppWrapper
taskiq_broker = AppWrapper(app)
# Code below omitted 👇
```
A little feature: instead of using a final `message` argument, you can set a message callback to collect information right before sending:
```python
async def collect_information_to_send():
return "Message to send"
taskiq_broker.task(
message=collect_information_to_send,
...,
)
```
Also, you can send a multiple message by one task call just using generator message callback with `yield`
```python
async def collect_information_to_send():
"""Sends 10 messages per task call."""
for i in range(10):
yield i
taskiq_broker.task(
message=collect_information_to_send,
...,
)
```
Raw data
{
"_id": null,
"home_page": null,
"name": "taskiq-faststream",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "FastStream, async, distributed, taskiq, tasks",
"author": null,
"author_email": "Taskiq team <taskiq@no-reply.com>, Nikita Pastukhov <nikita@pastukhov-dev.com>",
"download_url": "https://files.pythonhosted.org/packages/a2/06/60842ee94492550ba3e6d0b9380f52f7e73dfcc65bced8ebef84d895803f/taskiq_faststream-0.1.8.tar.gz",
"platform": null,
"description": "# Taskiq - FastStream\n\n<p align=\"center\">\n <a href=\"https://github.com/taskiq-python/taskiq-faststream/actions/workflows/test.yml\" target=\"_blank\">\n <img src=\"https://github.com/taskiq-python/taskiq-faststream/actions/workflows/test.yml/badge.svg\" alt=\"Tests status\"/>\n </a>\n <a href=\"https://pypi.org/project/taskiq-faststream/\" target=\"_blank\">\n <img src=\"https://img.shields.io/pypi/v/taskiq-faststream?label=pypi%20package\" alt=\"Package version\">\n </a>\n <a href=\"https://pepy.tech/project/taskiq-faststream\" target=\"_blank\">\n <img src=\"https://static.pepy.tech/personalized-badge/taskiq-faststream?period=month&units=international_system&left_color=grey&right_color=blue\" alt=\"downloads\"/>\n </a>\n <a href=\"https://pypi.org/project/taskiq-faststream\" target=\"_blank\">\n <img src=\"https://img.shields.io/pypi/pyversions/taskiq-faststream.svg\" alt=\"Supported Python versions\">\n </a>\n <a href=\"https://github.com/taskiq-python/taskiq-faststream/blob/master/LICENSE\" target=\"_blank\">\n <img alt=\"GitHub\" src=\"https://img.shields.io/github/license/taskiq-python/taskiq-faststream?color=%23007ec6\">\n </a>\n</p>\n\n---\n\nThe current package is just a wrapper for [**FastStream**](https://faststream.airt.ai/0.2/?utm_source=github&utm_medium=acquisition&utm_campaign=measure) objects to make them compatible with [**Taskiq**](https://taskiq-python.github.io/) library.\n\nThe main goal of it - provide **FastStream** with a great **Taskiq** tasks [scheduling](https://taskiq-python.github.io/guide/scheduling-tasks.html) feature.\n\n## Installation\n\nIf you already have **FastStream** project to interact with your Message Broker, you can add scheduling to it by installing just a **taskiq-faststream**\n\n```bash\npip install taskiq-faststream\n```\n\nIf you starting with a clear project, you can specify **taskiq-faststream** broker by the following distributions:\n\n```bash\npip install taskiq-faststream[rabbit]\n# or\npip install taskiq-faststream[kafka]\n# or\npip install taskiq-faststream[nats]\n```\n\n## Usage\n\nThe package gives you two classes: `AppWrapper` and `BrokerWrapper`\n\nThese are just containers for the related **FastStream** objects to make them **taskiq**-compatible\n\nTo create scheduling tasks for your broker, just wrap it to `BrokerWrapper` and use it like a regular **taskiq** Broker.\n\n```python\n# regular FastStream code\nfrom faststream.nats import NatsBroker\n\nbroker = NatsBroker()\n\n@broker.subscriber(\"test-subject\")\nasync def handler(msg: str):\n print(msg)\n\n# taskiq-faststream scheduling\nfrom taskiq.schedule_sources import LabelScheduleSource\nfrom taskiq_faststream import BrokerWrapper, StreamScheduler\n\n# wrap FastStream object\ntaskiq_broker = BrokerWrapper(broker)\n\n# create periodic task\ntaskiq_broker.task(\n message=\"Hi!\",\n # If you are using RabbitBroker, then you need to replace subject with queue.\n # If you are using KafkaBroker, then you need to replace subject with topic.\n subject=\"test-subject\",\n schedule=[{\n \"cron\": \"* * * * *\",\n }],\n)\n\n# create scheduler object\nscheduler = StreamScheduler(\n broker=taskiq_broker,\n sources=[LabelScheduleSource(taskiq_broker)],\n)\n```\n\nTo run the scheduler, just use the following command\n\n```bash\ntaskiq scheduler module:scheduler\n```\n\nAlso, you can wrap your **FastStream** application the same way (allows to use lifespan events and AsyncAPI documentation):\n\n```python\n# regular FastStream code\nfrom faststream import FastStream\nfrom faststream.nats import NatsBroker\n\nbroker = NatsBroker()\napp = FastStream(broker)\n\n@broker.subscriber(\"test-subject\")\nasync def handler(msg: str):\n print(msg)\n\n# wrap FastStream object\nfrom taskiq_faststream import AppWrapper\ntaskiq_broker = AppWrapper(app)\n\n# Code below omitted \ud83d\udc47\n```\n\nA little feature: instead of using a final `message` argument, you can set a message callback to collect information right before sending:\n\n```python\nasync def collect_information_to_send():\n return \"Message to send\"\n\ntaskiq_broker.task(\n message=collect_information_to_send,\n ...,\n)\n```\n\nAlso, you can send a multiple message by one task call just using generator message callback with `yield`\n\n```python\nasync def collect_information_to_send():\n \"\"\"Sends 10 messages per task call.\"\"\"\n for i in range(10):\n yield i\n\ntaskiq_broker.task(\n message=collect_information_to_send,\n ...,\n)\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "FastStream - taskiq integration to schedule FastStream tasks",
"version": "0.1.8",
"project_urls": {
"Homepage": "https://github.com/taskiq-python/taskiq-faststream",
"Source": "https://github.com/taskiq-python/taskiq-faststream",
"Tracker": "https://github.com/taskiq-python/taskiq-faststream/issues"
},
"split_keywords": [
"faststream",
" async",
" distributed",
" taskiq",
" tasks"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "4e55320ff98350683c375d142c18ef05bd802973fff2acbc817960d60e1f17f8",
"md5": "1d73ca135e3141b8c80351739ba51799",
"sha256": "44c50a4808f8bb4657eceb21cf22c8fc1905b248e10de94fed85172909266262"
},
"downloads": -1,
"filename": "taskiq_faststream-0.1.8-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1d73ca135e3141b8c80351739ba51799",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 8198,
"upload_time": "2024-04-22T20:38:25",
"upload_time_iso_8601": "2024-04-22T20:38:25.171447Z",
"url": "https://files.pythonhosted.org/packages/4e/55/320ff98350683c375d142c18ef05bd802973fff2acbc817960d60e1f17f8/taskiq_faststream-0.1.8-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "a20660842ee94492550ba3e6d0b9380f52f7e73dfcc65bced8ebef84d895803f",
"md5": "f357c0917512a554782457fa8fe91311",
"sha256": "d93a57409f3cabb908ac356e26b04bfb3f289247d8b8c6e58cf87ee2117579b0"
},
"downloads": -1,
"filename": "taskiq_faststream-0.1.8.tar.gz",
"has_sig": false,
"md5_digest": "f357c0917512a554782457fa8fe91311",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 9833,
"upload_time": "2024-04-22T20:38:27",
"upload_time_iso_8601": "2024-04-22T20:38:27.702515Z",
"url": "https://files.pythonhosted.org/packages/a2/06/60842ee94492550ba3e6d0b9380f52f7e73dfcc65bced8ebef84d895803f/taskiq_faststream-0.1.8.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-04-22 20:38:27",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "taskiq-python",
"github_project": "taskiq-faststream",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "taskiq-faststream"
}
JSON
