45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from src import domain, dto
|
|
|
|
if TYPE_CHECKING:
|
|
from .. import Usecase
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
async def get_purchase_plan_channels(
|
|
self: 'Usecase', input: dto.GetPurchasePlanChannelsInput
|
|
) -> list[dto.PurchasePlanChannelOutput]:
|
|
context = await self.ensure_workspace_permission(
|
|
input.workspace_id, input.user_id, domain.PermissionKey.PLACEMENTS_READ
|
|
)
|
|
|
|
project = await self.database.get_project(input.workspace_id, project_id=input.project_id)
|
|
if not project:
|
|
raise domain.ProjectNotFound(input.project_id)
|
|
|
|
context.ensure_project_permission(domain.PermissionKey.PLACEMENTS_READ, project.id)
|
|
|
|
plan = await self.ensure_active_purchase_plan(project)
|
|
|
|
plan_channels = await self.database.get_purchase_plan_channels(plan.id)
|
|
log.debug('Fetched %s channels for purchase plan %s (project %s)', len(plan_channels), plan.id, project.id)
|
|
|
|
return [
|
|
dto.PurchasePlanChannelOutput(
|
|
id=plan_channel.id,
|
|
status=plan_channel.status,
|
|
planned_cost=plan_channel.planned_cost,
|
|
comment=plan_channel.comment,
|
|
channel=dto.ChannelOutput(
|
|
id=plan_channel.channel.id,
|
|
telegram_id=plan_channel.channel.telegram_id,
|
|
title=plan_channel.channel.title,
|
|
username=plan_channel.channel.username,
|
|
),
|
|
)
|
|
for plan_channel in plan_channels
|
|
]
|