Added tasks

app/telegram/tasks/tasks.py

@@ -0,0 +1,77 @@
+import asyncio
+import logging.config
+
+from logger_helper.logger_helper import LOGGING_CONFIG
+
+logging.config.dictConfig(LOGGING_CONFIG)
+logger = logging.getLogger("tasks")
+
+user_start_tasks_merged = {}
+user_start_tasks_switch = {}
+user_stop_tasks = {}
+
+
+async def add_start_task_merged(user_id: int, task: asyncio.Task):
+    """Add task to user_start_tasks dict"""
+    if user_id in user_start_tasks_merged:
+        old_task = user_start_tasks_merged[user_id]
+        if not old_task.done():
+            old_task.cancel()
+            try:
+                await old_task
+            except asyncio.CancelledError:
+                pass
+    user_start_tasks_merged[user_id] = task
+
+
+async def add_start_task_switch(user_id: int, task: asyncio.Task):
+    """Add task to user_start_tasks dict"""
+    if user_id in user_start_tasks_switch:
+        old_task = user_start_tasks_switch[user_id]
+        if not old_task.done():
+            old_task.cancel()
+            try:
+                await old_task
+            except asyncio.CancelledError:
+                pass
+    user_start_tasks_switch[user_id] = task
+
+
+async def add_stop_task(user_id: int, task: asyncio.Task):
+    """Add task to user_stop_tasks dict"""
+    if user_id in user_stop_tasks:
+        old_task = user_stop_tasks[user_id]
+        if not old_task.done():
+            old_task.cancel()
+            try:
+                await old_task
+            except asyncio.CancelledError:
+                pass
+    user_stop_tasks[user_id] = task
+
+
+def cancel_start_task_merged(user_id: int):
+    """Cancel task from user_start_tasks dict"""
+    if user_id in user_start_tasks_merged:
+        task = user_start_tasks_merged[user_id]
+        if not task.done():
+            task.cancel()
+        del user_start_tasks_merged[user_id]
+
+
+def cancel_start_task_switch(user_id: int):
+    """Cancel task from user_start_tasks dict"""
+    if user_id in user_start_tasks_switch:
+        task = user_start_tasks_switch[user_id]
+        if not task.done():
+            task.cancel()
+        del user_start_tasks_switch[user_id]
+
+
+def cancel_stop_task(user_id: int):
+    """Cancel task from user_stop_tasks dict"""
+    if user_id in user_stop_tasks:
+        task = user_stop_tasks[user_id]
+        if not task.done():
+            task.cancel()
+        del user_stop_tasks[user_id]