forked from kodorvan/stcs
78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
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]
|