39 lines
1.6 KiB
Python
39 lines
1.6 KiB
Python
"""Тонкая обёртка над LiveKit `RoomService` — единственная точка для мокирования в тестах.
|
||
|
||
Используется maintenance-задачей автоосвобождения комнат (`workers/tasks/
|
||
maintenance.py`). Вынесена в отдельный модуль намеренно: реальный HTTP-вызов
|
||
к LiveKit не нужен и не воспроизводим в тестах — тесты подменяют
|
||
`delete_livekit_room` целиком.
|
||
"""
|
||
|
||
import logging
|
||
|
||
from livekit import api
|
||
|
||
from core.config import get_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def delete_livekit_room(room_name: str) -> None:
|
||
"""Удалить LiveKit-комнату по имени (= `rooms.permanent_link`), отключив всех участников.
|
||
|
||
Идемпотентно: удаление уже отсутствующей комнаты — не ошибка (LiveKit
|
||
отвечает `TwirpError`/`ServerError` с `code == "not_found"`, который
|
||
здесь поглощается; см. документацию `livekit-api` RoomService).
|
||
"""
|
||
settings = get_settings()
|
||
lkapi = api.LiveKitAPI(
|
||
settings.livekit_url,
|
||
api_key=settings.livekit_api_key,
|
||
api_secret=settings.livekit_api_secret,
|
||
)
|
||
try:
|
||
await lkapi.room.delete_room(api.DeleteRoomRequest(room=room_name))
|
||
except api.TwirpError as exc:
|
||
if exc.code != "not_found":
|
||
raise
|
||
logger.info("delete_livekit_room: комната %s уже отсутствует в LiveKit", room_name)
|
||
finally:
|
||
await lkapi.aclose()
|