53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
# bot/services.py
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Optional, Protocol, Tuple, Coroutine, Any, Awaitable
|
|
from django.utils import timezone
|
|
from asgiref.sync import sync_to_async
|
|
from .models import TelegramBot, TelegramChat
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Протокол (интерфейс) репозитория — удобно мокать в тестах
|
|
from typing import Coroutine
|
|
|
|
class IChatRepository(Protocol):
|
|
async def upsert_chat(
|
|
self, *, bot: TelegramBot, chat_id: int, chat_type: str,
|
|
title: str = "", username: str = "", is_member: bool = True,
|
|
touch_last_message: bool = False,
|
|
) -> Coroutine[Any, Any, Tuple[TelegramChat, bool]]: ...
|
|
|
|
# Реализация на Django ORM
|
|
class DjangoChatRepository:
|
|
@sync_to_async
|
|
def upsert_chat(
|
|
self, *, bot: TelegramBot, chat_id: int, chat_type: str,
|
|
title: str = "", username: str = "", is_member: bool = True,
|
|
touch_last_message: bool = False,
|
|
):
|
|
defaults = {
|
|
"type": chat_type,
|
|
"title": title or "",
|
|
"username": username or "",
|
|
"is_member": is_member,
|
|
"joined_at": timezone.now() if is_member else None,
|
|
"left_at": None if is_member else timezone.now(),
|
|
}
|
|
if touch_last_message:
|
|
defaults["last_message_at"] = timezone.now()
|
|
|
|
obj, created = TelegramChat.objects.update_or_create(
|
|
bot=bot, chat_id=chat_id, defaults=defaults
|
|
)
|
|
return obj, created
|
|
|
|
@dataclass
|
|
class BotServices:
|
|
"""Контейнер зависимостей для конкретного бота."""
|
|
bot: TelegramBot
|
|
chats: IChatRepository
|
|
|
|
def build_services(bot: TelegramBot) -> BotServices:
|
|
return BotServices(bot=bot, chats=DjangoChatRepository())
|