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

workspace avatar and send builded creative

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2026-01-20 22:33:31 +03:00
parent e8ff317566
commit 233afb5451
26 changed files with 490 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/adapter/s3.py
View File

@@ -16,6 +16,7 @@ class S3Config(BaseModel):
SECRET_ACCESS_KEY: str
BUCKET_NAME: str
REGION: str = 'us-east-1'
PUBLIC_BASE_URL: str | None = None
class S3(S3Storage):
@@ -103,3 +104,9 @@ class S3(S3Storage):
) as client:
await client.delete_object(Bucket=self.config.BUCKET_NAME, Key=key)
log.info(f'Deleted file from S3: {key}')
def public_url(self, key: str) -> str:
base_url = self.config.PUBLIC_BASE_URL
if not base_url:
base_url = f'{self.config.ENDPOINT_URL.rstrip("/")}/{self.config.BUCKET_NAME}'
return f'{base_url.rstrip("/")}/{key.lstrip("/")}'

67
src/adapter/telegram_bot.py
View File

@@ -9,18 +9,77 @@ log = logging.getLogger(__name__)
class TelegramBot(TelegramBase, TelegramBotWriter):
async def send_message(self, text: str, chat_id: int) -> None:
await self.bot.send_message(chat_id=chat_id, text=text)
async def send_message(
self, text: str, chat_id: int, parse_mode: str | None = None, disable_preview: bool = False
) -> None:
await self.bot.send_message(
chat_id=chat_id, text=text, parse_mode=parse_mode, disable_web_page_preview=disable_preview
)
async def create_chat_invite_link(self, chat_id: int, requires_approval: bool = False) -> str:
invite_link = await self.bot.create_chat_invite_link(chat_id=chat_id, creates_join_request=requires_approval)
return invite_link.invite_link
async def send_message_with_inline_keyboard(
self, text: str, chat_id: int, buttons: list[list[InlineKeyboardButton]]
self,
text: str,
chat_id: int,
buttons: list[list[InlineKeyboardButton]],
parse_mode: str | None = None,
disable_preview: bool = False,
) -> None:
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons)
await self.bot.send_message(chat_id=chat_id, text=text, reply_markup=keyboard)
await self.bot.send_message(
chat_id=chat_id,
text=text,
reply_markup=keyboard,
parse_mode=parse_mode,
disable_web_page_preview=disable_preview,
)
async def send_media_with_inline_keyboard(
self,
text: str,
chat_id: int,
media_type: str,
media_file_id: str,
buttons: list[list[InlineKeyboardButton]],
parse_mode: str | None = None,
) -> None:
keyboard = InlineKeyboardMarkup(inline_keyboard=buttons) if buttons else None
if media_type == 'photo':
await self.bot.send_photo(
chat_id=chat_id,
photo=media_file_id,
caption=text,
parse_mode=parse_mode,
reply_markup=keyboard,
)
return
if media_type == 'video':
await self.bot.send_video(
chat_id=chat_id,
video=media_file_id,
caption=text,
parse_mode=parse_mode,
reply_markup=keyboard,
)
return
if media_type == 'animation':
await self.bot.send_animation(
chat_id=chat_id,
animation=media_file_id,
caption=text,
parse_mode=parse_mode,
reply_markup=keyboard,
)
return
await self.bot.send_message(
chat_id=chat_id,
text=text,
parse_mode=parse_mode,
reply_markup=keyboard,
)
async def edit_message_text(self, text: str, chat_id: int, message_id: int) -> None:
await self.bot.edit_message_text(chat_id=chat_id, message_id=message_id, text=text, reply_markup=None)

26
src/controller/http_v1/internal.py
View File

@@ -1,4 +1,4 @@
from typing import Annotated, Literal, Union
from typing import Annotated, Literal
from fastapi.routing import APIRouter
from pydantic import BaseModel, Field
@@ -86,13 +86,11 @@ class BotPermissionsEventRequest(dto.UpdateProjectPermissionsInput):
EventRequest = Annotated[
Union[
SubscriptionEventRequest,
UnsubscriptionEventRequest,
BotAddedEventRequest,
BotRemovedEventRequest,
BotPermissionsEventRequest,
],
SubscriptionEventRequest
| UnsubscriptionEventRequest
| BotAddedEventRequest
| BotRemovedEventRequest
| BotPermissionsEventRequest,
Field(discriminator='type'),
]
@@ -117,15 +115,15 @@ async def handle_event(input: EventRequest) -> None:
return
if isinstance(input, BotAddedEventRequest):
data = dto.ConnectProjectInput(**input.dict(exclude={'type'}))
await deps.get_usecase().tg_add_project(input=data)
add_data = dto.ConnectProjectInput(**input.dict(exclude={'type'}))
await deps.get_usecase().tg_add_project(input=add_data)
return
if isinstance(input, BotRemovedEventRequest):
data = dto.DisconnectProjectByTgIdInput(**input.dict(exclude={'type'}))
await deps.get_usecase().disconnect_project_by_tg_id(input=data)
remove_data = dto.DisconnectProjectByTgIdInput(**input.dict(exclude={'type'}))
await deps.get_usecase().disconnect_project_by_tg_id(input=remove_data)
return
if isinstance(input, BotPermissionsEventRequest):
data = dto.UpdateProjectPermissionsInput(**input.dict(exclude={'type'}))
await deps.get_usecase().update_project_permissions(input=data)
permissions_data = dto.UpdateProjectPermissionsInput(**input.dict(exclude={'type'}))
await deps.get_usecase().update_project_permissions(input=permissions_data)

25
src/controller/http_v1/workspaces.py
View File

@@ -1,7 +1,7 @@
import uuid
from typing import Annotated
from fastapi import Depends
from fastapi import Depends, File, UploadFile
from fastapi.routing import APIRouter
from fastapi_pagination import Page, paginate
@@ -42,3 +42,26 @@ async def delete_workspace(
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> None:
await deps.get_usecase().delete_workspace(workspace_id, current_user.user_id)
@workspaces_router.post('/{workspace_id}/avatar')
async def upload_workspace_avatar(
workspace_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
file: UploadFile = File(...),
) -> dto.WorkspaceMembershipOutput:
avatar_data = await file.read()
return await deps.get_usecase().update_workspace_avatar(
workspace_id,
current_user.user_id,
avatar_data,
file.content_type,
)
@workspaces_router.delete('/{workspace_id}/avatar')
async def delete_workspace_avatar(
workspace_id: uuid.UUID,
current_user: Annotated[JWTPayload, Depends(deps.get_current_user)],
) -> dto.WorkspaceMembershipOutput:
return await deps.get_usecase().delete_workspace_avatar(workspace_id, current_user.user_id)

8
src/domain/__init__.py
View File

@@ -24,6 +24,8 @@ __all__ = (
'replace_invite_link_with_tag',
'validate_media_size',
'MAX_CREATIVE_MEDIA_BYTES',
'validate_workspace_avatar_size',
'MAX_WORKSPACE_AVATAR_BYTES',
'Post',
'Subscription',
'PostViewsHistory',
@@ -41,6 +43,7 @@ __all__ = (
'WorkspaceInviteAlreadyExists',
'WorkspaceInviteAlreadyProcessed',
'WorkspaceMemberAlreadyExists',
'WorkspaceAvatarTooLarge',
'LoginTokenNotFound',
'LoginTokenExpired',
'LoginTokenAlreadyUsed',
@@ -48,6 +51,7 @@ __all__ = (
'ProjectChannelConflict',
'PlacementNotFound',
'PlacementPostNotFound',
'PlacementHasPosts',
'ChannelNotFound',
'ChannelAlreadyExists',
'ChannelNoAdminRights',
@@ -80,6 +84,7 @@ from .error import (
LoginTokenAlreadyUsed,
LoginTokenExpired,
LoginTokenNotFound,
PlacementHasPosts,
PlacementNotFound,
PlacementPostNotFound,
ProjectChannelConflict,
@@ -88,6 +93,7 @@ from .error import (
UserByUsernameNotFound,
UserNotFound,
WorkspaceAccessDenied,
WorkspaceAvatarTooLarge,
WorkspaceInviteAlreadyExists,
WorkspaceInviteAlreadyProcessed,
WorkspaceInviteNotFound,
@@ -110,6 +116,7 @@ from .subscription import Subscription, SubscriptionStatus
from .telegram_user import TelegramUser
from .user import User
from .workspace import (
MAX_WORKSPACE_AVATAR_BYTES,
PermissionKey,
PermissionScopeType,
Workspace,
@@ -119,6 +126,7 @@ from .workspace import (
WorkspaceUserPermission,
WorkspaceUserPermissionScope,
WorkspaceUserStatus,
validate_workspace_avatar_size,
)
from .workspace_permissions import (
WorkspacePermissionContext,

8
src/domain/creative.py
View File

@@ -40,7 +40,10 @@ class Creative(TimestampedModel):
MAX_CREATIVE_MEDIA_BYTES = 20 * 1024 * 1024
_INVITE_LINK_HTML = re.compile(r'<a\s+href=["\']https?://t\.me/(?:\+|joinchat/)[a-zA-Z0-9_-]+["\']>([^<]*)</a>', re.IGNORECASE)
_INVITE_LINK_HTML = re.compile(
r'<a\s+href=["\']https?://t\.me/(?:\+|joinchat/)[a-zA-Z0-9_-]+["\']>([^<]*)</a>',
re.IGNORECASE,
)
_INVITE_LINK_PLAIN = re.compile(r'https?://t\.me/(?:\+|joinchat/)[a-zA-Z0-9_-]+', re.IGNORECASE)
_INVITE_LINK_REPLACED = re.compile(r'<a\s+class=["\']tg-link["\']>([^<]*)</a>', re.IGNORECASE)
_INVITE_LINK_REPLACED_URL = re.compile(
@@ -72,7 +75,8 @@ def replace_invite_link_with_tag(text: str) -> str:
return _INVITE_LINK_REPLACED_URL.sub('<a class="tg-link"></a>', text)
if html_count == 1:
def _replace_html(match: re.Match) -> str:
def _replace_html(match: re.Match[str]) -> str:
inner = match.group(1).strip()
if _INVITE_LINK_PLAIN.fullmatch(inner):
return '<a class="tg-link"></a>'

4
src/domain/error.py
View File

@@ -129,3 +129,7 @@ def WorkspaceInviteAlreadyProcessed() -> HTTPException:
def WorkspaceMemberAlreadyExists() -> HTTPException:
return HTTPException(status.HTTP_409_CONFLICT, 'User is already a workspace member')
def WorkspaceAvatarTooLarge(max_bytes: int) -> HTTPException:
return HTTPException(status.HTTP_400_BAD_REQUEST, f'Workspace avatar is too large (max {max_bytes} bytes)')

2
src/domain/placement.py
View File

@@ -59,7 +59,7 @@ class Placement(TimestampedModel):
project: fields.ForeignKeyRelation['Project'] = fields.ForeignKeyField(
'models.Project', related_name='placements', on_delete=fields.CASCADE, index=True
)
creative: fields.ForeignKeyRelation['Creative'] = fields.ForeignKeyField(
creative: fields.ForeignKeyRelation['Creative'] | None = fields.ForeignKeyField(
'models.Creative', related_name='placements', on_delete=fields.CASCADE, null=True, index=True
)
channel: fields.ForeignKeyRelation['Channel'] = fields.ForeignKeyField(

12
src/domain/workspace.py
View File

@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING
from tortoise import fields
from .base import TimestampedModel
from .error import WorkspaceAvatarTooLarge
if TYPE_CHECKING:
from .channel import Channel
@@ -19,6 +20,7 @@ if TYPE_CHECKING:
class Workspace(TimestampedModel):
id = fields.UUIDField(pk=True)
name = fields.CharField(max_length=255)
avatar_s3_key = fields.CharField(max_length=512, null=True)
class Meta:
table = 'workspace'
@@ -149,3 +151,13 @@ class WorkspaceUserPermissionScope(TimestampedModel):
class Meta:
table = 'workspace_user_permission_scope'
MAX_WORKSPACE_AVATAR_BYTES = 5 * 1024 * 1024
def validate_workspace_avatar_size(avatar_data: bytes | None) -> None:
if avatar_data is None:
return
if len(avatar_data) > MAX_WORKSPACE_AVATAR_BYTES:
raise WorkspaceAvatarTooLarge(MAX_WORKSPACE_AVATAR_BYTES)

2
src/dto/__init__.py
View File

@@ -158,8 +158,8 @@ from .purchase import (
PlacementDetails,
PlacementOutput,
PlacementPostOutput,
PostOutput,
PlacementWithPostsOutput,
PostOutput,
UpdatePlacementInput,
)
from .user import UserOutput

3
src/dto/purchase.py
View File

@@ -6,7 +6,6 @@ import pydantic
from src.domain.placement import CostType, InviteLinkType, PlacementStatus, PlacementType
from .channel import ChannelOutput
from .creative import CreativePreviewOutput
class CostInfo(pydantic.BaseModel):
@@ -57,8 +56,6 @@ class PlacementWithPostsOutput(PlacementOutput):
placement_post: PlacementPostOutput | None = None
class CreatePlacementChannelInput(pydantic.BaseModel):
"""Input для создания одного placement (channel + детали)"""

1
src/dto/workspace.py
View File

@@ -8,6 +8,7 @@ from src import domain
class WorkspaceMembershipOutput(pydantic.BaseModel):
id: uuid.UUID
name: str
avatar_url: str | None = None
class CreateWorkspaceOutput(WorkspaceMembershipOutput): ...

27
src/usecase/__init__.py
View File

@@ -51,19 +51,38 @@ from .workspace.accept_workspace_invite import accept_workspace_invite
from .workspace.create_workspace import create_workspace
from .workspace.create_workspace_invite import create_workspace_invite
from .workspace.delete_workspace import delete_workspace
from .workspace.delete_workspace_avatar import delete_workspace_avatar
from .workspace.get_workspace_invites import get_workspace_invites
from .workspace.get_workspace_members import get_workspace_members
from .workspace.get_workspaces import get_workspaces
from .workspace.tg_accept_workspace_invite import tg_accept_workspace_invite
from .workspace.update_workspace import update_workspace
from .workspace.update_workspace_avatar import update_workspace_avatar
from .workspace.update_workspace_member_permissions import update_workspace_member_permissions
class TelegramBotWriter(typing.Protocol):
async def send_message(self, text: str, chat_id: int) -> None: ...
async def send_message(
self, text: str, chat_id: int, parse_mode: str | None = None, disable_preview: bool = False
) -> None: ...
async def send_message_with_inline_keyboard(
self, text: str, chat_id: int, buttons: list[list['InlineKeyboardButton']]
self,
text: str,
chat_id: int,
buttons: list[list['InlineKeyboardButton']],
parse_mode: str | None = None,
disable_preview: bool = False,
) -> None: ...
async def send_media_with_inline_keyboard(
self,
text: str,
chat_id: int,
media_type: str,
media_file_id: str,
buttons: list[list['InlineKeyboardButton']],
parse_mode: str | None = None,
) -> None: ...
async def edit_message_text(self, text: str, chat_id: int, message_id: int) -> None: ...
@@ -94,6 +113,8 @@ class S3Storage(typing.Protocol):
async def delete(self, key: str) -> None: ...
def public_url(self, key: str) -> str: ...
@dataclass
class Usecase:
@@ -194,6 +215,8 @@ class Usecase:
create_workspace = create_workspace
update_workspace = update_workspace
delete_workspace = delete_workspace
update_workspace_avatar = update_workspace_avatar
delete_workspace_avatar = delete_workspace_avatar
get_workspace_members = get_workspace_members
update_workspace_member_permissions = update_workspace_member_permissions
create_workspace_invite = create_workspace_invite

2
src/usecase/auth/validate_login_token.py
View File

@@ -39,7 +39,7 @@ async def validate_login_token(self: 'Usecase', input: dto.ValidateLoginTokenInp
message_id=login_token.message_id,
)
except Exception:
logging.getLogger(__name__).exception("Failed to update login message")
logging.getLogger(__name__).exception('Failed to update login message')
access_token = self.jwt_encoder.encode_access_token(
user_id=user.id,

53
src/usecase/purchase/build_placement_creative.py
View File

@@ -3,6 +3,8 @@ import re
import uuid
from typing import TYPE_CHECKING
from aiogram.types import InlineKeyboardButton
from src import domain, dto
if TYPE_CHECKING:
@@ -18,7 +20,7 @@ def _inject_invite_link(text: str, invite_link: str) -> str:
if not text:
return text
def _replace(match: re.Match) -> str:
def _replace(match: re.Match[str]) -> str:
inner = match.group(1).strip()
if inner:
return f'<a href="{invite_link}">{inner}</a>'
@@ -27,7 +29,7 @@ def _inject_invite_link(text: str, invite_link: str) -> str:
return _INVITE_LINK_TAG.sub(_replace, text)
def _build_buttons(buttons: list[dict], invite_link: str) -> list[dto.CreativeButton]:
def _build_buttons(buttons: list[dict[str, str]], invite_link: str) -> list[dto.CreativeButton]:
result: list[dto.CreativeButton] = []
for raw in buttons:
text = raw.get('text')
@@ -40,6 +42,18 @@ def _build_buttons(buttons: list[dict], invite_link: str) -> list[dto.CreativeBu
return result
def _format_channel_name(channel: domain.Channel) -> str:
if channel.title:
return channel.title
if channel.username:
return f'@{channel.username}'
return 'Без названия'
def _build_keyboard_buttons(buttons: list[dto.CreativeButton]) -> list[list[InlineKeyboardButton]]:
return [[InlineKeyboardButton(text=btn.text, url=btn.url)] for btn in buttons]
async def build_placement_creative(
self: 'Usecase',
placement_id: uuid.UUID,
@@ -47,9 +61,7 @@ async def build_placement_creative(
project_id: uuid.UUID,
user_id: uuid.UUID,
) -> dto.CreativePreviewOutput:
context = await self.ensure_workspace_permission(
workspace_id, user_id, domain.PermissionKey.PLACEMENTS_WRITE
)
context = await self.ensure_workspace_permission(workspace_id, user_id, domain.PermissionKey.PLACEMENTS_WRITE)
project = await self.database.get_project(workspace_id, project_id=project_id)
if not project:
@@ -89,4 +101,35 @@ async def build_placement_creative(
media_file_id=creative.media_file_id,
buttons=_build_buttons(creative.buttons, invite_link),
)
user = await self.database.get_user(user_id=user_id)
if not user or not user.telegram_user:
raise domain.UserNotFound(user_id)
chat_id = user.telegram_user.telegram_id
channel_name = _format_channel_name(placement.channel)
header = f'<b>Канал:</b> {channel_name}'
await self.telegram_bot.send_message(header, chat_id, parse_mode='HTML')
keyboard_buttons = _build_keyboard_buttons(preview.buttons)
if preview.media_type and preview.media_file_id:
await self.telegram_bot.send_media_with_inline_keyboard(
text=preview.text,
chat_id=chat_id,
media_type=preview.media_type,
media_file_id=preview.media_file_id,
buttons=keyboard_buttons,
parse_mode='HTML',
)
elif keyboard_buttons:
await self.telegram_bot.send_message_with_inline_keyboard(
preview.text,
chat_id,
keyboard_buttons,
parse_mode='HTML',
disable_preview=True,
)
else:
await self.telegram_bot.send_message(preview.text, chat_id, parse_mode='HTML', disable_preview=True)
return preview

6
src/usecase/purchase/create_placements.py
View File

@@ -97,8 +97,10 @@ async def create_placements(
# Объединяем общие детали с деталями конкретного канала
channel_details = channel_input.details
creative_id = (
channel_details.creative_id if channel_details and channel_details.creative_id else None
) or (details.creative_id if details and details.creative_id else None) or input.creative_id
(channel_details.creative_id if channel_details and channel_details.creative_id else None)
or (details.creative_id if details and details.creative_id else None)
or input.creative_id
)
if not creative_id:
raise domain.CreativeNotFound()
creative = creatives_by_id.get(creative_id)

4
src/usecase/purchase/get_placements.py
View File

@@ -61,7 +61,9 @@ async def get_placements(self: 'Usecase', input: dto.GetPlacementsInput) -> dto.
if placement_posts_for_placement:
if len(placement_posts_for_placement) > 1:
log.warning(
'Placement %s has %s placement_posts, returning latest', placement.id, len(placement_posts_for_placement)
'Placement %s has %s placement_posts, returning latest',
placement.id,
len(placement_posts_for_placement),
)
placement_post_output = placement_posts_for_placement[0]
placement_outputs.append(

4
src/usecase/purchase/update_placement.py
View File

@@ -18,9 +18,7 @@ async def update_placement(
project_id: uuid.UUID,
user_id: uuid.UUID,
) -> dto.PlacementWithPostsOutput:
context = await self.ensure_workspace_permission(
workspace_id, user_id, domain.PermissionKey.PLACEMENTS_WRITE
)
context = await self.ensure_workspace_permission(workspace_id, user_id, domain.PermissionKey.PLACEMENTS_WRITE)
project = await self.database.get_project(workspace_id, project_id=project_id)
if not project:

1
src/usecase/workspace/create_workspace.py
View File

@@ -30,4 +30,5 @@ async def create_workspace(
return dto.CreateWorkspaceOutput(
id=workspace.id,
name=workspace.name,
avatar_url=self.s3.public_url(workspace.avatar_s3_key) if workspace.avatar_s3_key else None,
)

42
src/usecase/workspace/delete_workspace_avatar.py Normal file
View File

@@ -0,0 +1,42 @@
import asyncio
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 delete_workspace_avatar(
self: 'Usecase', workspace_id: uuid.UUID, user_id: uuid.UUID
) -> dto.WorkspaceMembershipOutput:
await self.ensure_workspace_permission(workspace_id, user_id, domain.PermissionKey.ADMIN_FULL)
membership = await self.database.get_workspace_membership(workspace_id, user_id)
if not membership or not membership.workspace:
raise domain.WorkspaceNotFound(workspace_id)
workspace = membership.workspace
old_key = workspace.avatar_s3_key
if old_key:
workspace.avatar_s3_key = None
await self.database.update_workspace(workspace)
async def delete_old_avatar() -> None:
try:
await self.s3.delete(old_key)
log.info('Deleted workspace avatar from S3: %s', old_key)
except Exception as exc:
log.warning('Failed to delete workspace avatar from S3: %s', exc)
asyncio.create_task(delete_old_avatar())
return dto.WorkspaceMembershipOutput(
id=workspace.id,
name=workspace.name,
avatar_url=None,
)

7
src/usecase/workspace/get_workspaces.py
View File

@@ -14,6 +14,13 @@ async def get_workspaces(self: 'Usecase', user_id: uuid.UUID) -> list[dto.Worksp
dto.WorkspaceMembershipOutput(
id=membership.workspace_id,
name=membership.workspace.name,
avatar_url=_build_avatar_url(self, membership.workspace.avatar_s3_key),
)
for membership in memberships
]
def _build_avatar_url(self: 'Usecase', avatar_key: str | None) -> str | None:
if not avatar_key:
return None
return self.s3.public_url(avatar_key)

1
src/usecase/workspace/update_workspace.py
View File

@@ -25,4 +25,5 @@ async def update_workspace(
return dto.WorkspaceMembershipOutput(
id=workspace.id,
name=workspace.name,
avatar_url=self.s3.public_url(workspace.avatar_s3_key) if workspace.avatar_s3_key else None,
)

53
src/usecase/workspace/update_workspace_avatar.py Normal file
View File

@@ -0,0 +1,53 @@
import asyncio
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_workspace_avatar(
self: 'Usecase',
workspace_id: uuid.UUID,
user_id: uuid.UUID,
avatar_data: bytes,
content_type: str | None,
) -> dto.WorkspaceMembershipOutput:
await self.ensure_workspace_permission(workspace_id, user_id, domain.PermissionKey.ADMIN_FULL)
membership = await self.database.get_workspace_membership(workspace_id, user_id)
if not membership or not membership.workspace:
raise domain.WorkspaceNotFound(workspace_id)
domain.validate_workspace_avatar_size(avatar_data)
file_id = uuid.uuid4()
avatar_key = f'workspaces/{workspace_id}/avatars/{file_id}'
await self.s3.upload(avatar_key, avatar_data, content_type or 'application/octet-stream')
workspace = membership.workspace
old_key = workspace.avatar_s3_key
workspace.avatar_s3_key = avatar_key
await self.database.update_workspace(workspace)
if old_key:
async def delete_old_avatar() -> None:
try:
await self.s3.delete(old_key)
log.info('Deleted old workspace avatar from S3: %s', old_key)
except Exception as exc:
log.warning('Failed to delete old workspace avatar from S3: %s', exc)
asyncio.create_task(delete_old_avatar())
return dto.WorkspaceMembershipOutput(
id=workspace.id,
name=workspace.name,
avatar_url=self.s3.public_url(avatar_key),
)