damanukyan/b2b
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: update

Browse Source
This commit is contained in:
Daniil Manukian
2026-06-09 12:31:45 +03:00
parent 41ef2e8e4e
commit 4ccbe2aa2c
57 changed files with 12 additions and 3165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
src/application/commands/change_email_complete.py
View File

@@ -1,63 +0,0 @@
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ILogger, ICache
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.database.decorators import transactional
class ChangeEmailCompleteCommand:
def __init__(
self,
unit_of_work: IUnitOfWork,
hash_service: IHashService,
cache: ICache,
logger: ILogger,
):
self._unit_of_work = unit_of_work
self._hash_service = hash_service
self._cache = cache
self._logger = logger
@transactional
async def __call__(self, *, user_id: str, code: str) -> bool:
code = (code or '').strip()
NEW_USER_PREFIX = 'change_email:new_user:'
NEW_CODE_PREFIX = 'change_email:new_code:'
new_user_key = f'{NEW_USER_PREFIX}{user_id}'
new_code_key = f'{NEW_CODE_PREFIX}{code}'
cached_user_id = await self._cache.get(new_code_key)
if not cached_user_id:
self._logger.info(f'Change email complete failed: code not found (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
if cached_user_id != user_id:
self._logger.info(f'Change email complete failed: code-user mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
raw_value = await self._cache.get(new_user_key)
if not raw_value:
self._logger.info(f'Change email complete failed: user key missing (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
separator_idx = raw_value.index(':')
code_hash = raw_value[:separator_idx]
new_email = raw_value[separator_idx + 1:]
ok = await self._hash_service.verify(hashed_value=code_hash, plain_value=code)
if not ok:
self._logger.info(f'Change email complete failed: code hash mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
user = await self._unit_of_work.user_repository.set_email(user_id=user_id, email=new_email)
await self._cache.set_user(user_id, user)
try:
await self._cache.delete(new_code_key)
await self._cache.delete(new_user_key)
except Exception as e:
self._logger.warning(f'Change email complete cleanup failed (user_id={user_id}): {e}')
self._logger.info(f'Email changed for user_id={user_id}')
return True

145
src/application/commands/change_email_confirm_old.py
View File

@@ -1,145 +0,0 @@
import secrets
from datetime import datetime, timezone
from ulid import ULID
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ICache, ILogger, IQueueMessanger
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config import settings
from src.infrastructure.context_vars import trace_id_var
from src.infrastructure.database.decorators import transactional
class ChangeEmailConfirmOldCommand:
def __init__(
self,
hash_service: IHashService,
cache: ICache,
unit_of_work: IUnitOfWork,
logger: ILogger,
messanger: IQueueMessanger,
):
self._hash_service = hash_service
self._unit_of_work = unit_of_work
self._cache = cache
self._logger = logger
self._messanger = messanger
@transactional
async def __call__(self, *, user_id: str, code: str, new_email: str) -> bool:
TTL = 300
MAX_ATTEMPTS = 20
OLD_USER_PREFIX = 'change_email:old_user:'
OLD_CODE_PREFIX = 'change_email:old_code:'
NEW_USER_PREFIX = 'change_email:new_user:'
NEW_CODE_PREFIX = 'change_email:new_code:'
code = (code or '').strip()
old_user_key = f'{OLD_USER_PREFIX}{user_id}'
old_code_key = f'{OLD_CODE_PREFIX}{code}'
cached_user_id = await self._cache.get(old_code_key)
if not cached_user_id:
self._logger.info(f'Change email confirm-old failed: code not found (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
if cached_user_id != user_id:
self._logger.info(f'Change email confirm-old failed: code-user mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
code_hash = await self._cache.get(old_user_key)
if not code_hash:
self._logger.info(f'Change email confirm-old failed: user key missing (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
ok = await self._hash_service.verify(hashed_value=code_hash, plain_value=code)
if not ok:
self._logger.info(f'Change email confirm-old failed: code hash mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
if user.email and user.email.lower() == new_email.lower():
self._logger.info(f'Change email confirm-old failed: new email same as current (user_id={user_id})')
raise ApplicationException(400, 'New email must differ from the current one')
email_taken = await self._unit_of_work.user_repository.email_exists(email=new_email)
if email_taken:
self._logger.info(f'Change email confirm-old failed: new email already taken (user_id={user_id})')
raise ApplicationException(409, 'Email already in use')
try:
await self._cache.delete(old_code_key)
await self._cache.delete(old_user_key)
except Exception as e:
self._logger.warning(f'Change email confirm-old cleanup failed (user_id={user_id}): {e}')
trace_id = trace_id_var.get()
if not trace_id or trace_id == 'N/A':
trace_id = None
new_user_key = f'{NEW_USER_PREFIX}{user_id}'
for _ in range(MAX_ATTEMPTS):
new_code = f'{secrets.randbelow(1_000_000):06d}'
new_code_key = f'{NEW_CODE_PREFIX}{new_code}'
new_code_hash = await self._hash_service.hash(new_code)
reserved = await self._cache.set_nx(new_code_key, user_id, ttl=TTL)
if not reserved:
continue
saved = await self._cache.set(new_user_key, f'{new_code_hash}:{new_email}', ttl=TTL)
if not saved:
await self._cache.delete(new_code_key)
self._logger.error(f'Change email confirm-old failed: cannot save new code hash for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
message_id = str(ULID())
now = datetime.now(timezone.utc).isoformat()
metadata = {
'trace_id': trace_id,
'source': 'user-service',
'timestamp': now,
'message_id': message_id,
}
payload = {
'email': new_email,
'code': new_code,
'ttl_seconds': TTL,
}
message = {
'event': 'change_email_new',
'payload': payload,
'metadata': metadata,
}
self._logger.info(f'Change email new code created for user_id={user_id}')
try:
await self._messanger.publish_to_queue(
queue=settings.RABBIT_EMAIL_CODE_QUEUE,
message=message,
persist=True,
correlation_id=trace_id,
message_id=message_id,
headers={'trace_id': trace_id} if trace_id else None,
)
except Exception as exception:
try:
await self._cache.delete(new_user_key)
await self._cache.delete(new_code_key)
except Exception as rollback_err:
self._logger.error(f'Publish failed and rollback cache failed for user_id={user_id}: {str(rollback_err)}')
self._logger.error(f'Failed to publish change email new code for user_id={user_id}: {str(exception)}')
raise ApplicationException(503, 'Temporary error. Please try again.')
return True
self._logger.error(f'Change email confirm-old failed: code space exhausted for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')

126
src/application/commands/change_email_start.py
View File

@@ -1,126 +0,0 @@
import secrets
from datetime import datetime, timezone
from ulid import ULID
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ICache, ILogger, IQueueMessanger
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config import settings
from src.infrastructure.context_vars import trace_id_var
from src.infrastructure.database.decorators import transactional
class ChangeEmailStartCommand:
def __init__(
self,
hash_service: IHashService,
cache: ICache,
unit_of_work: IUnitOfWork,
logger: ILogger,
messanger: IQueueMessanger,
):
self._hash_service = hash_service
self._unit_of_work = unit_of_work
self._cache = cache
self._logger = logger
self._messanger = messanger
@transactional
async def __call__(self, user_id: str) -> bool:
TTL = 300
LOCK_TTL = 30
MAX_ATTEMPTS = 20
USER_PREFIX = 'change_email:old_user:'
CODE_PREFIX = 'change_email:old_code:'
LOCK_PREFIX = 'change_email:lock:'
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
if not user.email:
self._logger.warning(f'User {user_id} does not have an email address')
raise ApplicationException(404, f'User {user_id} does not have an email address')
trace_id = trace_id_var.get()
if not trace_id or trace_id == 'N/A':
trace_id = None
lock_key = f'{LOCK_PREFIX}{user_id}'
locked = await self._cache.set_nx(lock_key, '1', ttl=LOCK_TTL)
if not locked:
self._logger.info(f'Change email throttled by lock (user_id={user_id})')
raise ApplicationException(429, 'Too many requests. Please wait.')
try:
user_key = f'{USER_PREFIX}{user_id}'
existing = await self._cache.get(user_key)
if existing:
self._logger.info(f'Change email denied: code already exists for user_id={user_id}')
raise ApplicationException(429, 'Code already sent. Please wait before retrying.')
for _ in range(MAX_ATTEMPTS):
code = f'{secrets.randbelow(1_000_000):06d}'
code_key = f'{CODE_PREFIX}{code}'
code_hash = await self._hash_service.hash(code)
reserved = await self._cache.set_nx(code_key, user_id, ttl=TTL)
if not reserved:
continue
saved = await self._cache.set(user_key, code_hash, ttl=TTL)
if not saved:
await self._cache.delete(code_key)
self._logger.error(f'Change email failed: cannot save code hash for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
message_id = str(ULID())
now = datetime.now(timezone.utc).isoformat()
metadata = {
'trace_id': trace_id,
'source': 'user-service',
'timestamp': now,
'message_id': message_id,
}
payload = {
'email': user.email,
'code': code,
'ttl_seconds': TTL,
}
message = {
'event': 'change_email_old',
'payload': payload,
'metadata': metadata,
}
self._logger.info(f'Change email old code created for user_id={user_id}')
try:
await self._messanger.publish_to_queue(
queue=settings.RABBIT_EMAIL_CODE_QUEUE,
message=message,
persist=True,
correlation_id=trace_id,
message_id=message_id,
headers={'trace_id': trace_id} if trace_id else None,
)
except Exception as exception:
try:
await self._cache.delete(user_key)
await self._cache.delete(code_key)
except Exception as rollback_err:
self._logger.error(f'Publish failed and rollback cache failed for user_id={user_id}: {str(rollback_err)}')
self._logger.error(f'Failed to publish change email old code for user_id={user_id}: {str(exception)}')
raise ApplicationException(503, 'Temporary error. Please try again.')
return True
self._logger.error(f'Change email failed: code space exhausted for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
finally:
await self._cache.delete(lock_key)

81
src/application/commands/change_password_complete.py
View File

@@ -1,81 +0,0 @@
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ILogger, ICache
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.database.decorators import transactional
class ChangePasswordCompleteCommand:
def __init__(
self,
unit_of_work: IUnitOfWork,
hash_service: IHashService,
cache: ICache,
logger: ILogger,
):
self._unit_of_work = unit_of_work
self._hash_service = hash_service
self._cache = cache
self._logger = logger
@transactional
async def __call__(
self,
*,
user_id: str,
code: str,
new_password: str,
confirm_password: str,
) -> bool:
code = (code or '').strip()
USER_PREFIX = 'change_password:user:'
CODE_PREFIX = 'change_password:code:'
user_key = f'{USER_PREFIX}{user_id}'
code_key = f'{CODE_PREFIX}{code}'
if new_password != confirm_password:
self._logger.info(f'Change password failed: passwords do not match (user_id={user_id})')
raise ApplicationException(400, 'Passwords do not match')
cached_user_id = await self._cache.get(code_key)
if not cached_user_id:
self._logger.info(f'Change password failed: code not found (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
if cached_user_id != user_id:
self._logger.info(f'Change password failed: code-user mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
code_hash = await self._cache.get(user_key)
if not code_hash:
self._logger.info(f'Change password failed: user key missing (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
ok = await self._hash_service.verify(hashed_value=code_hash, plain_value=code)
if not ok:
self._logger.info(f'Change password failed: code hash mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
current_password_hash = await self._unit_of_work.user_repository.get_password_hash(user_id=user_id)
is_same = await self._hash_service.verify(hashed_value=current_password_hash, plain_value=new_password)
if is_same:
self._logger.info(f'Change password failed: new password same as current (user_id={user_id})')
raise ApplicationException(400, 'New password must differ from the current one')
new_password_hash = await self._hash_service.hash(new_password)
user = await self._unit_of_work.user_repository.set_password(
user_id=user_id,
password_hash=new_password_hash,
)
await self._cache.set_user(user_id, user)
try:
await self._cache.delete(code_key)
await self._cache.delete(user_key)
except Exception as e:
self._logger.warning(f'Change password cleanup failed (user_id={user_id}): {e}')
self._logger.info(f'Password changed for user_id={user_id}')
return True

126
src/application/commands/change_password_start.py
View File

@@ -1,126 +0,0 @@
import secrets
from datetime import datetime, timezone
from ulid import ULID
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ICache, ILogger, IQueueMessanger
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config import settings
from src.infrastructure.context_vars import trace_id_var
from src.infrastructure.database.decorators import transactional
class ChangePasswordStartCommand:
def __init__(
self,
hash_service: IHashService,
cache: ICache,
unit_of_work: IUnitOfWork,
logger: ILogger,
messanger: IQueueMessanger,
):
self._hash_service = hash_service
self._unit_of_work = unit_of_work
self._cache = cache
self._logger = logger
self._messanger = messanger
@transactional
async def __call__(self, user_id: str) -> bool:
TTL = 300
LOCK_TTL = 30
MAX_ATTEMPTS = 20
USER_PREFIX = 'change_password:user:'
CODE_PREFIX = 'change_password:code:'
LOCK_PREFIX = 'change_password:lock:'
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
if not user.email:
self._logger.warning(f'User {user_id} does not have an email address')
raise ApplicationException(404, f'User {user_id} does not have an email address')
trace_id = trace_id_var.get()
if not trace_id or trace_id == 'N/A':
trace_id = None
lock_key = f'{LOCK_PREFIX}{user_id}'
locked = await self._cache.set_nx(lock_key, '1', ttl=LOCK_TTL)
if not locked:
self._logger.info(f'Change password throttled by lock (user_id={user_id})')
raise ApplicationException(429, 'Too many requests. Please wait.')
try:
user_key = f'{USER_PREFIX}{user_id}'
existing = await self._cache.get(user_key)
if existing:
self._logger.info(f'Change password denied: code already exists for user_id={user_id}')
raise ApplicationException(429, 'Code already sent. Please wait before retrying.')
for _ in range(MAX_ATTEMPTS):
code = f'{secrets.randbelow(1_000_000):06d}'
code_key = f'{CODE_PREFIX}{code}'
code_hash = await self._hash_service.hash(code)
reserved = await self._cache.set_nx(code_key, user_id, ttl=TTL)
if not reserved:
continue
saved = await self._cache.set(user_key, code_hash, ttl=TTL)
if not saved:
await self._cache.delete(code_key)
self._logger.error(f'Change password failed: cannot save code hash for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
message_id = str(ULID())
now = datetime.now(timezone.utc).isoformat()
metadata = {
'trace_id': trace_id,
'source': 'user-service',
'timestamp': now,
'message_id': message_id,
}
payload = {
'email': user.email,
'code': code,
'ttl_seconds': TTL,
}
message = {
'event': 'change_password',
'payload': payload,
'metadata': metadata,
}
self._logger.info(f'Change password code created for user_id={user_id}')
try:
await self._messanger.publish_to_queue(
queue=settings.RABBIT_EMAIL_CODE_QUEUE,
message=message,
persist=True,
correlation_id=trace_id,
message_id=message_id,
headers={'trace_id': trace_id} if trace_id else None,
)
except Exception as exception:
try:
await self._cache.delete(user_key)
await self._cache.delete(code_key)
except Exception as rollback_err:
self._logger.error(f'Publish failed and rollback cache failed for user_id={user_id}: {str(rollback_err)}')
self._logger.error(f'Failed to publish change password email for user_id={user_id}: {str(exception)}')
raise ApplicationException(503, 'Temporary error. Please try again.')
return True
self._logger.error(f'Change password failed: code space exhausted for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
finally:
await self._cache.delete(lock_key)

56
src/application/commands/delete_avatar.py
View File

@@ -1,56 +0,0 @@
from __future__ import annotations
from botocore.exceptions import ClientError
from src.application.abstractions import IUnitOfWork
from src.application.contracts import ICache, ILogger, IS3
from src.application.domain.entities import UserEntity
from src.infrastructure.database.decorators import transactional
class DeleteAvatarCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger, cache: ICache, s3: IS3):
self._unit_of_work = unit_of_work
self._logger = logger
self._cache = cache
self._s3 = s3
@transactional
async def _load_user(self, user_id: str) -> UserEntity:
user = await self._unit_of_work.user_repository.get_user_by_id(user_id)
self._logger.debug(f'DeleteAvatar _load_user user_id={user_id} has_avatar_link={bool(user.avatar_link)}')
return user
async def __call__(self, user_id: str) -> UserEntity:
prior = await self._load_user(user_id)
link = prior.avatar_link
self._logger.info(f'DeleteAvatar start user_id={user_id} had_link={bool(link)}')
if link:
key = self._s3.object_key_from_public_url(link)
self._logger.debug(f'DeleteAvatar parsed_object_key user_id={user_id} has_key={bool(key)}')
if not key:
self._logger.warning(
f'DeleteAvatar could not parse avatar URL for S3 user_id={user_id} link_len={len(link)}'
)
if key:
self._logger.info(f'DeleteAvatar S3 delete start user_id={user_id} key={key}')
try:
await self._s3.delete_object(key=key)
self._logger.info(f'DeleteAvatar S3 delete done user_id={user_id} key={key}')
except ClientError as exc:
code = exc.response.get('Error', {}).get('Code', '')
if code not in ('NoSuchKey', '404'):
self._logger.warning(f'DeleteAvatar S3 delete failed user_id={user_id} code={code}: {exc}')
else:
self._logger.debug(f'DeleteAvatar S3 object already absent user_id={user_id} code={code}')
user = await self._clear_avatar_link(user_id)
self._logger.debug(f'DeleteAvatar DB cleared user_id={user_id} entity_has_link={bool(user.avatar_link)}')
await self._cache.set_user(user_id, user)
self._logger.debug(f'DeleteAvatar cache updated user_id={user_id}')
self._logger.info(f'Avatar removed user_id={user_id}')
return user
@transactional
async def _clear_avatar_link(self, user_id: str) -> UserEntity:
self._logger.debug(f'DeleteAvatar DB transaction set_avatar_link user_id={user_id} link=None')
return await self._unit_of_work.user_repository.set_avatar_link(user_id, None)

83
src/application/commands/forgot_password_complete.py
View File

@@ -1,83 +0,0 @@
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ILogger, ICache
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.database.decorators import transactional
class ForgotPasswordCompleteCommand:
def __init__(
self,
unit_of_work: IUnitOfWork,
hash_service: IHashService,
cache: ICache,
logger: ILogger,
):
self._unit_of_work = unit_of_work
self._hash_service = hash_service
self._cache = cache
self._logger = logger
@staticmethod
def _normalize_email(email: str) -> str:
return email.strip().lower()
@transactional
async def __call__(
self,
*,
email: str,
code: str,
new_password: str,
confirm_password: str,
) -> bool:
code = (code or '').strip()
normalized = self._normalize_email(email)
EMAIL_PREFIX = 'forgot_password:email:'
CODE_PREFIX = 'forgot_password:code:'
if new_password != confirm_password:
self._logger.info('Forgot password failed: passwords do not match')
raise ApplicationException(400, 'Passwords do not match')
code_key = f'{CODE_PREFIX}{code}'
cached_email = await self._cache.get(code_key)
if not cached_email:
self._logger.info('Forgot password failed: code not found')
raise ApplicationException(400, 'Invalid or expired code')
if cached_email != normalized:
self._logger.info('Forgot password failed: code-email mismatch')
raise ApplicationException(400, 'Invalid or expired code')
email_key = f'{EMAIL_PREFIX}{normalized}'
code_hash = await self._cache.get(email_key)
if not code_hash:
self._logger.info('Forgot password failed: email key missing')
raise ApplicationException(400, 'Invalid or expired code')
ok = await self._hash_service.verify(hashed_value=code_hash, plain_value=code)
if not ok:
self._logger.info('Forgot password failed: code hash mismatch')
raise ApplicationException(400, 'Invalid or expired code')
user = await self._unit_of_work.user_repository.get_user_by_email(normalized)
if user is None:
self._logger.info('Forgot password failed: user not found after valid code')
raise ApplicationException(400, 'Invalid or expired code')
new_password_hash = await self._hash_service.hash(new_password)
user = await self._unit_of_work.user_repository.set_password(
user_id=user.id,
password_hash=new_password_hash,
)
await self._cache.set_user(user.id, user)
try:
await self._cache.delete(code_key)
await self._cache.delete(email_key)
except Exception as e:
self._logger.warning(f'Forgot password cleanup failed (user_id={user.id}): {e}')
self._logger.info(f'Password reset via forgot flow for user_id={user.id}')
return True

132
src/application/commands/forgot_password_start.py
View File

@@ -1,132 +0,0 @@
import secrets
from datetime import datetime, timezone
from ulid import ULID
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ICache, ILogger, IQueueMessanger
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config import settings
from src.infrastructure.context_vars import trace_id_var
from src.infrastructure.database.decorators import transactional
class ForgotPasswordStartCommand:
def __init__(
self,
hash_service: IHashService,
cache: ICache,
unit_of_work: IUnitOfWork,
logger: ILogger,
messanger: IQueueMessanger,
):
self._hash_service = hash_service
self._unit_of_work = unit_of_work
self._cache = cache
self._logger = logger
self._messanger = messanger
@staticmethod
def _normalize_email(email: str) -> str:
return email.strip().lower()
@transactional
async def __call__(self, email: str) -> bool:
TTL = 300
LOCK_TTL = 30
MAX_ATTEMPTS = 20
EMAIL_PREFIX = 'forgot_password:email:'
CODE_PREFIX = 'forgot_password:code:'
LOCK_PREFIX = 'forgot_password:lock:'
normalized = self._normalize_email(email)
user = await self._unit_of_work.user_repository.get_user_by_email(normalized)
if user is None:
self._logger.info(f'Forgot password start: no user for email hash lookup')
return True
trace_id = trace_id_var.get()
if not trace_id or trace_id == 'N/A':
trace_id = None
lock_key = f'{LOCK_PREFIX}{normalized}'
locked = await self._cache.set_nx(lock_key, '1', ttl=LOCK_TTL)
if not locked:
self._logger.info(f'Forgot password throttled by lock (user_id={user.id})')
raise ApplicationException(429, 'Too many requests. Please wait.')
try:
email_key = f'{EMAIL_PREFIX}{normalized}'
existing = await self._cache.get(email_key)
if existing:
self._logger.info(f'Forgot password denied: code already exists for user_id={user.id}')
raise ApplicationException(429, 'Code already sent. Please wait before retrying.')
for _ in range(MAX_ATTEMPTS):
code = f'{secrets.randbelow(1_000_000):06d}'
code_key = f'{CODE_PREFIX}{code}'
code_hash = await self._hash_service.hash(code)
reserved = await self._cache.set_nx(code_key, normalized, ttl=TTL)
if not reserved:
continue
saved = await self._cache.set(email_key, code_hash, ttl=TTL)
if not saved:
await self._cache.delete(code_key)
self._logger.error(f'Forgot password failed: cannot save code hash for user_id={user.id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
message_id = str(ULID())
now = datetime.now(timezone.utc).isoformat()
metadata = {
'trace_id': trace_id,
'source': 'user-service',
'timestamp': now,
'message_id': message_id,
}
payload = {
'email': normalized,
'code': code,
'ttl_seconds': TTL,
}
message = {
'event': 'forgot_password',
'payload': payload,
'metadata': metadata,
}
self._logger.info(f'Forgot password code created for user_id={user.id}')
try:
await self._messanger.publish_to_queue(
queue=settings.RABBIT_EMAIL_CODE_QUEUE,
message=message,
persist=True,
correlation_id=trace_id,
message_id=message_id,
headers={'trace_id': trace_id} if trace_id else None,
)
except Exception as exception:
try:
await self._cache.delete(email_key)
await self._cache.delete(code_key)
except Exception as rollback_err:
self._logger.error(
f'Publish failed and rollback cache failed for user_id={user.id}: {str(rollback_err)}'
)
self._logger.error(f'Failed to publish forgot password email for user_id={user.id}: {str(exception)}')
raise ApplicationException(503, 'Temporary error. Please try again.')
return True
self._logger.error(f'Forgot password failed: code space exhausted for user_id={user.id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
finally:
await self._cache.delete(lock_key)

17
src/application/commands/get_me.py
View File

@@ -1,17 +0,0 @@
from src.application.abstractions import IUnitOfWork
from src.application.contracts import ILogger, ICache
from src.application.domain.entities import UserEntity
from src.infrastructure.database.decorators import transactional
class GetMeCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger, cache: ICache):
self._unit_of_work = unit_of_work
self._logger = logger
self._cache = cache
@transactional
async def __call__(self, user_id: str) -> UserEntity:
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
self._logger.info(f'User ID: {user.id}')
return user

100
src/application/commands/set_avatar.py
View File

@@ -1,100 +0,0 @@
from __future__ import annotations
from datetime import datetime, timezone
from PIL import UnidentifiedImageError
from ulid import ULID
from botocore.exceptions import ClientError
from src.application.abstractions import IUnitOfWork
from src.application.contracts import ICache, ILogger, IS3
from src.application.domain.entities import UserEntity
from src.application.domain.exceptions import BadRequestException, ServiceUnavailableException
from src.infrastructure.config import settings
from src.infrastructure.database.decorators import transactional
from src.infrastructure.media.webp import image_bytes_to_webp
class SetAvatarCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger, cache: ICache, s3: IS3):
self._unit_of_work = unit_of_work
self._logger = logger
self._cache = cache
self._s3 = s3
@transactional
async def _load_user(self, user_id: str) -> UserEntity:
user = await self._unit_of_work.user_repository.get_user_by_id(user_id)
self._logger.debug(f'Avatar _load_user user_id={user_id} has_avatar_link={bool(user.avatar_link)}')
return user
async def __call__(self, user_id: str, image_bytes: bytes) -> tuple[UserEntity, int]:
prior = await self._load_user(user_id)
old_link = prior.avatar_link
self._logger.info(
f'SetAvatar start user_id={user_id} input_bytes={len(image_bytes)} had_previous_link={bool(old_link)}'
)
try:
webp_bytes = image_bytes_to_webp(image_bytes)
except UnidentifiedImageError as exc:
raise BadRequestException(message='Unsupported or corrupt image') from exc
except Exception as exc:
self._logger.exception(str(exc))
raise BadRequestException(message='Could not process image') from exc
self._logger.debug(f'SetAvatar webp_ready bytes={len(webp_bytes)}')
pid = user_id.replace('/', '').replace('.', '_')
name_id = str(ULID())
ts = int(datetime.now(timezone.utc).timestamp() * 1000)
prefix = settings.S3_AVATAR_KEY_PREFIX.strip().strip('/')
fname = f'{name_id}_{pid}_{ts}.webp'
object_key = f'{prefix}/{fname}' if prefix else fname
self._logger.info(f'SetAvatar S3 upload start user_id={user_id} key={object_key} webp_bytes={len(webp_bytes)}')
try:
url = await self._s3.upload_bytes(key=object_key, body=webp_bytes, content_type='image/webp')
except ClientError as exc:
self._logger.exception(str(exc))
raise ServiceUnavailableException(message='S3 upload failed') from exc
self._logger.info(f'SetAvatar S3 upload done user_id={user_id} key={object_key} public_url_len={len(url)}')
user = await self._save_avatar_link(user_id, url)
self._logger.info(
f'SetAvatar DB updated user_id={user_id} key={object_key} '
f'entity_avatar_link_len={len(user.avatar_link or "")}'
)
await self._cache.set_user(user_id, user)
self._logger.debug(f'SetAvatar cache updated user_id={user_id}')
if old_link:
old_key = self._s3.object_key_from_public_url(old_link)
if not old_key:
self._logger.warning(
f'SetAvatar could not parse old avatar URL for S3 delete user_id={user_id} '
f'old_link_len={len(old_link)}'
)
elif old_key == object_key:
self._logger.debug(f'SetAvatar skip delete same object key user_id={user_id} key={object_key}')
else:
self._logger.info(f'SetAvatar S3 delete old object user_id={user_id} old_key={old_key}')
try:
await self._s3.delete_object(key=old_key)
self._logger.info(f'SetAvatar S3 old object removed user_id={user_id} old_key={old_key}')
except ClientError as exc:
code = exc.response.get('Error', {}).get('Code', '')
if code not in ('NoSuchKey', '404'):
self._logger.warning(f'S3 delete old avatar failed user_id={user_id} code={code}: {exc}')
else:
self._logger.debug(f'SetAvatar old object already gone user_id={user_id} code={code}')
self._logger.info(f'Avatar set for user_id={user_id} key={object_key}')
return user, len(webp_bytes)
@transactional
async def _save_avatar_link(self, user_id: str, avatar_link: str) -> UserEntity:
self._logger.debug(f'SetAvatar DB transaction set_avatar_link user_id={user_id} link_len={len(avatar_link)}')
return await self._unit_of_work.user_repository.set_avatar_link(user_id, avatar_link)

68
src/application/commands/set_encrypted_mnemonic_complete.py
View File

@@ -1,68 +0,0 @@
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ILogger, ICache
from src.application.domain.entities import UserEntity
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.database.decorators import transactional
class SetEncryptedMnemonicCompleteCommand:
def __init__(
self,
unit_of_work: IUnitOfWork,
hash_service: IHashService,
cache: ICache,
logger: ILogger,
):
self._unit_of_work = unit_of_work
self._hash_service = hash_service
self._cache = cache
self._logger = logger
@transactional
async def __call__(self, *, user_id: str, code: str, encrypted_mnemonic: str) -> UserEntity:
code = (code or '').strip()
USER_PREFIX = 'encrypted_mnemonic:user:'
CODE_PREFIX = 'encrypted_mnemonic:code:'
user_key = f'{USER_PREFIX}{user_id}'
code_key = f'{CODE_PREFIX}{code}'
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
if user.encrypted_mnemonic is not None:
self._logger.info(f'Encrypted mnemonic already set for user_id={user_id}')
raise ApplicationException(409, 'Encrypted mnemonic already set and cannot be changed')
cached_user_id = await self._cache.get(code_key)
if not cached_user_id:
self._logger.info(f'Encrypted mnemonic set failed: code not found (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
if cached_user_id != user_id:
self._logger.info(f'Encrypted mnemonic set failed: code-user mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
code_hash = await self._cache.get(user_key)
if not code_hash:
self._logger.info(f'Encrypted mnemonic set failed: user key missing (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
ok = await self._hash_service.verify(hashed_value=code_hash, plain_value=code)
if not ok:
self._logger.info(f'Encrypted mnemonic set failed: code hash mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
user = await self._unit_of_work.user_repository.set_encrypted_mnemonic(
user_id=user_id,
encrypted_mnemonic=encrypted_mnemonic,
)
await self._cache.set_user(user_id, user)
try:
await self._cache.delete(code_key)
await self._cache.delete(user_key)
except Exception as e:
self._logger.warning(f'Encrypted mnemonic set cleanup failed (user_id={user_id}): {e}')
self._logger.info(f'Encrypted mnemonic set for user_id={user_id}')
return user

130
src/application/commands/set_encrypted_mnemonic_start.py
View File

@@ -1,130 +0,0 @@
import secrets
from datetime import datetime, timezone
from ulid import ULID
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ICache, ILogger, IQueueMessanger
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config import settings
from src.infrastructure.context_vars import trace_id_var
from src.infrastructure.database.decorators import transactional
class SetEncryptedMnemonicStartCommand:
def __init__(
self,
hash_service: IHashService,
cache: ICache,
unit_of_work: IUnitOfWork,
logger: ILogger,
messanger: IQueueMessanger,
):
self._hash_service = hash_service
self._unit_of_work = unit_of_work
self._cache = cache
self._logger = logger
self._messanger = messanger
@transactional
async def __call__(self, user_id: str) -> bool:
TTL = 300
LOCK_TTL = 30
MAX_ATTEMPTS = 20
USER_PREFIX = 'encrypted_mnemonic:user:'
CODE_PREFIX = 'encrypted_mnemonic:code:'
LOCK_PREFIX = 'encrypted_mnemonic:lock:'
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
if user.encrypted_mnemonic is not None:
self._logger.info(f'Encrypted mnemonic already set for user_id={user_id}')
raise ApplicationException(409, 'Encrypted mnemonic already set and cannot be changed')
if not user.email:
self._logger.warning(f'User {user_id} does not have an email address')
raise ApplicationException(404, f'User {user_id} does not have an email address')
trace_id = trace_id_var.get()
if not trace_id or trace_id == 'N/A':
trace_id = None
lock_key = f'{LOCK_PREFIX}{user_id}'
locked = await self._cache.set_nx(lock_key, '1', ttl=LOCK_TTL)
if not locked:
self._logger.info(f'Encrypted mnemonic set throttled by lock (user_id={user_id})')
raise ApplicationException(429, 'Too many requests. Please wait.')
try:
user_key = f'{USER_PREFIX}{user_id}'
existing = await self._cache.get(user_key)
if existing:
self._logger.info(f'Encrypted mnemonic set denied: code already exists for user_id={user_id}')
raise ApplicationException(429, 'Code already sent. Please wait before retrying.')
for _ in range(MAX_ATTEMPTS):
code = f'{secrets.randbelow(1_000_000):06d}'
code_key = f'{CODE_PREFIX}{code}'
code_hash = await self._hash_service.hash(code)
reserved = await self._cache.set_nx(code_key, user_id, ttl=TTL)
if not reserved:
continue
saved = await self._cache.set(user_key, code_hash, ttl=TTL)
if not saved:
await self._cache.delete(code_key)
self._logger.error(f'Encrypted mnemonic set failed: cannot save code hash for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
message_id = str(ULID())
now = datetime.now(timezone.utc).isoformat()
metadata = {
'trace_id': trace_id,
'source': 'user-service',
'timestamp': now,
'message_id': message_id,
}
payload = {
'email': user.email,
'code': code,
'ttl_seconds': TTL,
}
message = {
'event': 'encrypted_mnemonic_set',
'payload': payload,
'metadata': metadata,
}
self._logger.info(f'Encrypted mnemonic set code created for user_id={user_id}')
try:
await self._messanger.publish_to_queue(
queue=settings.RABBIT_EMAIL_CODE_QUEUE,
message=message,
persist=True,
correlation_id=trace_id,
message_id=message_id,
headers={'trace_id': trace_id} if trace_id else None,
)
except Exception as exception:
try:
await self._cache.delete(user_key)
await self._cache.delete(code_key)
except Exception as rollback_err:
self._logger.error(f'Publish failed and rollback cache failed for user_id={user_id}: {str(rollback_err)}')
self._logger.error(f'Failed to publish encrypted mnemonic set email for user_id={user_id}: {str(exception)}')
raise ApplicationException(503, 'Temporary error. Please try again.')
return True
self._logger.error(f'Encrypted mnemonic set failed: code space exhausted for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
finally:
await self._cache.delete(lock_key)

18
src/application/commands/set_phone.py
View File

@@ -1,18 +0,0 @@
from src.application.abstractions import IUnitOfWork
from src.application.contracts import ILogger, ICache
from src.application.domain.entities import UserEntity
from src.infrastructure.database.decorators import transactional
class SetPhoneCommand:
def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger, cache: ICache):
self._unit_of_work = unit_of_work
self._logger = logger
self._cache = cache
@transactional
async def __call__(self, user_id: str, phone: str) -> UserEntity:
user = await self._unit_of_work.user_repository.set_phone(user_id=user_id, phone=phone)
await self._cache.set_user(user_id, user)
self._logger.info(f'Set phone for user {user_id}')
return user

76
src/application/commands/update_bank_details_complete.py
View File

@@ -1,76 +0,0 @@
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ILogger, ICache
from src.application.domain.entities import UserEntity
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.database.decorators import transactional
class UpdateBankDetailsCompleteCommand:
def __init__(
self,
unit_of_work: IUnitOfWork,
hash_service: IHashService,
cache: ICache,
logger: ILogger,
):
self._unit_of_work = unit_of_work
self._hash_service = hash_service
self._cache = cache
self._logger = logger
@transactional
async def __call__(
self,
*,
user_id: str,
code: str,
passport_data: str | None = None,
inn: str | None = None,
erc20: str | None = None,
) -> UserEntity:
code = (code or '').strip()
USER_PREFIX = 'bank_details:user:'
CODE_PREFIX = 'bank_details:code:'
user_key = f'{USER_PREFIX}{user_id}'
code_key = f'{CODE_PREFIX}{code}'
cached_user_id = await self._cache.get(code_key)
if not cached_user_id:
self._logger.info(f'Bank details update failed: code not found (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
if cached_user_id != user_id:
self._logger.info(f'Bank details update failed: code-user mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
code_hash = await self._cache.get(user_key)
if not code_hash:
self._logger.info(f'Bank details update failed: user key missing (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
ok = await self._hash_service.verify(hashed_value=code_hash, plain_value=code)
if not ok:
self._logger.info(f'Bank details update failed: code hash mismatch (user_id={user_id})')
raise ApplicationException(400, 'Invalid or expired code')
fields = {}
if passport_data is not None:
fields['passport_data'] = passport_data
if inn is not None:
fields['inn'] = inn
if erc20 is not None:
fields['erc20'] = erc20
user = await self._unit_of_work.user_repository.set_bank_details(user_id, **fields)
await self._cache.set_user(user_id, user)
try:
await self._cache.delete(code_key)
await self._cache.delete(user_key)
except Exception as e:
self._logger.warning(f'Bank details update cleanup failed (user_id={user_id}): {e}')
self._logger.info(f'Bank details updated for user_id={user_id}, fields={list(fields.keys())}')
return user

126
src/application/commands/update_bank_details_start.py
View File

@@ -1,126 +0,0 @@
import secrets
from datetime import datetime, timezone
from ulid import ULID
from src.application.abstractions import IUnitOfWork
from src.application.contracts import IHashService, ICache, ILogger, IQueueMessanger
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.config import settings
from src.infrastructure.context_vars import trace_id_var
from src.infrastructure.database.decorators import transactional
class UpdateBankDetailsStartCommand:
def __init__(
self,
hash_service: IHashService,
cache: ICache,
unit_of_work: IUnitOfWork,
logger: ILogger,
messanger: IQueueMessanger,
):
self._hash_service = hash_service
self._unit_of_work = unit_of_work
self._cache = cache
self._logger = logger
self._messanger = messanger
@transactional
async def __call__(self, user_id: str) -> bool:
TTL = 300
LOCK_TTL = 30
MAX_ATTEMPTS = 20
USER_PREFIX = 'bank_details:user:'
CODE_PREFIX = 'bank_details:code:'
LOCK_PREFIX = 'bank_details:lock:'
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
if not user.email:
self._logger.warning(f'User {user_id} does not have an email address')
raise ApplicationException(status_code=404, message=f'User {user_id} does not have an email address')
trace_id = trace_id_var.get()
if not trace_id or trace_id == 'N/A':
trace_id = None
lock_key = f'{LOCK_PREFIX}{user_id}'
locked = await self._cache.set_nx(lock_key, '1', ttl=LOCK_TTL)
if not locked:
self._logger.info(f'Bank details update throttled by lock (user_id={user_id})')
raise ApplicationException(429, 'Too many requests. Please wait.')
try:
user_key = f'{USER_PREFIX}{user_id}'
existing = await self._cache.get(user_key)
if existing:
self._logger.info(f'Bank details update denied: code already exists for user_id={user_id}')
raise ApplicationException(429, 'Code already sent. Please wait before retrying.')
for _ in range(MAX_ATTEMPTS):
code = f'{secrets.randbelow(1_000_000):06d}'
code_key = f'{CODE_PREFIX}{code}'
code_hash = await self._hash_service.hash(code)
reserved = await self._cache.set_nx(code_key, user_id, ttl=TTL)
if not reserved:
continue
saved = await self._cache.set(user_key, code_hash, ttl=TTL)
if not saved:
await self._cache.delete(code_key)
self._logger.error(f'Bank details update failed: cannot save code hash for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
message_id = str(ULID())
now = datetime.now(timezone.utc).isoformat()
metadata = {
'trace_id': trace_id,
'source': 'user-service',
'timestamp': now,
'message_id': message_id,
}
payload = {
'email': user.email,
'code': code,
'ttl_seconds': TTL,
}
message = {
'event': 'bank_details_update',
'payload': payload,
'metadata': metadata,
}
self._logger.info(f'Bank details update code created for user_id={user_id}')
try:
await self._messanger.publish_to_queue(
queue=settings.RABBIT_EMAIL_CODE_QUEUE,
message=message,
persist=True,
correlation_id=trace_id,
message_id=message_id,
headers={'trace_id': trace_id} if trace_id else None,
)
except Exception as exception:
try:
await self._cache.delete(user_key)
await self._cache.delete(code_key)
except Exception as rollback_err:
self._logger.error(f'Publish failed and rollback cache failed for user_id={user_id}: {str(rollback_err)}')
self._logger.error(f'Failed to publish bank details update email for user_id={user_id}: {str(exception)}')
raise ApplicationException(503, 'Temporary error. Please try again.')
return True
self._logger.error(f'Bank details update failed: code space exhausted for user_id={user_id}')
raise ApplicationException(503, 'Temporary error. Please try again.')
finally:
await self._cache.delete(lock_key)

4
src/application/contracts/__init__.py
View File

@@ -1,7 +1,3 @@
from src.application.contracts.i_logger import ILogger
from src.application.contracts.i_jwt_service import IJwtService
from src.application.contracts.i_csrf_service import ICsrfService
from src.application.contracts.i_cache import ICache
from src.application.contracts.i_hash_service import IHashService
from src.application.contracts.i_queue_messanger import IQueueMessanger
from src.application.contracts.i_s3 import IS3

30
src/application/contracts/i_cache.py
View File

@@ -1,30 +0,0 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from src.application.domain.entities.user import UserEntity
class ICache(ABC):
@abstractmethod
async def set(self, key: str, value: str, ttl: int) -> bool:
raise NotImplementedError
@abstractmethod
async def set_nx(self, key: str, value: str, ttl: int) -> bool:
raise NotImplementedError
@abstractmethod
async def get(self, key: str) -> str | None:
raise NotImplementedError
@abstractmethod
async def delete(self, key: str) -> bool:
raise NotImplementedError
@abstractmethod
async def get_user(self, user_id: str) -> dict | None:
raise NotImplementedError
@abstractmethod
async def set_user(self, user_id: str, user: UserEntity, ttl: int = 300) -> None:
raise NotImplementedError

12
src/application/contracts/i_hash_service.py
View File

@@ -1,12 +0,0 @@
from abc import ABC, abstractmethod
class IHashService(ABC):
@abstractmethod
async def hash(self, value: str) -> str:
raise NotImplementedError
@abstractmethod
async def verify(self, hashed_value: str, plain_value: str) -> bool:
raise NotImplementedError

40
src/application/contracts/i_queue_messanger.py
View File

@@ -1,40 +0,0 @@
from abc import ABC, abstractmethod
from typing import Mapping, Any
class IQueueMessanger(ABC):
@abstractmethod
async def connect(self) -> None:
raise NotImplementedError
@abstractmethod
async def close(self) -> None:
raise NotImplementedError
@abstractmethod
async def publish_to_queue(
self,
queue: str,
message: Any,
*,
persist: bool = True,
headers: Mapping[str, Any] | None = None,
correlation_id: str | None = None,
message_id: str | None = None,
) -> None:
raise NotImplementedError
@abstractmethod
async def publish(
self,
message: Any,
*,
exchange: str,
routing_key: str,
persist: bool = True,
headers: Mapping[str, Any] | None = None,
correlation_id: str | None = None,
message_id: str | None = None,
) -> None:
raise NotImplementedError

17
src/application/contracts/i_s3.py
View File

@@ -1,17 +0,0 @@
from __future__ import annotations
from typing import Protocol, runtime_checkable
@runtime_checkable
class IS3(Protocol):
async def upload_bytes(self, *, key: str, body: bytes, content_type: str) -> str:
...
async def delete_object(self, *, key: str) -> None:
...
def object_key_from_public_url(self, url: str) -> str | None:
...

21
src/application/domain/password_policy.py
View File

@@ -1,21 +0,0 @@
import re
SPECIAL_CHARS = '!@#$%^&*()_+-=.,:;?/[]{}<>'
def validate_password_strength(password: str) -> str:
if re.search(r'\s', password):
raise ValueError('Password must not contain whitespace')
if len(password) < 12:
raise ValueError('Password must be at least 12 characters')
if not re.search(r'[a-z]', password):
raise ValueError('Password must contain at least one lowercase letter')
if not re.search(r'[A-Z]', password):
raise ValueError('Password must contain at least one uppercase letter')
if not re.search(r'\d', password):
raise ValueError('Password must contain at least one digit')
if not any(c in SPECIAL_CHARS for c in password):
raise ValueError(
'Password must contain at least one special character from: !@#$%^&*()_+-=.,:;?/[]{}<>'
)
return password

2
src/infrastructure/cache/__init__.py vendored
View File

@@ -1,2 +0,0 @@
from src.infrastructure.cache.client import create_redis_client
from src.infrastructure.cache.keydb_client import KeydbCache

16
src/infrastructure/cache/client.py vendored
View File

@@ -1,16 +0,0 @@
import redis.asyncio as redis
from redis.asyncio.client import Redis
from src.infrastructure.config import settings
def create_redis_client() -> Redis:
return redis.from_url(
settings.REDIS_URL,
max_connections=50,
decode_responses=True,
socket_timeout=5,
socket_connect_timeout=5,
health_check_interval=30,
retry_on_timeout=True,
socket_keepalive=True,
)

52
src/infrastructure/cache/keydb_client.py vendored
View File

@@ -1,52 +0,0 @@
from __future__ import annotations
import orjson
from redis.asyncio.client import Redis
from src.application.contracts import ICache
from src.application.domain.entities.user import UserEntity
class KeydbCache(ICache):
USER_PREFIX = 'user:me'
def __init__(self, redis_client: Redis):
self._r = redis_client
async def set(self, key: str, value: str, ttl: int) -> bool:
return bool(await self._r.set(key, value, ex=ttl))
async def set_nx(self, key: str, value: str, ttl: int) -> bool:
return bool(await self._r.set(key, value, ex=ttl, nx=True))
async def get(self, key: str) -> str | None:
return await self._r.get(key)
async def delete(self, key: str) -> bool:
return (await self._r.delete(key)) > 0
async def get_user(self, user_id: str) -> dict | None:
raw = await self._r.get(f'{self.USER_PREFIX}:{user_id}')
if raw is None:
return None
return orjson.loads(raw)
async def set_user(self, user_id: str, user: UserEntity, ttl: int = 300) -> None:
data = orjson.dumps({
'id': user.id,
'email': user.email,
'first_name': user.first_name,
'middle_name': user.middle_name,
'last_name': user.last_name,
'birth_date': str(user.birth_date) if user.birth_date else None,
'encrypted_mnemonic': user.encrypted_mnemonic,
'phone': user.phone,
'passport_data': user.passport_data,
'inn': user.inn,
'erc20': user.erc20,
'avatar_link': user.avatar_link,
'kyc_verified': user.kyc_verified,
'is_deleted': user.is_deleted,
'created_at': user.created_at.isoformat() if user.created_at else None,
'updated_at': user.updated_at.isoformat() if user.updated_at else None,
'kyc_verified_at': user.kyc_verified_at.isoformat() if user.kyc_verified_at else None,
})
await self._r.set(f'{self.USER_PREFIX}:{user_id}', data, ex=ttl)

128
src/infrastructure/config/settings.py
View File

@@ -1,7 +1,7 @@
from __future__ import annotations
from functools import lru_cache
from typing import Any, List, Literal, Mapping
from typing import Any, List, Literal
from urllib.parse import quote
from dotenv import find_dotenv, load_dotenv
@@ -34,13 +34,11 @@ class Settings(BaseSettings):
VAULT_NAMESPACE: str | None = None
VAULT_MOUNT_POINT: str = 'dev-secrets'
VAULT_DATABASE_SECRET_PATH: str = 'database'
VAULT_RABBIT_SECRET_PATH: str = 'rabbitmq'
VAULT_CSRF_SECRET_PATH: str = 'csrf'
VAULT_DOCS_SECRET_PATH: str = 'docs'
VAULT_JWT_KID_PATH: str = 'jwt/kid'
VAULT_JWT_KIDS_PREFIX: str = 'jwt/kids'
VAULT_CRYPTO_MASTER_KEY_PATH: str = 'crypto'
VAULT_S3_SECRET_PATH: str = 's3/avatars'
DATABASE_URL_DIRECT: str | None = Field(default=None, validation_alias='DATABASE_URL')
DATABASE_HOST: str = ''
@@ -76,30 +74,6 @@ class Settings(BaseSettings):
JWT_ALGORITHM: str = 'RS256'
JWT_KEYS_REFRESH_SECONDS: int = 3600
REDIS_HOST: str = 'keydb'
REDIS_PORT: int = 6379
REDIS_PASSWORD: str | None = None
REDIS_DB: int = 0
RABBIT_HOST: str = 'localhost'
RABBIT_PORT: int = 5672
RABBIT_USER: str = 'guest'
RABBIT_PASSWORD: str = 'guest'
RABBIT_VHOST: str = '/'
RABBIT_PUBLISH_PERSIST: bool = True
RABBIT_CONNECT_TIMEOUT: int = 5
RABBIT_EMAIL_CODE_QUEUE: str = 'email.verification_code'
S3_BUCKET: str = ''
S3_REGION: str = 'us-east-1'
S3_ACCESS_KEY_ID: str = ''
S3_SECRET_ACCESS_KEY: str = ''
S3_ENDPOINT_URL: str = ''
S3_PUBLIC_BASE_URL: str = ''
S3_REGRU_PUBLIC_WEBSITE_HOST: bool = False
S3_AVATAR_KEY_PREFIX: str = 'avatars'
LOG_LEVEL: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO'
LOG_FORMAT: Literal['JSON', 'TEXT'] = 'JSON'
@@ -110,78 +84,7 @@ class Settings(BaseSettings):
return str(value)
return ''
def _reset_s3_config(self) -> None:
object.__setattr__(self, 'S3_BUCKET', '')
object.__setattr__(self, 'S3_ACCESS_KEY_ID', '')
object.__setattr__(self, 'S3_SECRET_ACCESS_KEY', '')
object.__setattr__(self, 'S3_ENDPOINT_URL', '')
object.__setattr__(self, 'S3_PUBLIC_BASE_URL', '')
object.__setattr__(self, 'S3_REGION', 'us-east-1')
object.__setattr__(self, 'S3_REGRU_PUBLIC_WEBSITE_HOST', False)
object.__setattr__(self, 'S3_AVATAR_KEY_PREFIX', 'avatars')
@staticmethod
def _vault_kv(mapping: Mapping[str, Any], *keys: str) -> Any:
for k in keys:
if k in mapping and mapping[k] is not None:
return mapping[k]
return None
def _apply_s3_from_vault_secret(self, s3: dict[str, Any]) -> None:
bucket_name = (
self._vault_kv(s3, 'bucket_name', 'BUCKET_NAME', 'bucket')
or self._vault_kv(s3, 'S3_BUCKET', 'bucketName')
)
endpoint_url = (
self._vault_kv(s3, 's3_endpoint_url', 'S3_ENDPOINT_URL', 'endpoint_url', 'ENDPOINT_URL')
or self._vault_kv(s3, 'endpoint')
)
ak = (
self._vault_kv(s3, 's3_access_key_id', 'S3_ACCESS_KEY_ID', 'ACCESS_KEY_ID', 'access_key_id')
or self._vault_kv(s3, 'AWS_ACCESS_KEY_ID')
)
sk = (
self._vault_kv(s3, 's3_secret_access_key', 'S3_SECRET_ACCESS_KEY', 'SECRET_ACCESS_KEY')
or self._vault_kv(s3, 'AWS_SECRET_ACCESS_KEY')
)
if bucket_name is None or str(bucket_name).strip() == '':
raise ValueError('Vault S3 secret must contain bucket_name')
if endpoint_url is None or str(endpoint_url).strip() == '':
raise ValueError('Vault S3 secret must contain s3_endpoint_url')
if ak is None or str(ak).strip() == '':
raise ValueError('Vault S3 secret must contain s3_access_key_id')
if sk is None or str(sk).strip() == '':
raise ValueError('Vault S3 secret must contain s3_secret_access_key')
object.__setattr__(self, 'S3_BUCKET', str(bucket_name).strip())
object.__setattr__(self, 'S3_ENDPOINT_URL', str(endpoint_url).strip())
object.__setattr__(self, 'S3_ACCESS_KEY_ID', str(ak).strip())
object.__setattr__(self, 'S3_SECRET_ACCESS_KEY', str(sk).strip())
region = (
self._vault_kv(s3, 's3_region', 'S3_REGION', 'region')
)
if region is not None and str(region).strip() != '':
object.__setattr__(self, 'S3_REGION', str(region).strip())
public_base = (
self._vault_kv(s3, 's3_public_base_url', 'S3_PUBLIC_BASE_URL', 'public_base_url')
or self._vault_kv(s3, 'public_url')
)
if public_base is not None and str(public_base).strip() != '':
object.__setattr__(self, 'S3_PUBLIC_BASE_URL', str(public_base).strip())
prefix = self._vault_kv(s3, 'avatar_key_prefix', 'S3_AVATAR_KEY_PREFIX', 's3_avatar_key_prefix')
if prefix is not None and str(prefix).strip() != '':
object.__setattr__(self, 'S3_AVATAR_KEY_PREFIX', str(prefix).strip())
rf = (
self._vault_kv(s3, 's3_reg_ru_public_website_host', 'S3_REGRU_PUBLIC_WEBSITE_HOST')
)
if rf is not None:
v = str(rf).strip().lower()
if v in {'1', 'true', 'yes', 'on'}:
object.__setattr__(self, 'S3_REGRU_PUBLIC_WEBSITE_HOST', True)
elif v in {'0', 'false', 'no', 'off'}:
object.__setattr__(self, 'S3_REGRU_PUBLIC_WEBSITE_HOST', False)
def model_post_init(self, __context: Any) -> None:
self._reset_s3_config()
if not self.VAULT_ROLE_ID.strip() or not self.VAULT_SECRET_ID.strip():
if not self.DATABASE_URL:
raise ValueError(
@@ -218,19 +121,6 @@ class Settings(BaseSettings):
if kv(db, 'PASSWORD', 'password') is not None:
object.__setattr__(self, 'DATABASE_PASSWORD', str(kv(db, 'PASSWORD', 'password')))
rabbit = client.read_secret_optional(self.VAULT_RABBIT_SECRET_PATH)
if rabbit:
if kv(rabbit, 'HOST', 'host') is not None:
object.__setattr__(self, 'RABBIT_HOST', str(kv(rabbit, 'HOST', 'host')))
if kv(rabbit, 'PORT', 'port') is not None:
object.__setattr__(self, 'RABBIT_PORT', _as_int(kv(rabbit, 'PORT', 'port'), self.RABBIT_PORT))
if kv(rabbit, 'USER', 'user') is not None:
object.__setattr__(self, 'RABBIT_USER', str(kv(rabbit, 'USER', 'user')))
if kv(rabbit, 'PASSWORD', 'password') is not None:
object.__setattr__(self, 'RABBIT_PASSWORD', str(kv(rabbit, 'PASSWORD', 'password')))
if kv(rabbit, 'VHOST', 'vhost') is not None:
object.__setattr__(self, 'RABBIT_VHOST', str(kv(rabbit, 'VHOST', 'vhost')))
csrf = client.read_secret_optional(self.VAULT_CSRF_SECRET_PATH)
if csrf and kv(csrf, 'KEY', 'key') is not None:
key = str(kv(csrf, 'KEY', 'key'))
@@ -246,12 +136,6 @@ class Settings(BaseSettings):
if p is not None:
object.__setattr__(self, 'DOCS_PASSWORD', str(p))
s3_rel_path = self.VAULT_S3_SECRET_PATH.strip()
if s3_rel_path:
s3_secret_data = client.read_secret_optional(s3_rel_path)
if s3_secret_data:
self._apply_s3_from_vault_secret(s3_secret_data)
if not self.DATABASE_URL:
raise ValueError('Database URL could not be built from Vault database secret')
@@ -289,16 +173,6 @@ class Settings(BaseSettings):
quoted_password = quote(password, safe='')
return f'postgresql+asyncpg://{quoted_user}:{quoted_password}@{host}:{port}/{name}'
@property
def REDIS_URL(self) -> str:
auth = f':{self.REDIS_PASSWORD}@' if self.REDIS_PASSWORD else ''
return f'redis://{auth}{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}'
@property
def RABBIT_URL(self) -> str:
vhost = '%2F' if self.RABBIT_VHOST == '/' else self.RABBIT_VHOST.lstrip('/')
return f'amqp://{self.RABBIT_USER}:{self.RABBIT_PASSWORD}@{self.RABBIT_HOST}:{self.RABBIT_PORT}/{vhost}'
@property
def EXCLUDED_PATHS(self) -> List[str]:
return ['/docs', '/redoc', '/openapi.json', '/ping', '/health']

18
src/infrastructure/media/webp.py
View File

@@ -1,18 +0,0 @@
from __future__ import annotations
from io import BytesIO
from PIL import Image
def image_bytes_to_webp(raw: bytes, *, quality: int = 82) -> bytes:
im = Image.open(BytesIO(raw))
if im.mode == 'P':
im = im.convert('RGBA')
elif im.mode == 'LA':
im = im.convert('RGBA')
elif im.mode not in ('RGBA', 'RGB'):
im = im.convert('RGB')
out = BytesIO()
im.save(out, format='WEBP', quality=quality)
return out.getvalue()

1
src/infrastructure/messanger/__init__.py
View File

@@ -1 +0,0 @@
from src.infrastructure.messanger.rabbit_client import RabbitClient

72
src/infrastructure/messanger/rabbit_client.py
View File

@@ -1,72 +0,0 @@
from typing import Any, Mapping
from faststream.rabbit import RabbitBroker
from src.application.contracts import IQueueMessanger
from src.infrastructure.config import settings
class RabbitClient(IQueueMessanger):
def __init__(self) -> None:
self._broker = RabbitBroker(
settings.RABBIT_URL,
)
self._connected = False
async def connect(self) -> None:
if self._connected:
return
await self._broker.connect()
self._connected = True
async def close(self) -> None:
if not self._connected:
return
await self._broker.close()
self._connected = False
async def _ensure_connected(self) -> None:
if not self._connected:
await self.connect()
async def publish_to_queue(
self,
queue: str,
message: Any,
*,
persist: bool | None = None,
headers: Mapping[str, Any] | None = None,
correlation_id: str | None = None,
message_id: str | None = None,
) -> None:
await self._ensure_connected()
await self._broker.publish(
message,
queue=queue,
persist=settings.RABBIT_PUBLISH_PERSIST if persist is None else persist,
headers=headers,
correlation_id=correlation_id,
message_id=message_id,
)
async def publish(
self,
message: Any,
*,
exchange: str,
routing_key: str,
persist: bool | None = None,
headers: Mapping[str, Any] | None = None,
correlation_id: str | None = None,
message_id: str | None = None,
) -> None:
await self._ensure_connected()
await self._broker.publish(
message,
exchange=exchange,
routing_key=routing_key,
persist=settings.RABBIT_PUBLISH_PERSIST if persist is None else persist,
headers=headers,
correlation_id=correlation_id,
message_id=message_id,
)

17
src/infrastructure/security/hash.py
View File

@@ -1,17 +0,0 @@
import bcrypt
from src.application.contracts import IHashService, ILogger
class HashService(IHashService):
def __init__(self, logger: ILogger):
self._logger = logger
async def hash(self, value: str) -> str:
hashed_value = bcrypt.hashpw(value.encode(), bcrypt.gensalt())
self._logger.info(f'Hash value {hashed_value.decode()}')
return hashed_value.decode()
async def verify(self, hashed_value: str, plain_value: str) -> bool:
self._logger.info(f'Hash value {hashed_value[:10]}')
return bcrypt.checkpw(plain_value.encode(), hashed_value.encode())

125
src/infrastructure/storage/s3_service.py
View File

@@ -1,125 +0,0 @@
from __future__ import annotations
from aiobotocore.session import get_session
class S3Service:
def __init__(
self,
*,
bucket: str,
region: str,
access_key_id: str | None,
secret_access_key: str | None,
public_base_url: str | None,
endpoint_url: str | None,
use_reg_ru_website_public_host: bool,
):
self._bucket = bucket
self._region = region or 'us-east-1'
self._access_key_id = access_key_id
self._secret_access_key = secret_access_key
pb = (public_base_url or '').strip().rstrip('/')
self._public_base_url = pb if pb else None
self._endpoint_url = endpoint_url.strip().rstrip('/') if endpoint_url and endpoint_url.strip() else None
self._use_reg_ru_website_public_host = use_reg_ru_website_public_host
@staticmethod
def _url_prefix_variants(prefix: str) -> list[str]:
p = prefix.rstrip('/') + '/'
out = [p]
if p.startswith('https://'):
out.append('http://' + p[8:])
elif p.startswith('http://'):
out.append('https://' + p[7:])
return out
def _public_url_prefixes(self) -> list[str]:
acc: list[str] = []
pb = self._public_base_url
if pb:
acc.extend(self._url_prefix_variants(pb))
ep = self._endpoint_url
if ep:
base = f'{ep.rstrip("/")}/{self._bucket}'
acc.extend(self._url_prefix_variants(base))
if ep and self._use_reg_ru_website_public_host and 's3.regru.cloud' in ep.lower():
wh = f'https://{self._bucket}.website.regru.cloud'
acc.extend(self._url_prefix_variants(wh))
if not ep:
if self._region == 'us-east-1':
h = f'https://{self._bucket}.s3.amazonaws.com'
else:
h = f'https://{self._bucket}.s3.{self._region}.amazonaws.com'
acc.extend(self._url_prefix_variants(h))
seen: set[str] = set()
uniq: list[str] = []
for x in sorted(acc, key=len, reverse=True):
if x not in seen:
seen.add(x)
uniq.append(x)
return uniq
def object_key_from_public_url(self, url: str) -> str | None:
u = (url or '').strip()
if not u:
return None
for p in self._public_url_prefixes():
if u.startswith(p):
k = u[len(p):].split('?', 1)[0].split('#', 1)[0]
return k if k else None
return None
def _object_url(self, key: str) -> str:
if self._public_base_url:
return f'{self._public_base_url}/{key}'
endpoint = self._endpoint_url
if endpoint:
if (
self._use_reg_ru_website_public_host
and 's3.regru.cloud' in endpoint.lower()
):
return f'https://{self._bucket}.website.regru.cloud/{key}'
return f'{endpoint}/{self._bucket}/{key}'
region = self._region
if region == 'us-east-1':
host = 's3.amazonaws.com'
else:
host = f's3.{region}.amazonaws.com'
return f'https://{self._bucket}.{host}/{key}'
async def upload_bytes(self, *, key: str, body: bytes, content_type: str) -> str:
session = get_session()
kw: dict[str, object] = {'region_name': self._region}
aid = self._access_key_id
sk = self._secret_access_key
ep = self._endpoint_url
if aid:
kw['aws_access_key_id'] = aid
if sk:
kw['aws_secret_access_key'] = sk
if ep:
kw['endpoint_url'] = ep
async with session.create_client('s3', **kw) as client:
await client.put_object(
Bucket=self._bucket,
Key=key,
Body=body,
ContentType=content_type,
)
return self._object_url(key)
async def delete_object(self, *, key: str) -> None:
session = get_session()
kw: dict[str, object] = {'region_name': self._region}
aid = self._access_key_id
sk = self._secret_access_key
ep = self._endpoint_url
if aid:
kw['aws_access_key_id'] = aid
if sk:
kw['aws_secret_access_key'] = sk
if ep:
kw['endpoint_url'] = ep
async with session.create_client('s3', **kw) as client:
await client.delete_object(Bucket=self._bucket, Key=key)

4
src/main.py
View File

@@ -10,7 +10,6 @@ from starlette.middleware.cors import CORSMiddleware
from starlette.exceptions import HTTPException
from fastapi.exceptions import RequestValidationError
from src.application.domain.exceptions import ApplicationException, UnauthorizedException
from src.infrastructure.cache import create_redis_client
from src.infrastructure.vault import JwtKeyStore, start_jwt_keys_scheduler
from src.infrastructure.utils import generate_instance_id
from src.infrastructure.logger import logger
@@ -46,8 +45,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
logger.set_instance_id(instance_id)
logger.info(f'B2B service instance started with id {instance_id}')
app.state.redis = create_redis_client()
if not settings.VAULT_ROLE_ID.strip() or not settings.VAULT_SECRET_ID.strip():
raise RuntimeError('VAULT_ROLE_ID and VAULT_SECRET_ID must be set')
@@ -78,7 +75,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
)
logger.info('Crypto master key loaded from Vault')
yield
await app.state.redis.aclose()
logger.info(f'B2B service instance ended with id {instance_id}')

46
src/presentation/decorators/cache.py
View File

@@ -1,46 +0,0 @@
from __future__ import annotations
import functools
from typing import Any, Awaitable, Callable
from fastapi import Request
from fastapi.responses import ORJSONResponse
from src.infrastructure.cache import KeydbCache
from src.infrastructure.logger import get_logger
from src.presentation.dependencies.cache import get_redis
def cached(*, prefix: str) -> Callable:
def decorator(func: Callable[..., Awaitable[Any]]):
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
logger = get_logger()
request = kwargs.get('request')
if not isinstance(request, Request):
for a in args:
if isinstance(a, Request):
request = a
break
auth = kwargs.get('auth')
user_id = getattr(auth, 'user_id', None) if auth else None
if request is None or user_id is None:
return await func(*args, **kwargs)
cache_key = f'{prefix}:{user_id}'
try:
redis = get_redis(request)
cache = KeydbCache(redis)
hit = await cache.get_user(user_id)
if hit is not None:
logger.debug(f'Cache hit key={cache_key}')
return ORJSONResponse(status_code=200, content=hit)
except Exception as e:
logger.warning(f'Cache read failed key={cache_key} error={e}')
return await func(*args, **kwargs)
return wrapper
return decorator

171
src/presentation/decorators/rate_limit.py
View File

@@ -1,171 +0,0 @@
from __future__ import annotations
import functools
import inspect
import hashlib
from typing import Any, Awaitable, Callable, Literal, Optional, Protocol, runtime_checkable
from fastapi import Request
from redis.asyncio.client import Redis
from src.application.contracts import ILogger
from src.application.domain.exceptions import ApplicationException
from src.infrastructure.logger import get_logger
from src.presentation.dependencies import get_redis
def _find_request(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Request:
req = kwargs.get('request')
if isinstance(req, Request):
return req
for a in args:
if isinstance(a, Request):
return a
raise RuntimeError('rate_limit decorator requires fastapi.Request argument')
def _client_ip(request: Request) -> str:
xff = request.headers.get('x-forwarded-for')
if xff:
return xff.split(',')[0].strip()
if request.client:
return request.client.host
return 'unknown'
_LUA_INCR_EXPIRE_TTL = '''
local key = KEYS[1]
local window = tonumber(ARGV[1])
local current = redis.call('INCR', key)
if current == 1 then
redis.call('EXPIRE', key, window)
end
local ttl = redis.call('TTL', key)
return { current, ttl }
'''
Scope = Literal['ip', 'device', 'user', 'key']
@runtime_checkable
class KeyBuilder1(Protocol):
def __call__(self, request: Request) -> str: ...
@runtime_checkable
class KeyBuilder3(Protocol):
def __call__(self, request: Request, args: tuple[Any, ...], kwargs: dict[str, Any]) -> str: ...
KeyBuilder = KeyBuilder1 | KeyBuilder3
def _call_key_builder(builder: KeyBuilder, request: Request, args: tuple[Any, ...], kwargs: dict[str, Any]) -> str:
try:
sig = inspect.signature(builder)
if len(sig.parameters) >= 3:
return builder(request, args, kwargs)
return builder(request)
except Exception as e:
try:
return builder(request, args, kwargs)
except Exception:
raise e
def _email_rl_key(request: Request, args: tuple[Any, ...], kwargs: dict[str, Any]) -> str:
body = kwargs.get('body')
if body is None and args:
for a in args:
if hasattr(a, 'email'):
body = a
break
email = (getattr(body, 'email', '') or '').strip().lower()
if not email:
email = _client_ip(request)
digest = hashlib.sha256(email.encode('utf-8')).hexdigest()[:24]
return f'email:{digest}'
def rate_limit(
*,
limit: int,
window_seconds: int,
scope: Scope = 'ip',
key_prefix: str = 'rl',
key_builder: Optional[KeyBuilder] = None,
fail_open: bool = True,
) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]:
if limit <= 0:
raise ValueError('rate_limit: limit must be > 0')
if window_seconds <= 0:
raise ValueError('rate_limit: window_seconds must be > 0')
if scope == 'key' and not key_builder:
raise ValueError('rate_limit: scope="key" requires key_builder')
def decorator(func: Callable[..., Awaitable[Any]]):
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any):
request = _find_request(args, kwargs)
logger: ILogger = get_logger()
if scope == 'ip':
ident = _client_ip(request)
elif scope == 'device':
ident = request.cookies.get('device_id') or _client_ip(request)
elif scope == 'user':
user = getattr(request.state, 'user', None)
user_id = getattr(user, 'id', None) if user else None
ident = str(user_id) if user_id else _client_ip(request)
else:
try:
ident = _call_key_builder(key_builder, request, args, kwargs) # type: ignore[arg-type]
except Exception as e:
logger.error(f'RateLimit key_builder failed error={str(e)}')
raise ApplicationException(500, 'Rate limiter key_builder failed')
route = request.url.path
method = request.method
redis_key = f'{key_prefix}:{scope}:{method}:{route}:{ident}'
logger.debug(f'RateLimit check key={redis_key} limit={limit} window={window_seconds}')
try:
redis: Redis = get_redis(request)
result = await redis.eval(
_LUA_INCR_EXPIRE_TTL,
1,
redis_key,
str(window_seconds),
)
count = int(result[0])
ttl_raw = int(result[1]) if result and len(result) > 1 else window_seconds
ttl = window_seconds if ttl_raw < 0 else ttl_raw
except Exception as e:
logger.error(f'RateLimit redis failure key={redis_key} error={str(e)}')
if fail_open:
logger.warning(f'RateLimit fail-open activated key={redis_key}')
return await func(*args, **kwargs)
raise ApplicationException(503, 'Rate limiter unavailable')
if count > limit:
retry_after = max(ttl, 0)
logger.warning(f'RateLimit exceeded key={redis_key} count={count} limit={limit} retry_after={retry_after}')
raise ApplicationException(
status_code=429,
message='Too Many Requests',
headers={'Retry-After': str(retry_after)},
)
logger.debug(f'RateLimit passed key={redis_key} count={count}')
return await func(*args, **kwargs)
return wrapper
return decorator

1
src/presentation/dependencies/__init__.py
View File

@@ -4,4 +4,3 @@ from src.presentation.dependencies.commands import (
get_list_purchase_requests_command,
)
from src.presentation.dependencies.security import get_jwt_service
from src.presentation.dependencies.cache import get_redis, get_cache

12
src/presentation/dependencies/cache.py
View File

@@ -1,12 +0,0 @@
from fastapi import Depends, Request
from redis.asyncio.client import Redis
from src.application.contracts import ICache
from src.infrastructure.cache import KeydbCache
def get_redis(request: Request) -> Redis:
return request.app.state.redis
def get_cache(redis_client: Redis = Depends(get_redis)) -> ICache:
return KeydbCache(redis_client)

8
src/presentation/dependencies/queue_messanger.py
View File

@@ -1,8 +0,0 @@
from functools import lru_cache
from src.application.contracts import IQueueMessanger
from src.infrastructure.messanger import RabbitClient
@lru_cache(maxsize=1)
def get_rabbit() -> IQueueMessanger:
return RabbitClient()

32
src/presentation/dependencies/s3_storage.py
View File

@@ -1,32 +0,0 @@
from __future__ import annotations
from src.application.contracts import IS3
from src.application.domain.exceptions import ServiceUnavailableException
from src.infrastructure.config import settings
from src.infrastructure.storage.s3_service import S3Service
_s3singleton: IS3 | None = None
def get_s3_storage() -> IS3:
global _s3singleton
if _s3singleton is not None:
return _s3singleton
if not settings.S3_BUCKET.strip():
raise ServiceUnavailableException(message='S3 is not configured')
endpoint = settings.S3_ENDPOINT_URL.strip()
pub = settings.S3_PUBLIC_BASE_URL.strip()
if not pub and not endpoint:
raise ServiceUnavailableException(message='Set S3_ENDPOINT_URL (or S3_PUBLIC_BASE_URL for a custom CDN base)')
ak = settings.S3_ACCESS_KEY_ID.strip() or None
sk = settings.S3_SECRET_ACCESS_KEY.strip() or None
_s3singleton = S3Service(
bucket=settings.S3_BUCKET.strip(),
region=settings.S3_REGION.strip() or 'us-east-1',
access_key_id=ak,
secret_access_key=sk,
public_base_url=pub if pub else None,
endpoint_url=endpoint if endpoint else None,
use_reg_ru_website_public_host=settings.S3_REGRU_PUBLIC_WEBSITE_HOST,
)
return _s3singleton

13
src/presentation/dependencies/security.py
View File

@@ -1,22 +1,11 @@
from functools import lru_cache
from fastapi import Depends
from src.application.contracts import IJwtService, ILogger, IHashService
from src.application.contracts import IJwtService, ILogger
from src.infrastructure.security.jwt import JwtService
from src.infrastructure.vault import JwtKeyStore
from src.presentation.dependencies.logger import get_logger
@lru_cache(maxsize=1)
def _hash_service(logger: ILogger) -> IHashService:
from src.infrastructure.security.hash import HashService
return HashService(logger=logger)
def get_hash_service(logger: ILogger = Depends(get_logger)) -> IHashService:
return _hash_service(logger)
@lru_cache(maxsize=1)
def _jwt_service(logger: ILogger) -> IJwtService:
key_store = JwtKeyStore.get_instance()

55
src/presentation/routing/account.py
View File

@@ -1,55 +0,0 @@
from fastapi import APIRouter, Depends, Request
from fastapi.responses import ORJSONResponse
from starlette import status
from src.application.commands.get_me import GetMeCommand
from src.application.contracts import ILogger
from src.application.domain.dto import AuthContext
from src.presentation.decorators import require_access_token
from src.presentation.dependencies.commands import get_get_me_command
from src.presentation.dependencies.logger import get_logger
from src.presentation.decorators import csrf_protect
from src.presentation.schemas.api_errors import ApiErrorPayload, ApiValidationErrorsPayload
from src.presentation.schemas.me_public import MeUserPublicResponse
from src.presentation.serializers import me_user_public
account_router = APIRouter()
@account_router.get(
path='/',
response_class=ORJSONResponse,
status_code=status.HTTP_200_OK,
response_model=MeUserPublicResponse,
summary='Текущий пользователь',
description=(
'Возвращает профиль, в том числе поле avatar_link — публичный HTTPS URL '
'текущего аватара (WebP в объектном хранилище), либо null, если аватар не задан. '
'Изображение по этому URL отдаёт не Users API, а хранилище; отдельного эндпоинта загрузки байтов аватара нет. '
'Загрузить или сменить аватар можно через PATCH /me/settings/avatar. '
'Защита CSRF: cookie csrf_token и заголовок X-CSRF-Token с тем же значением.'
),
responses={
status.HTTP_401_UNAUTHORIZED: {
'description': 'Не передан или неверен access token.',
'model': ApiErrorPayload,
},
status.HTTP_403_FORBIDDEN: {
'description': 'Ошибка проверки CSRF (нет пары cookie/заголовка, несовпадение или просрочен токен).',
'model': ApiErrorPayload,
},
status.HTTP_422_UNPROCESSABLE_ENTITY: {
'description': 'Ошибка валидации входных данных (например, заголовков).',
'model': ApiValidationErrorsPayload,
},
},
)
@csrf_protect()
async def me(
request: Request,
auth: AuthContext = Depends(require_access_token),
command: GetMeCommand = Depends(get_get_me_command),
logger: ILogger = Depends(get_logger),
) -> MeUserPublicResponse:
user = await command(user_id=auth.user_id)
logger.info(f'Get user: {user.id}')
return me_user_public(user)

384
src/presentation/routing/account_settings.py
View File

@@ -1,384 +0,0 @@
from fastapi import APIRouter, Depends, Request
from fastapi.responses import ORJSONResponse
from starlette import status
from src.application.commands import (
SetPhoneCommand,
SetAvatarCommand,
DeleteAvatarCommand,
ChangePasswordStartCommand,
ChangePasswordCompleteCommand,
ForgotPasswordStartCommand,
ForgotPasswordCompleteCommand,
)
from src.application.domain.dto import AuthContext
from src.presentation.decorators import require_access_token, csrf_protect, rate_limit, email_rl_key
from src.presentation.dependencies import (
get_delete_avatar_command,
get_set_avatar_command,
get_set_phone_command,
get_change_password_start_command,
get_change_password_complete_command,
get_forgot_password_start_command,
get_forgot_password_complete_command,
)
from src.presentation.schemas import (
SetAvatarRequest,
SetPhoneRequest,
ChangePasswordConfirmRequest,
ForgotPasswordStartRequest,
ForgotPasswordCompleteRequest,
)
from src.presentation.schemas.api_errors import ApiErrorPayload, ApiValidationErrorsPayload
from src.presentation.schemas.me_public import MeUserPublicResponse, SetAvatarPublicResponse
from src.presentation.serializers import me_user_public
account_settings_router = APIRouter(prefix='/settings')
_SET_AVATAR_ERROR_RESPONSES: dict[int, dict[str, object]] = {
status.HTTP_400_BAD_REQUEST: {
'description': 'Битый или неподдерживаемый формат изображения, либо Base64 ошибочный.',
'model': ApiErrorPayload,
},
status.HTTP_401_UNAUTHORIZED: {
'description': 'Не передан или неверен access token.',
'model': ApiErrorPayload,
},
status.HTTP_404_NOT_FOUND: {
'description': 'Учётная запись не найдена.',
'model': ApiErrorPayload,
},
status.HTTP_422_UNPROCESSABLE_ENTITY: {
'description': 'Тело запроса не соответствует схеме (например, неверный Base64 или превышен размер).',
'model': ApiValidationErrorsPayload,
},
status.HTTP_500_INTERNAL_SERVER_ERROR: {
'description': 'Внутренняя ошибка сервера; клиенту отдаётся обобщённое сообщение.',
'model': ApiErrorPayload,
},
status.HTTP_503_SERVICE_UNAVAILABLE: {
'description': 'S3 не сконфигурирован, ошибка записи в хранилище или временная недоступность сервиса.',
'model': ApiErrorPayload,
},
}
_PASSWORD_ERROR_RESPONSES: dict[int, dict[str, object]] = {
status.HTTP_400_BAD_REQUEST: {
'description': 'Неверный или просроченный код, пароли не совпадают или совпадают с текущим.',
'model': ApiErrorPayload,
},
status.HTTP_401_UNAUTHORIZED: {
'description': 'Не передан или неверен access token.',
'model': ApiErrorPayload,
},
status.HTTP_404_NOT_FOUND: {
'description': 'Учётная запись не найдена или у пользователя нет email.',
'model': ApiErrorPayload,
},
status.HTTP_422_UNPROCESSABLE_ENTITY: {
'description': 'Тело запроса не соответствует схеме (код, длина пароля).',
'model': ApiValidationErrorsPayload,
},
status.HTTP_429_TOO_MANY_REQUESTS: {
'description': 'Код уже отправлен или слишком частые запросы.',
'model': ApiErrorPayload,
},
status.HTTP_500_INTERNAL_SERVER_ERROR: {
'description': 'Внутренняя ошибка сервера.',
'model': ApiErrorPayload,
},
status.HTTP_503_SERVICE_UNAVAILABLE: {
'description': 'Временная ошибка отправки кода или сохранения в кеш.',
'model': ApiErrorPayload,
},
}
_FORGOT_PASSWORD_ERROR_RESPONSES: dict[int, dict[str, object]] = {
status.HTTP_400_BAD_REQUEST: {
'description': 'Неверный или просроченный код, пароли не совпадают.',
'model': ApiErrorPayload,
},
status.HTTP_422_UNPROCESSABLE_ENTITY: {
'description': 'Тело запроса не соответствует схеме.',
'model': ApiValidationErrorsPayload,
},
status.HTTP_429_TOO_MANY_REQUESTS: {
'description': 'Код уже отправлен или слишком частые запросы.',
'model': ApiErrorPayload,
},
status.HTTP_500_INTERNAL_SERVER_ERROR: {
'description': 'Внутренняя ошибка сервера.',
'model': ApiErrorPayload,
},
status.HTTP_503_SERVICE_UNAVAILABLE: {
'description': 'Временная ошибка отправки кода или сохранения в кеш.',
'model': ApiErrorPayload,
},
}
_DELETE_AVATAR_ERROR_RESPONSES: dict[int, dict[str, object]] = {
status.HTTP_401_UNAUTHORIZED: {
'description': 'Не передан или неверен access token.',
'model': ApiErrorPayload,
},
status.HTTP_404_NOT_FOUND: {
'description': 'Учётная запись не найдена.',
'model': ApiErrorPayload,
},
status.HTTP_500_INTERNAL_SERVER_ERROR: {
'description': 'Внутренняя ошибка сервера; клиенту отдаётся обобщённое сообщение.',
'model': ApiErrorPayload,
},
status.HTTP_503_SERVICE_UNAVAILABLE: {
'description': 'S3 не сконфигурирован или временная недоступность удаления объекта.',
'model': ApiErrorPayload,
},
}
@account_settings_router.patch(path='/phone', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
@csrf_protect()
async def set_phone(
request: Request,
body: SetPhoneRequest,
auth: AuthContext = Depends(require_access_token),
command: SetPhoneCommand = Depends(get_set_phone_command),
):
user = await command(user_id=auth.user_id, phone=body.phone)
return ORJSONResponse(status_code=status.HTTP_200_OK, content={'phone': user.phone})
@account_settings_router.patch(
path='/avatar',
response_class=ORJSONResponse,
status_code=status.HTTP_200_OK,
response_model=SetAvatarPublicResponse,
summary='Обновить аватар',
description=(
'Принимает фото в Base64, сохраняет как WebP в объектном хранилище и записывает публичный URL в профиль. '
'После успешной записи удаляется предыдущий объект в S3 (если ссылку удаётся сопоставить с ключом).'
),
response_description=(
'Профиль пользователя в том же формате, что и GET /me, плюс размер сохранённого WebP.'
),
responses=_SET_AVATAR_ERROR_RESPONSES,
)
@csrf_protect()
async def set_avatar(
request: Request,
body: SetAvatarRequest,
auth: AuthContext = Depends(require_access_token),
command: SetAvatarCommand = Depends(get_set_avatar_command),
) -> SetAvatarPublicResponse:
user, webp_size = await command(user_id=auth.user_id, image_bytes=body.decoded_bytes)
pub = me_user_public(user)
return SetAvatarPublicResponse(**pub.model_dump(), webp_size_bytes=webp_size)
@account_settings_router.delete(
path='/avatar',
response_class=ORJSONResponse,
status_code=status.HTTP_200_OK,
response_model=MeUserPublicResponse,
summary='Удалить аватар',
description=(
'Удаляет файл в объектном хранилище при известном URL и обнуляет avatar_link в профиле.'
),
responses=_DELETE_AVATAR_ERROR_RESPONSES,
)
@csrf_protect()
async def delete_avatar(
request: Request,
auth: AuthContext = Depends(require_access_token),
command: DeleteAvatarCommand = Depends(get_delete_avatar_command),
) -> MeUserPublicResponse:
user = await command(user_id=auth.user_id)
return me_user_public(user)
@account_settings_router.post(
path='/password/start',
response_class=ORJSONResponse,
status_code=status.HTTP_200_OK,
summary='Запросить код для смены пароля',
description='Отправляет шестизначный код на email текущего пользователя. Повторный запрос возможен после истечения TTL.',
responses=_PASSWORD_ERROR_RESPONSES,
)
@csrf_protect()
async def change_password_start(
request: Request,
auth: AuthContext = Depends(require_access_token),
command: ChangePasswordStartCommand = Depends(get_change_password_start_command),
):
result = await command(user_id=auth.user_id)
return ORJSONResponse(status_code=status.HTTP_200_OK, content={'success': result})
@account_settings_router.post(
path='/password/complete',
response_class=ORJSONResponse,
status_code=status.HTTP_200_OK,
summary='Подтвердить смену пароля',
description=(
'Принимает код из письма, новый пароль и его подтверждение. '
'Новый пароль должен отличаться от текущего и соответствовать политике сложности (минимум 12 символов).'
),
responses=_PASSWORD_ERROR_RESPONSES,
)
@csrf_protect()
async def change_password_complete(
request: Request,
body: ChangePasswordConfirmRequest,
auth: AuthContext = Depends(require_access_token),
command: ChangePasswordCompleteCommand = Depends(get_change_password_complete_command),
):
result = await command(
user_id=auth.user_id,
code=body.code,
new_password=body.new_password,
confirm_password=body.confirm_password,
)
return ORJSONResponse(status_code=status.HTTP_200_OK, content={'success': result})
@account_settings_router.post(
path='/password/forgot/start',
response_class=ORJSONResponse,
status_code=status.HTTP_200_OK,
summary='Запросить код для восстановления пароля',
description=(
'Принимает email. Если учётная запись существует, отправляет шестизначный код. '
'Ответ всегда успешный при валидном email (без раскрытия наличия аккаунта).'
),
responses=_FORGOT_PASSWORD_ERROR_RESPONSES,
)
@rate_limit(limit=5, window_seconds=300, scope='key', key_prefix='rl', key_builder=email_rl_key)
async def forgot_password_start(
request: Request,
body: ForgotPasswordStartRequest,
command: ForgotPasswordStartCommand = Depends(get_forgot_password_start_command),
):
result = await command(email=body.email)
return ORJSONResponse(status_code=status.HTTP_200_OK, content={'success': result})
@account_settings_router.post(
path='/password/forgot/complete',
response_class=ORJSONResponse,
status_code=status.HTTP_200_OK,
summary='Установить новый пароль по коду из письма',
description=(
'Принимает email, код из письма, новый пароль и подтверждение. '
'Пароль: минимум 12 символов, строчная и заглавная буква, цифра, спецсимвол, без пробелов.'
),
responses=_FORGOT_PASSWORD_ERROR_RESPONSES,
)
@rate_limit(limit=10, window_seconds=300, scope='key', key_prefix='rl', key_builder=email_rl_key)
async def forgot_password_complete(
request: Request,
body: ForgotPasswordCompleteRequest,
command: ForgotPasswordCompleteCommand = Depends(get_forgot_password_complete_command),
):
result = await command(
email=body.email,
code=body.code,
new_password=body.new_password,
confirm_password=body.confirm_password,
)
return ORJSONResponse(status_code=status.HTTP_200_OK, content={'success': result})
#
# @account_settings_router.post(path='/encrypted-mnemonic/start', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
# async def encrypted_mnemonic_start(
# request: Request,
# auth: AuthContext = Depends(require_access_token),
# command: SetEncryptedMnemonicStartCommand = Depends(get_set_encrypted_mnemonic_start_command),
# ):
# result = await command(user_id=auth.user_id)
# return {'success': result}
#
#
# @account_settings_router.post(path='/encrypted-mnemonic/complete', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
# async def encrypted_mnemonic_complete(
# request: Request,
# body: EncryptedMnemonicConfirmRequest,
# auth: AuthContext = Depends(require_access_token),
# command: SetEncryptedMnemonicCompleteCommand = Depends(get_set_encrypted_mnemonic_complete_command),
# ):
# user = await command(
# user_id=auth.user_id,
# code=body.code,
# encrypted_mnemonic=body.encrypted_mnemonic,
# )
# return {'encrypted_mnemonic': user.encrypted_mnemonic}
#
#
# @account_settings_router.post(path='/email/start', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
# async def change_email_start(
# request: Request,
# auth: AuthContext = Depends(require_access_token),
# command: ChangeEmailStartCommand = Depends(get_change_email_start_command),
# ):
# result = await command(user_id=auth.user_id)
# return {'success': result}
#
#
# @account_settings_router.post(path='/email/confirm-old', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
# async def change_email_confirm_old(
# request: Request,
# body: ChangeEmailConfirmOldRequest,
# auth: AuthContext = Depends(require_access_token),
# command: ChangeEmailConfirmOldCommand = Depends(get_change_email_confirm_old_command),
# ):
# result = await command(user_id=auth.user_id, code=body.code, new_email=body.new_email)
# return {'success': result}
#
#
# @account_settings_router.post(path='/email/complete', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
# async def change_email_complete(
# request: Request,
# body: ChangeEmailCompleteRequest,
# auth: AuthContext = Depends(require_access_token),
# command: ChangeEmailCompleteCommand = Depends(get_change_email_complete_command),
# ):
# result = await command(user_id=auth.user_id, code=body.code)
# return {'success': result}
#
#
# @account_settings_router.post(path='/bank/start', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
# async def bank_details_start(
# request: Request,
# auth: AuthContext = Depends(require_access_token),
# command: UpdateBankDetailsStartCommand = Depends(get_update_bank_details_start_command),
# ):
# result = await command(user_id=auth.user_id)
# return {'success': result}
#
#
# @account_settings_router.post(path='/bank/complete', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
# async def bank_details_complete(
# request: Request,
# body: BankConfirmRequest,
# auth: AuthContext = Depends(require_access_token),
# command: UpdateBankDetailsCompleteCommand = Depends(get_update_bank_details_complete_command),
# ):
# user = await command(
# user_id=auth.user_id,
# code=body.code,
# passport_data=body.passport_data,
# inn=body.inn,
# erc20=body.erc20,
# )
# return ORJSONResponse(
# status_code=status.HTTP_200_OK,
# content={
# 'passport_data': user.passport_data,
# 'inn': user.inn,
# 'erc20': user.erc20,
# },
# )

7
src/presentation/routing/deals.py
View File

@@ -1,7 +0,0 @@
from fastapi import APIRouter
deals_router = APIRouter(prefix='/deals')
@deals_router.get(path='')
async def deals():
pass

11
src/presentation/routing/devices.py
View File

@@ -1,11 +0,0 @@
from fastapi import APIRouter, Request
from fastapi.responses import ORJSONResponse
from starlette import status
devices_router = APIRouter(prefix='/devices')
@devices_router.get(path='/', response_class=ORJSONResponse, status_code=status.HTTP_200_OK)
async def devices(
request: Request,
):
pass

9
src/presentation/routing/purchase_requests.py
View File

@@ -1,4 +1,4 @@
from fastapi import APIRouter, Depends, Query, status
from fastapi import APIRouter, Depends, Query, Request, status
from src.application.commands import (
CreatePurchaseRequestCommand,
@@ -7,6 +7,7 @@ from src.application.commands import (
)
from src.application.domain.dto import AuthContext
from src.presentation.decorators.auth import require_access_token
from src.presentation.decorators.csrf import csrf_protect
from src.presentation.dependencies.commands import (
get_create_purchase_request_command,
get_get_purchase_request_command,
@@ -23,7 +24,9 @@ purchase_requests_router = APIRouter(prefix='/purchase-requests', tags=['purchas
@purchase_requests_router.post('', response_model=PurchaseRequestResponse, status_code=status.HTTP_201_CREATED)
@csrf_protect()
async def create_purchase_request(
request: Request,
body: CreatePurchaseRequestBody,
auth: AuthContext = Depends(require_access_token),
command: CreatePurchaseRequestCommand = Depends(get_create_purchase_request_command),
@@ -39,7 +42,9 @@ async def create_purchase_request(
@purchase_requests_router.get('', response_model=PurchaseRequestListResponse)
@csrf_protect()
async def list_purchase_requests(
request: Request,
status_filter: str | None = Query(default=None, alias='status'),
limit: int = Query(default=50, ge=1, le=200),
offset: int = Query(default=0, ge=0),
@@ -59,7 +64,9 @@ async def list_purchase_requests(
@purchase_requests_router.get('/{request_id}', response_model=PurchaseRequestResponse)
@csrf_protect()
async def get_purchase_request(
request: Request,
request_id: str,
auth: AuthContext = Depends(require_access_token),
command: GetPurchaseRequestCommand = Depends(get_get_purchase_request_command),

10
src/presentation/schemas/__init__.py
View File

@@ -1,10 +0,0 @@
from src.presentation.schemas.avatar import SetAvatarRequest
from src.presentation.schemas.phone import SetPhoneRequest
from src.presentation.schemas.bank import BankUpdateRequest, BankConfirmRequest
from src.presentation.schemas.encrypted_mnemonic import EncryptedMnemonicConfirmRequest
from src.presentation.schemas.password import (
ChangePasswordConfirmRequest,
ForgotPasswordStartRequest,
ForgotPasswordCompleteRequest,
)
from src.presentation.schemas.email import ChangeEmailConfirmOldRequest, ChangeEmailCompleteRequest

19
src/presentation/schemas/api_errors.py
View File

@@ -1,19 +0,0 @@
from __future__ import annotations
from pydantic import BaseModel, Field
class ApiErrorPayload(BaseModel):
detail: str = Field(description='Текстовое описание ошибки для клиента')
class ValidationErrorDetailItem(BaseModel):
loc: list[str | int] = Field(description='Путь к полю, вызвавшему ошибку')
msg: str = Field(description='Сообщение')
type: str = Field(description='Тип ошибки валидации')
class ApiValidationErrorsPayload(BaseModel):
detail: list[ValidationErrorDetailItem] = Field(
description='Список ошибок валидации тела или параметров запроса'
)

69
src/presentation/schemas/avatar.py
View File

@@ -1,69 +0,0 @@
from __future__ import annotations
from typing import Any
import base64
import binascii
from pydantic import BaseModel, Field, model_validator
AVATAR_MAX_BYTES = 10 * 1024 * 1024
_IMAGE_SIGNATURES = (
b'\xff\xd8\xff',
b'\x89PNG\r\n\x1a\n',
b'GIF87a',
b'GIF89a',
)
def _avatar_payload_to_bytes(photo_base64: str) -> bytes:
s = photo_base64.strip()
if not s:
raise ValueError('photo_base64 must not be empty')
if s.startswith('data:'):
parts = s.split(',', 1)
if len(parts) != 2:
raise ValueError('Invalid data URL')
s = parts[1].strip()
try:
data = base64.b64decode(s, validate=True)
except binascii.Error as exc:
raise ValueError('Invalid base64') from exc
if len(data) > AVATAR_MAX_BYTES:
raise ValueError(f'Photo must not exceed {AVATAR_MAX_BYTES} bytes')
if len(data) < 12:
raise ValueError('Photo data is too small')
ok = False
for sig in _IMAGE_SIGNATURES:
if data.startswith(sig):
ok = True
break
if not ok and data.startswith(b'RIFF') and len(data) > 12 and data[8:12] == b'WEBP':
ok = True
if not ok:
raise ValueError('Photo must be JPEG, PNG, GIF or WebP')
return data
class SetAvatarRequest(BaseModel):
photo_base64: str = Field(
...,
description='Изображение JPEG, PNG, GIF или WebP в Base64; допустим data URL (data:image/...;base64,...). Максимум 10 МБ после декодирования.',
)
decoded_bytes: bytes = Field(exclude=True)
@model_validator(mode='before')
@classmethod
def _decode_input(cls, data: Any):
if not isinstance(data, dict):
return data
raw = data.get('photo_base64')
if raw is None:
return data
if not isinstance(raw, str):
raise ValueError('photo_base64 must be a string')
stripped = raw.strip()
decoded = _avatar_payload_to_bytes(stripped)
return {'photo_base64': stripped, 'decoded_bytes': decoded}

116
src/presentation/schemas/bank.py
View File

@@ -1,116 +0,0 @@
import re
from typing import Self
from pydantic import BaseModel, field_validator, model_validator
class BankUpdateRequest(BaseModel):
passport_data: str | None = None
inn: str | None = None
erc20: str | None = None
@model_validator(mode='after')
def at_least_one(self) -> Self:
if not any([self.passport_data, self.inn, self.erc20]):
raise ValueError('At least one field is required')
return self
@field_validator('passport_data', 'inn', 'erc20')
@classmethod
def strip_optional(cls, v: str | None) -> str | None:
if v is None:
return None
s = v.strip()
return s or None
@field_validator('inn')
@classmethod
def validate_inn(cls, v: str | None) -> str | None:
if v is None:
return None
v = v.strip()
if not re.match(r'^\d{10}(\d{2})?$', v):
raise ValueError('INN must be 10 or 12 digits')
if len(v) > 12:
raise ValueError('INN is too long')
return v
@field_validator('erc20')
@classmethod
def validate_erc20(cls, v: str | None) -> str | None:
if v is None:
return None
v = v.strip()
if not re.match(r'^0x[a-fA-F0-9]{40}$', v):
raise ValueError('ERC20 address must be 0x followed by 40 hex characters')
return v
@field_validator('passport_data')
@classmethod
def validate_passport_data(cls, v: str | None) -> str | None:
if v is None:
return None
v = v.strip()
if len(v) > 255:
raise ValueError('Passport data is too long')
return v
class BankConfirmRequest(BaseModel):
code: str
passport_data: str | None = None
inn: str | None = None
erc20: str | None = None
@model_validator(mode='after')
def at_least_one_field(self) -> Self:
if not any([self.passport_data, self.inn, self.erc20]):
raise ValueError('At least one field is required')
return self
@field_validator('code')
@classmethod
def validate_code(cls, v: str) -> str:
v = v.strip()
if not re.match(r'^\d{6}$', v):
raise ValueError('Code must be exactly 6 digits')
return v
@field_validator('passport_data', 'inn', 'erc20')
@classmethod
def strip_optional(cls, v: str | None) -> str | None:
if v is None:
return None
s = v.strip()
return s or None
@field_validator('inn')
@classmethod
def validate_inn(cls, v: str | None) -> str | None:
if v is None:
return None
v = v.strip()
if not re.match(r'^\d{10}(\d{2})?$', v):
raise ValueError('INN must be 10 or 12 digits')
if len(v) > 12:
raise ValueError('INN is too long')
return v
@field_validator('erc20')
@classmethod
def validate_erc20(cls, v: str | None) -> str | None:
if v is None:
return None
v = v.strip()
if not re.match(r'^0x[a-fA-F0-9]{40}$', v):
raise ValueError('ERC20 address must be 0x followed by 40 hex characters')
return v
@field_validator('passport_data')
@classmethod
def validate_passport_data(cls, v: str | None) -> str | None:
if v is None:
return None
v = v.strip()
if len(v) > 255:
raise ValueError('Passport data is too long')
return v

35
src/presentation/schemas/email.py
View File

@@ -1,35 +0,0 @@
import re
from pydantic import BaseModel, field_validator
class ChangeEmailConfirmOldRequest(BaseModel):
code: str
new_email: str
@field_validator('code')
@classmethod
def validate_code(cls, v: str) -> str:
v = v.strip()
if not re.match(r'^\d{6}$', v):
raise ValueError('Code must be exactly 6 digits')
return v
@field_validator('new_email')
@classmethod
def validate_new_email(cls, v: str) -> str:
v = v.strip().lower()
if not re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', v):
raise ValueError('Invalid email address')
return v
class ChangeEmailCompleteRequest(BaseModel):
code: str
@field_validator('code')
@classmethod
def validate_code(cls, v: str) -> str:
v = v.strip()
if not re.match(r'^\d{6}$', v):
raise ValueError('Code must be exactly 6 digits')
return v

23
src/presentation/schemas/encrypted_mnemonic.py
View File

@@ -1,23 +0,0 @@
import re
from pydantic import BaseModel, field_validator
class EncryptedMnemonicConfirmRequest(BaseModel):
code: str
encrypted_mnemonic: str
@field_validator('code')
@classmethod
def validate_code(cls, v: str) -> str:
v = v.strip()
if not re.match(r'^\d{6}$', v):
raise ValueError('Code must be exactly 6 digits')
return v
@field_validator('encrypted_mnemonic')
@classmethod
def validate_encrypted_mnemonic(cls, v: str) -> str:
v = v.strip()
if not v:
raise ValueError('encrypted_mnemonic must not be empty')
return v

59
src/presentation/schemas/me_public.py
View File

@@ -1,59 +0,0 @@
from __future__ import annotations
from datetime import date, datetime
from pydantic import BaseModel, ConfigDict, Field
from src.application.domain.entities import UserEntity
class MeUserPublicResponse(BaseModel):
model_config = ConfigDict(from_attributes=False)
id: str | None = Field(None, description='Идентификатор пользователя')
email: str | None = Field(None, description='Email')
first_name: str | None = Field(None, description='Имя')
middle_name: str | None = Field(None, description='Отчество')
last_name: str | None = Field(None, description='Фамилия')
birth_date: date | None = Field(None, description='Дата рождения')
encrypted_mnemonic: str | None = Field(None, description='Шифрованная мнемоника')
phone: str | None = Field(None, description='Телефон')
passport_data: str | None = Field(None, description='Паспортные данные')
inn: str | None = Field(None, description='ИНН')
erc20: str | None = Field(None, description='ERC-20 адрес')
avatar_link: str | None = Field(None, description='HTTPS-ссылка на текущий аватар в хранилище')
kyc_verified: bool | None = Field(None, description='Признак пройденного KYC')
is_deleted: bool | None = Field(None, description='Удалён ли аккаунт')
created_at: datetime | None = Field(None, description='Время создания записи')
updated_at: datetime | None = Field(None, description='Время последнего обновления')
kyc_verified_at: datetime | None = Field(None, description='Время подтверждения KYC')
@classmethod
def from_user(cls, user: UserEntity) -> MeUserPublicResponse:
return cls(
id=user.id,
email=user.email,
first_name=user.first_name,
middle_name=user.middle_name,
last_name=user.last_name,
birth_date=user.birth_date,
encrypted_mnemonic=user.encrypted_mnemonic,
phone=user.phone,
passport_data=user.passport_data,
inn=user.inn,
erc20=user.erc20,
avatar_link=user.avatar_link,
kyc_verified=user.kyc_verified,
is_deleted=user.is_deleted,
created_at=user.created_at,
updated_at=user.updated_at,
kyc_verified_at=user.kyc_verified_at,
)
class SetAvatarPublicResponse(MeUserPublicResponse):
webp_size_bytes: int = Field(
...,
ge=0,
description='Размер сохранённого файла аватара в формате WebP, байты',
)

75
src/presentation/schemas/password.py
View File

@@ -1,75 +0,0 @@
import re
from typing import Self
from pydantic import BaseModel, field_validator, model_validator
from src.application.domain.password_policy import validate_password_strength
class ForgotPasswordStartRequest(BaseModel):
email: str
@field_validator('email')
@classmethod
def validate_email(cls, v: str) -> str:
v = v.strip().lower()
if not re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', v):
raise ValueError('Invalid email address')
return v
class ForgotPasswordCompleteRequest(BaseModel):
email: str
code: str
new_password: str
confirm_password: str
@field_validator('email')
@classmethod
def validate_email(cls, v: str) -> str:
v = v.strip().lower()
if not re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', v):
raise ValueError('Invalid email address')
return v
@model_validator(mode='after')
def passwords_match(self) -> Self:
if self.new_password != self.confirm_password:
raise ValueError('Passwords do not match')
return self
@field_validator('code')
@classmethod
def validate_code(cls, v: str) -> str:
v = v.strip()
if not re.match(r'^\d{6}$', v):
raise ValueError('Code must be exactly 6 digits')
return v
@field_validator('new_password')
@classmethod
def validate_new_password(cls, v: str) -> str:
return validate_password_strength(v)
class ChangePasswordConfirmRequest(BaseModel):
code: str
new_password: str
confirm_password: str
@model_validator(mode='after')
def passwords_match(self) -> Self:
if self.new_password != self.confirm_password:
raise ValueError('Passwords do not match')
return self
@field_validator('code')
@classmethod
def validate_code(cls, v: str) -> str:
v = v.strip()
if not re.match(r'^\d{6}$', v):
raise ValueError('Code must be exactly 6 digits')
return v
@field_validator('new_password')
@classmethod
def validate_new_password(cls, v: str) -> str:
return validate_password_strength(v)

16
src/presentation/schemas/phone.py
View File

@@ -1,16 +0,0 @@
import re
from pydantic import BaseModel, field_validator
class SetPhoneRequest(BaseModel):
phone: str
@field_validator('phone')
@classmethod
def validate_russian_phone(cls, v: str) -> str:
cleaned = re.sub(r'[\s\-\(\)]', '', v)
pattern = r'^(\+7|8)\d{10}$'
if not re.match(pattern, cleaned):
raise ValueError('Invalid Russian phone number')
normalized = '+7' + cleaned[-10:]
return normalized

1
src/presentation/serializers/__init__.py
View File

@@ -1 +0,0 @@
from src.presentation.serializers.me_user import me_user_payload, me_user_public

13
src/presentation/serializers/me_user.py
View File

@@ -1,13 +0,0 @@
from __future__ import annotations
from src.application.domain.entities import UserEntity
from src.presentation.schemas.me_public import MeUserPublicResponse
def me_user_public(user: UserEntity) -> MeUserPublicResponse:
return MeUserPublicResponse.from_user(user)
def me_user_payload(user: UserEntity) -> dict:
return me_user_public(user).model_dump(mode='json')