lavkach3

1
import asyncio
2
from contextvars import ContextVar
3
from uuid import uuid4
4

5
from taskiq import SimpleRetryMiddleware, TaskiqMiddleware
6
from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend, PubSubBroker
7

8
from core.helpers.broker.initializator import init
9

10
class TaskSession(TaskiqMiddleware):
11
    """Middleware to add retries."""
12

13
    def __init__(self,) -> None:
14
        ...
15

16
    def pre_execute(
17
            self,
18
            message: "TaskiqMessage",
19
    ) -> "Union[TaskiqMessage, Coroutine[Any, Any, TaskiqMessage]]":
20
        return message
21

22

23

24
redis_async_result = RedisAsyncResultBackend(
25
    redis_url="rediss://default:AVNS_w6X_JVOCj6vbTjwIowO@redis-do-user-15109425-0.c.db.ondigitalocean.com:25061",
26
    result_ex_time=60,  # Сколько хранить результаты в секундах
27
    ssl_cert_reqs=None,
28
    socket_timeout=360
29
)
30

31
# Or you can use PubSubBroker if you need broadcasting
32
list_brocker = ListQueueBroker(
33
    url="rediss://default:AVNS_w6X_JVOCj6vbTjwIowO@redis-do-user-15109425-0.c.db.ondigitalocean.com:25061",
34
    ssl_cert_reqs=None,
35
    queue_name='model',
36
    socket_timeout=360
37
).with_result_backend(redis_async_result).with_middlewares(SimpleRetryMiddleware(default_retry_count=3)).with_middlewares(TaskSession())
38
init(list_brocker, 'core.env:Env')
39
#
40
# async def main():
41
#     async with broker:
42
#         await broker.send_task("core.tasks:task", args=(1, 2), kwargs={"a": 3, "b": 4})
43
#
44
# if __name__ == "__main__":
45
#     asyncio.run(main())
46

47

48