diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..9f6ecb4 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,56 @@ +services: + pay: + container_name: pay-service + build: + context: . + dockerfile: Dockerfile + ports: + - "8000:8000" + environment: + PYTHONUNBUFFERED: "1" + APP_MODULE: "src.main:app" + APP_HOST: "0.0.0.0" + APP_PORT: "8000" + APP_WORKERS: "1" + env_file: + - .env + depends_on: + keydb: + condition: service_healthy + restart: no + + keydb: + image: eqalpha/keydb + container_name: keydb + restart: no + expose: + - "6379" + volumes: + - keydb_data:/data + command: + - keydb-server + - --requirepass + - keydb + - --dir + - /data + - --appendonly + - "yes" + - --appendfsync + - everysec + - --save + - "900" + - "1" + - --save + - "300" + - "10" + - --save + - "60" + - "10000" + healthcheck: + test: [ "CMD", "redis-cli", "-a", "keydb", "ping" ] + interval: 5s + timeout: 2s + retries: 20 + +volumes: + keydb_data: \ No newline at end of file diff --git a/src/application/abstractions/__init__.py b/src/application/abstractions/__init__.py new file mode 100644 index 0000000..76a0e8d --- /dev/null +++ b/src/application/abstractions/__init__.py @@ -0,0 +1 @@ +from src.application.abstractions.i_unit_of_work import IUnitOfWork \ No newline at end of file diff --git a/src/application/abstractions/i_unit_of_work.py b/src/application/abstractions/i_unit_of_work.py new file mode 100644 index 0000000..6a706bc --- /dev/null +++ b/src/application/abstractions/i_unit_of_work.py @@ -0,0 +1,19 @@ +from __future__ import annotations +from typing import Protocol, runtime_checkable +from src.application.abstractions.repositories import IUserRepository, ISessionRepository + + +@runtime_checkable +class IUnitOfWork(Protocol): + async def __aenter__(self) -> "IUnitOfWork": ... + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ... + + async def commit(self) -> None: ... + async def rollback(self) -> None: ... + + @property + def user_repository(self) -> IUserRepository: ... + + @property + def session_repository(self) -> ISessionRepository: ... + diff --git a/src/application/abstractions/repositories/__init__.py b/src/application/abstractions/repositories/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/application/command/create_order_command.py b/src/application/command/create_order_command.py index 4d22068..f227fe6 100644 --- a/src/application/command/create_order_command.py +++ b/src/application/command/create_order_command.py @@ -5,17 +5,8 @@ from src.presentation.schemas.order import CreateOrder class UserLoginStartCommand: 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 + pass @transactional diff --git a/src/infrastructure/database/repositories/__init__.py b/src/infrastructure/database/repositories/__init__.py index b6481fd..e69de29 100644 --- a/src/infrastructure/database/repositories/__init__.py +++ b/src/infrastructure/database/repositories/__init__.py @@ -1 +0,0 @@ -from src.infrastructure.database.repositories.user_repository import UserRepository \ No newline at end of file diff --git a/src/infrastructure/database/repositories/user_repository.py b/src/infrastructure/database/repositories/user_repository.py deleted file mode 100644 index e340e5f..0000000 --- a/src/infrastructure/database/repositories/user_repository.py +++ /dev/null @@ -1,114 +0,0 @@ -from __future__ import annotations -from fastapi import status -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.exc import IntegrityError, SQLAlchemyError -from src.application.contracts import ILogger -from src.application.domain.exceptions import ApplicationException -from src.application.abstractions.repositories import IUserRepository -from src.application.domain.entities import UserEntity -from src.infrastructure.database.models import UserModel - - -class UserRepository(IUserRepository): - def __init__(self, session: AsyncSession, logger: ILogger): - self._session = session - self._logger = logger - - async def create_user(self, email: str, password_hash: str) -> UserEntity: - user = UserModel(email=email, password_hash=password_hash) - self._session.add(user) - try: - await self._session.flush() - return UserEntity( - id=user.id, - email=user.email, - created_at=user.created_at, - kyc_verified=user.kyc_verified, - is_deleted=user.is_deleted - ) - - except IntegrityError: - self._logger.error(f'User already exists with email {user.email}') - raise ApplicationException( - status_code=status.HTTP_409_CONFLICT, - message='User with this email already exists', - ) - - except SQLAlchemyError as exception: - self._logger.exception(str(exception)) - raise ApplicationException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message=f'Database error: {str(exception)}', - ) - - async def get_user_by_email(self, email: str) -> UserEntity: - try: - stmt = ( - select(UserModel) - .where( - UserModel.email == email, - UserModel.is_deleted.is_(False), - ) - ) - - result = await self._session.execute(stmt) - user: UserModel | None = result.scalar_one_or_none() - - if user is None: - self._logger.warning(f'User not found with email {email}') - raise ApplicationException(status_code=status.HTTP_404_NOT_FOUND, message='User not found',) - - return UserEntity( - id=user.id, - email=user.email, - password_hash=user.password_hash, - first_name=user.first_name, - middle_name=user.middle_name, - last_name=user.last_name, - birth_date=user.birth_date, - crypto_wallet=user.crypto_wallet, - phone=user.phone, - bik=user.bik, - account_number=user.account_number, - card_number=user.card_number, - inn=user.inn, - kyc_verified_at=user.kyc_verified_at, - kyc_verified=user.kyc_verified, - is_deleted=user.is_deleted, - created_at=user.created_at, - updated_at=user.updated_at, - ) - - except ApplicationException: - raise - - except SQLAlchemyError as exception: - self._logger.exception(str(exception)) - raise ApplicationException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message=f'Database error: {str(exception)}', - ) - - async def exists_by_email(self, email: str) -> bool: - try: - stmt = ( - select(UserModel.id) - .where( - UserModel.email == email, - UserModel.is_deleted.is_(False), - ) - .limit(1) - ) - - result = await self._session.execute(stmt) - return result.scalar_one_or_none() is not None - - except SQLAlchemyError as exception: - self._logger.exception(str(exception)) - raise ApplicationException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - message=f'Database error: {str(exception)}', - ) - - diff --git a/src/infrastructure/database/unit_of_work.py b/src/infrastructure/database/unit_of_work.py index 3e2363c..c3429fc 100644 --- a/src/infrastructure/database/unit_of_work.py +++ b/src/infrastructure/database/unit_of_work.py @@ -1,8 +1,8 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.application.abstractions import IUnitOfWork -from src.application.abstractions.repositories import IUserRepository, ISessionRepository from src.application.contracts import ILogger -from src.infrastructure.database.repositories import UserRepository, SessionRepository +# from src.application.abstractions.repositories import IUserRepository, ISessionRepository +# from src.infrastructure.database.repositories import UserRepository, SessionRepository @@ -10,8 +10,8 @@ class UnitOfWork(IUnitOfWork): def __init__(self, session_factory: async_sessionmaker[AsyncSession], logger: ILogger): self.session_factory = session_factory self._session: AsyncSession = None - self._user_repository: IUserRepository = None - self._session_repository: ISessionRepository = None + # self._user_repository: IUserRepository = None + # self._session_repository: ISessionRepository = None self._logger: ILogger = logger async def __aenter__(self): @@ -29,14 +29,14 @@ class UnitOfWork(IUnitOfWork): self._logger.debug('Commit') await self._session.close() - @property - def user_repository(self) -> IUserRepository: - if self._user_repository is None: - self._user_repository = UserRepository(session=self._session, logger=self._logger) - return self._user_repository - - @property - def session_repository(self) -> ISessionRepository: - if self._session_repository is None: - self._session_repository = SessionRepository(session=self._session, logger=self._logger) - return self._session_repository + # @property + # def user_repository(self) -> IUserRepository: + # if self._user_repository is None: + # self._user_repository = UserRepository(session=self._session, logger=self._logger) + # return self._user_repository + # + # @property + # def session_repository(self) -> ISessionRepository: + # if self._session_repository is None: + # self._session_repository = SessionRepository(session=self._session, logger=self._logger) + # return self._session_repository diff --git a/src/infrastructure/messanger/__init__.py b/src/infrastructure/messanger/__init__.py new file mode 100644 index 0000000..0369f8a --- /dev/null +++ b/src/infrastructure/messanger/__init__.py @@ -0,0 +1 @@ +from src.infrastructure.messanger.rabbit_client import RabbitClient \ No newline at end of file diff --git a/src/infrastructure/messanger/rabbit_client.py b/src/infrastructure/messanger/rabbit_client.py new file mode 100644 index 0000000..d18db3f --- /dev/null +++ b/src/infrastructure/messanger/rabbit_client.py @@ -0,0 +1,72 @@ +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, + ) diff --git a/src/presentation/dependencies/__init__.py b/src/presentation/dependencies/__init__.py index 615c62b..af0fa45 100644 --- a/src/presentation/dependencies/__init__.py +++ b/src/presentation/dependencies/__init__.py @@ -1,16 +1,16 @@ -from src.presentation.dependencies.commands import ( - get_get_me_command, - get_set_phone_command, - get_set_crypto_wallet_start_command, - get_set_crypto_wallet_complete_command, - get_update_bank_details_start_command, - get_update_bank_details_complete_command, - get_change_password_start_command, - get_change_password_complete_command, - get_change_email_start_command, - get_change_email_confirm_old_command, - get_change_email_complete_command, -) +# from src.presentation.dependencies.commands import ( +# get_get_me_command, +# get_set_phone_command, +# get_set_crypto_wallet_start_command, +# get_set_crypto_wallet_complete_command, +# get_update_bank_details_start_command, +# get_update_bank_details_complete_command, +# get_change_password_start_command, +# get_change_password_complete_command, +# get_change_email_start_command, +# get_change_email_confirm_old_command, +# get_change_email_complete_command, +# ) from src.presentation.dependencies.security import get_jwt_service from src.presentation.dependencies.cache import get_redis, get_cache from src.presentation.dependencies.queue_messanger import get_rabbit diff --git a/src/presentation/dependencies/commands.py b/src/presentation/dependencies/commands.py index 1c139b1..25311fc 100644 --- a/src/presentation/dependencies/commands.py +++ b/src/presentation/dependencies/commands.py @@ -1,161 +1,17 @@ -from fastapi import Depends -from src.application.abstractions import IUnitOfWork -from src.application.commands import GetMeCommand, SetPhoneCommand, SetCryptoWalletStartCommand, SetCryptoWalletCompleteCommand, UpdateBankDetailsStartCommand, UpdateBankDetailsCompleteCommand, ChangePasswordStartCommand, ChangePasswordCompleteCommand, ChangeEmailStartCommand, ChangeEmailConfirmOldCommand, ChangeEmailCompleteCommand -from src.application.contracts import ILogger, ICache, IQueueMessanger, IHashService -from src.presentation.dependencies.cache import get_cache -from src.presentation.dependencies.logger import get_logger -from src.presentation.dependencies.queue_messanger import get_rabbit -from src.presentation.dependencies.security import get_hash_service -from src.presentation.dependencies.unit_of_work import get_unit_of_work +# from fastapi import Depends +# from src.application.abstractions import IUnitOfWork +# from src.application.commands import GetMeCommand, SetPhoneCommand, SetCryptoWalletStartCommand, SetCryptoWalletCompleteCommand, UpdateBankDetailsStartCommand, UpdateBankDetailsCompleteCommand, ChangePasswordStartCommand, ChangePasswordCompleteCommand, ChangeEmailStartCommand, ChangeEmailConfirmOldCommand, ChangeEmailCompleteCommand +# from src.application.contracts import ILogger, ICache, IQueueMessanger, IHashService +# from src.presentation.dependencies.cache import get_cache +# from src.presentation.dependencies.logger import get_logger +# from src.presentation.dependencies.queue_messanger import get_rabbit +# from src.presentation.dependencies.security import get_hash_service +# from src.presentation.dependencies.unit_of_work import get_unit_of_work -def get_get_me_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), -) -> GetMeCommand: - return GetMeCommand(logger=logger, unit_of_work=unit_of_work, cache=cache) - - -def get_set_phone_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), -) -> SetPhoneCommand: - return SetPhoneCommand(logger=logger, unit_of_work=unit_of_work, cache=cache) - - -def get_set_crypto_wallet_start_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - messanger: IQueueMessanger = Depends(get_rabbit), - hash_service: IHashService = Depends(get_hash_service), -) -> SetCryptoWalletStartCommand: - return SetCryptoWalletStartCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - messanger=messanger, - hash_service=hash_service, - ) - - -def get_set_crypto_wallet_complete_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - hash_service: IHashService = Depends(get_hash_service), -) -> SetCryptoWalletCompleteCommand: - return SetCryptoWalletCompleteCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - hash_service=hash_service, - ) - - -def get_change_password_start_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - messanger: IQueueMessanger = Depends(get_rabbit), - hash_service: IHashService = Depends(get_hash_service), -) -> ChangePasswordStartCommand: - return ChangePasswordStartCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - messanger=messanger, - hash_service=hash_service, - ) - - -def get_change_password_complete_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - hash_service: IHashService = Depends(get_hash_service), -) -> ChangePasswordCompleteCommand: - return ChangePasswordCompleteCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - hash_service=hash_service, - ) - - -def get_change_email_start_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - messanger: IQueueMessanger = Depends(get_rabbit), - hash_service: IHashService = Depends(get_hash_service), -) -> ChangeEmailStartCommand: - return ChangeEmailStartCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - messanger=messanger, - hash_service=hash_service, - ) - - -def get_change_email_confirm_old_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - messanger: IQueueMessanger = Depends(get_rabbit), - hash_service: IHashService = Depends(get_hash_service), -) -> ChangeEmailConfirmOldCommand: - return ChangeEmailConfirmOldCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - messanger=messanger, - hash_service=hash_service, - ) - - -def get_change_email_complete_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - hash_service: IHashService = Depends(get_hash_service), -) -> ChangeEmailCompleteCommand: - return ChangeEmailCompleteCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - hash_service=hash_service, - ) - - -def get_update_bank_details_start_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - messanger: IQueueMessanger = Depends(get_rabbit), - hash_service: IHashService = Depends(get_hash_service), -) -> UpdateBankDetailsStartCommand: - return UpdateBankDetailsStartCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - messanger=messanger, - hash_service=hash_service, - ) - - -def get_update_bank_details_complete_command( - logger: ILogger = Depends(get_logger), - unit_of_work: IUnitOfWork = Depends(get_unit_of_work), - cache: ICache = Depends(get_cache), - hash_service: IHashService = Depends(get_hash_service), -) -> UpdateBankDetailsCompleteCommand: - return UpdateBankDetailsCompleteCommand( - logger=logger, - unit_of_work=unit_of_work, - cache=cache, - hash_service=hash_service, - ) +# def get_get_me_command( +# logger: ILogger = Depends(get_logger), +# unit_of_work: IUnitOfWork = Depends(get_unit_of_work), +# cache: ICache = Depends(get_cache), +# ) -> GetMeCommand: +# return GetMeCommand(logger=logger, unit_of_work=unit_of_work, cache=cache)