root/tgex-backend
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: типизация для tortoise ORM

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2025-12-11 12:01:20 +03:00
parent 0f0de396e1
commit 02729ce81b
17 changed files with 181 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/adapter/postgres.py
View File

@@ -44,13 +44,13 @@ class Postgres(DatabaseBase):
if not channel_id and not telegram_id:
raise ValueError('Either channel_id or telegram_id must be provided')
filters = {'user_id': user_id}
query = domain.TargetChannel.filter(user_id=user_id)
if channel_id:
filters['id'] = channel_id
query = query.filter(id=channel_id)
if telegram_id:
filters['telegram_id'] = telegram_id
query = query.filter(telegram_id=telegram_id)
return await domain.TargetChannel.get_or_none(**filters)
return await query.first()
async def upsert_target_channel(self, channel: domain.TargetChannel) -> domain.TargetChannel:
existing = await self.get_target_channel(user_id=channel.user_id, telegram_id=channel.telegram_id)
@@ -78,20 +78,22 @@ class Postgres(DatabaseBase):
async def get_external_channel(
self, user_id: uuid.UUID, channel_id: uuid.UUID | None = None, telegram_id: int | None = None
) -> domain.ExternalChannel | None:
filters = {'user_id': user_id}
query = domain.ExternalChannel.filter(user_id=user_id)
if channel_id:
filters['id'] = channel_id
query = query.filter(id=channel_id)
if telegram_id:
filters['telegram_id'] = telegram_id
query = query.filter(telegram_id=telegram_id)
return await domain.ExternalChannel.get_or_none(**filters)
return await query.first()
async def create_external_channel(self, channel: domain.ExternalChannel) -> domain.ExternalChannel:
await channel.save()
return channel
async def get_external_channels_for_target(self, target_channel_id: uuid.UUID) -> list[domain.ExternalChannel]:
target = await domain.TargetChannel.get_or_none(id=target_channel_id).prefetch_related('external_channels')
target = await (
domain.TargetChannel.filter(id=target_channel_id).prefetch_related('external_channels').first()
)
if not target:
return []
return await target.external_channels.all()
@@ -110,7 +112,7 @@ class Postgres(DatabaseBase):
for target_id in target_channel_ids:
target_channel = await domain.TargetChannel.get_or_none(id=target_id)
if target_channel:
await external_channel.external_channels.add(target_channel)
await external_channel.target_channels.add(target_channel)
async def remove_external_channel_from_target(
self, external_channel_id: uuid.UUID, target_channel_id: uuid.UUID