socketIO и фоновые задачи

@sio.event
async def connect(sid, environ):
    logger.info(f'Client with id: {sid} was connected')
    if not await check_token(environ):
        logger.info(f'Client with id: {sid} have not correct token')
        raise ConnectionRefusedError('Token invalid')
 
    user: dict = await sio.get_session(sid)
    if not user:
        token: str = environ.get('HTTP_TOKEN')
        user = await get_user_info(token)
        patient_id: str = await get_patient_id(user)
        if len(patient_id) > 0:
            await sio.save_session(sid, {'user_id': user.get('id'), 'patient_id': patient_id})
    else:
        patient_id: str = user.get('patient_id')
 
    if len(patient_id) > 0 and not await is_client_connected(sid, patient_id):
        if clients_connections.get(patient_id) is None:
            clients_connections[patient_id] = [sid]
        else:
            clients_connections[patient_id].append(sid)
    await sio.start_background_task(send_notifications)
    await sio.send(f'You are connected with id: {sid}')

async def send_notifications():
    while True:
        logger.info('Search new unread notifications')
        response: ClientResponse = await requests.get(f'{HOST}/api/mm/notifications/', params={'status': 'unread'},
                                                      auth=AUTH)
        text: dict = await response.json()
        if response.status == 200 and text.get('data') is not None:
            for notification in text.get('data'):
                message: str = notification.get('payload').get('text', '')
                patient_id: str = notification.get('recipient').get('id', '')
                pk: str = notification.get('id', '')
 
                if clients_connections.get(patient_id) is not None:
                    for sid in clients_connections.get(patient_id):
                        await sio.send(message, sid)
                    await change_notify_status_to_read(pk)
                else:
                    title: str = notification.get('payload').get('title')
                    data = {
                        "target_id": patient_id,
                        "data": {},
                        "title": title,
                        "body": message,
                    }
                    response: ClientResponse = await requests.post(f'{HOST}/api/mm/push/send', json=data, auth=AUTH)
                    if response.status == 200:
                        await change_notify_status_to_read(pk)
                        logger.info('Notification push success.')
                    else:
                        logger.warning(f'Notification push failed. {str(response.reason)}')
        await sio.sleep(TIMEOUT_IN_SEC)

if __name__ == '__main__':
    web.run_app(app, host='localhost', port=5000)
    sio.start_background_task(send_notifications)

response: ClientResponse = await requests.get(f'{HOST}/api/mm/notifications/', params={'status': 'unread'}, auth=AUTH)

{'user_id': ['scoket_id']}
_redis.set(patient_id, ','.join(clients_connections[patient_id]))

import logging
 
import socketio
from celery import Celery
from redis import Redis
 
logging.basicConfig(level=logging.DEBUG)
 
logger = logging.getLogger()
 
 
_redis: Redis = Redis(password='156489', db=1)
sio = socketio.Client()
sio.connect('http://localhost:5000', headers={'token': 'task'})
 
celery: Celery = Celery('task', broker='redis://:156489@localhost/0', backend='redis://:156489@localhost/2')
celery.conf.update(
    task_serializer='json',
    accept_content=['json'],  # Ignore other content
    result_serializer='json',
    timezone='Europe/Samara',
    enable_utc=False,
)
 
@celery.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(10.0, test.s(), name='add every 10')
 
@celery.task(name='test')
def test():
    patient_ids: list = _redis.keys('*')
    for patient_id in patient_ids:
        for sid in _redis.get(patient_id.decode()).decode().split(','):
            sio.emit('notification', {sid: 'message from task'})

@sio.on('notification')
async def send_notification(sid, data):
    to, message = data
    logger.debug(sid, message)
    await sio.send(message, to=to)

import logging
from http import HTTPStatus
 
import socketio
 
from requests import Response
import requests
from celery import Celery
from redis import Redis
 
logging.basicConfig(level=logging.DEBUG)
 
logger = logging.getLogger()
 
HOST = 'http://localhost:8080'
AUTH = ('root', 'secret')
 
_redis: Redis = Redis(password='156489', db=1)
sio = socketio.Client()
 
celery: Celery = Celery('tasks', broker='redis://:156489@localhost/0', backend='redis://:156489@localhost/0')
 
@sio.event
def connect():
    print('connection established')
    sio.emit('message', {'message': 'Task is connected'})
 
 
@sio.event
def disconnect():
    print('disconnected from server')
 
 
@sio.event
def message(data):
    print('message received with ', data)
 
 
sio.connect('http://localhost:5000', headers={'token': 'task'})
 
 
@celery.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(10.0, send_notifications.s(), name='add every 10')
 
@celery.task(name='send_notifications')
def send_notifications():
    logger.info('Search new unread notifications')
    print(sio.connected)
    sio.emit('notification', {'key': 'val'})
    sio.send('from task')

print(sio.connected) == True