No project description provided
Project description
Celery service task
Classe permettant d'implémenter une tâche pour un worker Celery spécifique
Usage
# import
from celery_service_task.task import TaskBase
# implémentation d'une tâche simple
class Task(TaskBase):
def task(self, payload: Dict[str, Any]) -> bool:
print(self.conf) # la configuration est donnée par le worker Celery
print(payload) # le payload est déjà sous forme d'un dictionnaire
print(payload['transaction_id']) # identifiant de la transaction issue du payload
# Initialisation de la classe avec une configuration
t = Task(
conf={'token': 'tk'},
redis=redis_connection,
redis_key=redis_task_key
)
# simulation d'un payload JSON parsé
payload={'transaction_id': '123', 'hello': 'world'}
# lance la tâche si l'id de transaction n'est pas déjà enregistré
# Ici la tâche se lance
t.run_task(
payload=payload,
t_id=payload['transaction_id']
)
# Ici la tâche est considérée comme un replica
t.run_task(
payload=payload,
t_id=payload['transaction_id']
)
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 celery_service_task-0.8.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | e493402ebc3ff129caa4bc130e0e793260ea68b88b4e88bc572d6e2b43d5c383 |
|
MD5 | 0636b3083631bf0949bebdeb6337ece2 |
|
BLAKE2b-256 | ed5bae1330bc027d9e289c3e5db58bc1641f4273abb0eb7458b6c5a8a1cfdac7 |
Close
Hashes for celery_service_task-0.8.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 6e8c026af3e805d7f55d3254affbfc80ce82d8e0470411c337e9bc34821fa3a9 |
|
MD5 | a4ec3625b07c722c46d5d690b2d1f83b |
|
BLAKE2b-256 | bc023b935dc338877f30b3b0c3bc77b4b61d39abe652bc9fd2c3efb44b1f2ce2 |