FastAPI Queue
Project description
FastAPI queue
How to use
- Instance the class
Queue
withredis
instance andoptions
# queue_config.py
from fastapi_queue_task import Queue
from redis.asyncio.utils import from_url
redis = from_url(
f"redis://#REDIS_HOST:#REDIS_PORT/#REDIS_DATABASE_NAME",
encoding="utf-8",
decode_responses=True,
)
queue = Queue(redis, {'concurrency': 10, 'max_attempt': 3})
queue.run()
- The
Queue
class expose 2 methods that we can use:
# mail_service.py
from queue_config.py import queue
await queue.add_class_to_queue(
name="TASK_NAME",
data=CustomClass(),
)
await queue.add_to_queue(name="TASK_NAME", data={})
How to test in testpypi
- Increase the version in
pyproject.toml
- Run command
$ . ./build_and_test.sh
How to publish new version
- Increase the version in
pyproject.toml
- Run command
$ . ./build_and_publish.sh
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
Built Distribution
Close
Hashes for fastapi_queue_task-0.0.6-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 62c5c77b7db96188ad1b8328df017b9b45e376dbc06f5224e5b5617d89431556 |
|
MD5 | 8f4844f90c093def001ccc0773c3729b |
|
BLAKE2b-256 | 5a9875b1b7e8320e5a58f55aa0213b108be2a125cdf0ba6e53be37a1ae7aa4f4 |