48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
import logging
|
|
import uuid
|
|
from typing import TYPE_CHECKING
|
|
|
|
from src import domain, dto
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
async def update_external_channel(
|
|
self: 'Usecase',
|
|
channel_id: uuid.UUID,
|
|
input: dto.UpdateExternalChannelInput,
|
|
user_id: uuid.UUID,
|
|
workspace_id: uuid.UUID,
|
|
) -> dto.ExternalChannelOutput:
|
|
await self.ensure_workspace_access(workspace_id, user_id)
|
|
|
|
external_channel = await self.database.get_external_channel(workspace_id, channel_id=channel_id)
|
|
if not external_channel:
|
|
log.warning('External channel %s not found', channel_id)
|
|
raise domain.ExternalChannelNotFound(channel_id)
|
|
|
|
if input.title is not None:
|
|
external_channel.title = input.title
|
|
if input.username is not None:
|
|
external_channel.username = input.username
|
|
if input.description is not None:
|
|
external_channel.description = input.description
|
|
if input.subscribers_count is not None:
|
|
external_channel.subscribers_count = input.subscribers_count
|
|
|
|
await self.database.update_external_channel(external_channel)
|
|
|
|
log.info('External channel %s updated', channel_id)
|
|
|
|
return dto.ExternalChannelOutput(
|
|
id=external_channel.id,
|
|
telegram_id=external_channel.telegram_id,
|
|
title=external_channel.title,
|
|
username=external_channel.username,
|
|
description=external_channel.description,
|
|
subscribers_count=external_channel.subscribers_count,
|
|
)
|