RSHB/DFiP_Budget_planing
5
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

websocket-self-flag: добавил флаг is_self в сообщениях вебсокетов

Browse Source
This commit is contained in:
tsygankoviva 2026-06-22 16:48:51 +03:00
parent 361eeb24bb
commit a2841d8c83
1 changed files with 44 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
api/src/api/v1/websocket.py
View File

@ -1,3 +1,4 @@
import asyncio
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from dataclasses import asdict from dataclasses import asdict
import dataclasses import dataclasses
@ -152,7 +153,7 @@ class ConnectionManager:
async def broadcast_to_other( async def broadcast_to_other(
self, self,
message: str, data: dict,
user_id: int, user_id: int,
con_key: ConnectionKeyEnum, con_key: ConnectionKeyEnum,
**kwargs, **kwargs,
@ -162,6 +163,10 @@ class ConnectionManager:
if key not in self.connections[con_key]: if key not in self.connections[con_key]:
return return
data = data.copy()
data["is_self"] = False
message = dumps(data, default=_convert_error, ensure_ascii=False)
for con_info in self.connections[con_key][key]: for con_info in self.connections[con_key][key]:
if con_info.user_id != user_id: if con_info.user_id != user_id:
try: try:
@ -171,20 +176,41 @@ class ConnectionManager:
async def broadcast_to_all( async def broadcast_to_all(
self, self,
message: str, data: dict,
con_key: ConnectionKeyEnum, con_key: ConnectionKeyEnum,
user_id: int | None = None,
**kwargs, **kwargs,
): ):
key = frozenset(kwargs.items()) key = frozenset(kwargs.items())
if key not in self.connections[con_key]: if key not in self.connections[con_key]:
return return
data = data.copy()
if user_id is None:
data["is_self"] = None
default_message = dumps(data, default=_convert_error, ensure_ascii=False)
self_message = None
else:
data["is_self"] = False
default_message = dumps(data, default=_convert_error, ensure_ascii=False)
data["is_self"] = True
self_message = dumps(data, default=_convert_error, ensure_ascii=False)
for con_info in self.connections[con_key][key]: for con_info in self.connections[con_key][key]:
try: try:
await con_info.ws.send_text(message) if user_id is not None and con_info.user_id == user_id:
await con_info.ws.send_text(self_message)
else:
await con_info.ws.send_text(default_message)
except: except:
pass # Игнорируем недоступные соединения pass # Игнорируем недоступные соединения
async def send_back(self, message: str, websocket: WebSocket): async def send_back(self, data: dict, websocket: WebSocket):
data = data.copy()
data["is_self"] = True
message = dumps(data, default=_convert_error, ensure_ascii=False)
await websocket.send_text(message) await websocket.send_text(message)
async def get_data(self, websocket: WebSocket) -> dict: async def get_data(self, websocket: WebSocket) -> dict:
@ -665,16 +691,16 @@ async def process_websocket(
): ):
data["error"] = "Ячейка уже редактируется другим пользователем" data["error"] = "Ячейка уже редактируется другим пользователем"
await manager.send_back( await manager.send_back(
dumps(data, default=_convert_error, ensure_ascii=False), data,
websocket, websocket,
) )
continue continue
await manager.send_back( await manager.send_back(
dumps(data, default=_convert_error, ensure_ascii=False), data,
websocket, websocket,
) )
await manager.broadcast_to_other( await manager.broadcast_to_other(
message=dumps(data, default=_convert_error, ensure_ascii=False), data=data,
user_id=user_id, user_id=user_id,
con_key=con_key, con_key=con_key,
**kwargs_process, **kwargs_process,
@ -688,11 +714,11 @@ async def process_websocket(
form_key=kwargs_process, form_key=kwargs_process,
) )
await manager.send_back( await manager.send_back(
dumps(data, default=_convert_error, ensure_ascii=False), data,
websocket, websocket,
) )
await manager.broadcast_to_other( await manager.broadcast_to_other(
message=dumps(data, default=_convert_error, ensure_ascii=False), data=data,
user_id=user_id, user_id=user_id,
con_key=con_key, con_key=con_key,
**kwargs_process, **kwargs_process,
@ -707,7 +733,7 @@ async def process_websocket(
): ):
data["error"] = "Ячейка уже редактируется другим пользователем" data["error"] = "Ячейка уже редактируется другим пользователем"
await manager.send_back( await manager.send_back(
dumps(data, default=_convert_error, ensure_ascii=False), data,
websocket, websocket,
) )
continue continue
@ -719,7 +745,7 @@ async def process_websocket(
): ):
data["error"] = "Ячейка уже редактируется другим пользователем" data["error"] = "Ячейка уже редактируется другим пользователем"
await manager.send_back( await manager.send_back(
dumps(data, default=_convert_error, ensure_ascii=False), data,
websocket, websocket,
) )
continue continue
@ -733,7 +759,7 @@ async def process_websocket(
"Нельзя удалить строку: в этой строке есть редактируемые ячейки" "Нельзя удалить строку: в этой строке есть редактируемые ячейки"
) )
await manager.send_back( await manager.send_back(
dumps(data, default=_convert_error, ensure_ascii=False), data,
websocket, websocket,
) )
continue continue
@ -762,12 +788,13 @@ async def process_websocket(
"row_deleted", "row_deleted",
]: ]:
await manager.broadcast_to_all( await manager.broadcast_to_all(
message=dumps(data, default=_convert_error), data=data,
user_id=user_id,
**kwargs **kwargs
) )
else: else:
await manager.broadcast_to_other( await manager.broadcast_to_other(
message=dumps(data, default=_convert_error), data=data,
user_id=user_id, user_id=user_id,
**kwargs, **kwargs,
) )
@ -775,13 +802,13 @@ async def process_websocket(
except BasicAppException as e: except BasicAppException as e:
data["error"] = e.description or "Неизвестная ошибка" data["error"] = e.description or "Неизвестная ошибка"
await manager.send_back( await manager.send_back(
message=dumps(data, default=_convert_error), data=data,
websocket=websocket, websocket=websocket,
) )
except IntegrityError as e: except IntegrityError as e:
data["error"] = str(e) data["error"] = str(e)
await manager.send_back( await manager.send_back(
message=dumps(data, default=_convert_error), data=data,
websocket=websocket, websocket=websocket,
) )
@ -791,7 +818,7 @@ async def process_websocket(
if tp in handlers: if tp in handlers:
data["error"] = loads((await handlers[tp](request=None, exc=e)).body) data["error"] = loads((await handlers[tp](request=None, exc=e)).body)
await manager.send_back( await manager.send_back(
message=dumps(data, default=_convert_error), data=data,
websocket=websocket, websocket=websocket,
) )
else: else: