feat: creatives and external channels

src/usecase/__init__.py

@@ -0,0 +1,123 @@
+import datetime
+import typing
+from dataclasses import dataclass
+from uuid import UUID
+
+from src import domain
+
+from .creative.archive_creative import archive_creative
+from .creative.create_creative import create_creative
+from .creative.delete_creative import delete_creative
+from .creative.get_creative import get_creative
+from .creative.get_creatives import get_creatives
+from .creative.update_creative import update_creative
+from .external_channel.create_external_channel import create_external_channel
+from .external_channel.delete_external_channel import delete_external_channel
+from .external_channel.get_external_channels import get_external_channels
+from .external_channel.import_external_channels_from_excel import import_external_channels_from_excel
+from .external_channel.update_external_channel_links import update_external_channel_links
+from .target_channel.connect_target_chan import connect_target_chan
+from .target_channel.disconnect_target_chan import disconnect_target_chan
+from .target_channel.disconnect_target_chan_by_tg_id import disconnect_target_chan_by_tg_id
+from .target_channel.get_user_target_chans import get_user_target_chans
+from .target_channel.update_target_chan_permissions import update_target_chan_permissions
+from .telegram_login import telegram_login
+from .validate_login_token import validate_login_token
+
+
+class Database(typing.Protocol):
+    def transaction(self) -> typing.AsyncContextManager[None]: ...
+
+    async def get_user(self, *, user_id: UUID | None = None, telegram_id: int | None = None) -> domain.User | None: ...
+
+    async def create_user(self, user: domain.User) -> domain.User: ...
+
+    async def create_login_token(
+        self, user_id: UUID, token: str, expires_at: datetime.datetime
+    ) -> domain.LoginToken: ...
+
+    async def get_login_token(self, token: str) -> domain.LoginToken | None: ...
+
+    async def mark_token_as_used(self, token: str) -> None: ...
+
+    async def get_target_channel(
+        self, user_id: UUID, channel_id: UUID | None = None, telegram_id: int | None = None
+    ) -> domain.TargetChannel | None: ...
+
+    async def create_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel: ...
+
+    async def upsert_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel: ...
+
+    async def get_user_target_channels(self, user_id: UUID) -> list[domain.TargetChannel]: ...
+
+    async def update_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel: ...
+
+    async def get_external_channel(
+        self, user_id: UUID, channel_id: UUID | None = None, telegram_id: int | None = None
+    ) -> domain.ExternalChannel | None: ...
+
+    async def create_external_channel(self, channel: domain.ExternalChannel) -> domain.ExternalChannel: ...
+
+    async def get_external_channels_for_target(self, target_channel_id: UUID) -> list[domain.ExternalChannel]: ...
+
+    async def add_external_channel_to_targets(
+        self, external_channel_id: UUID, target_channel_ids: list[UUID]
+    ) -> None: ...
+
+    async def remove_external_channel_from_target(self, external_channel_id: UUID, target_channel_id: UUID) -> None: ...
+
+    async def delete_external_channel(self, channel_id: UUID) -> None: ...
+
+    async def get_creative(self, user_id: UUID, creative_id: UUID) -> domain.Creative | None: ...
+
+    async def create_creative(self, creative: domain.Creative) -> domain.Creative: ...
+
+    async def update_creative(self, creative: domain.Creative) -> domain.Creative: ...
+
+    async def get_user_creatives(
+        self,
+        user_id: UUID,
+        *,
+        target_channel_id: UUID | None = None,
+        include_archived: bool = False,
+    ) -> list[domain.Creative]: ...
+
+    async def delete_creative(self, creative_id: UUID) -> None: ...
+
+
+class TelegramWriter(typing.Protocol):
+    async def send_message(self, text: str, chat_id: int) -> None: ...
+
+    async def send_message_with_button(self, text: str, chat_id: int, button_text: str, button_url: str) -> None: ...
+
+    async def get_chat_member(self, chat_id: int, user_id: int) -> dict[str, typing.Any]: ...
+
+
+class JWTEncoder(typing.Protocol):
+    def encode_access_token(self, user_id: UUID, telegram_id: int, username: str | None = None) -> str: ...
+
+
+@dataclass
+class Usecase:
+    database: Database
+    telegram_writer: TelegramWriter
+    jwt_encoder: JWTEncoder
+
+    validate_login_token = validate_login_token
+    telegram_login = telegram_login
+    connect_target_chan = connect_target_chan
+    get_user_target_chans = get_user_target_chans
+    disconnect_target_chan = disconnect_target_chan
+    disconnect_target_chan_by_tg_id = disconnect_target_chan_by_tg_id
+    update_target_chan_permissions = update_target_chan_permissions
+    get_external_channels = get_external_channels
+    create_external_channel = create_external_channel
+    delete_external_channel = delete_external_channel
+    update_external_channel_links = update_external_channel_links
+    import_external_channels_from_excel = import_external_channels_from_excel
+    get_creatives = get_creatives
+    get_creative = get_creative
+    create_creative = create_creative
+    update_creative = update_creative
+    archive_creative = archive_creative
+    delete_creative = delete_creative