feat: add search

src/infrastructure/database/repositories/user_repository.py

@@ -3,13 +3,14 @@ from __future__ import annotations
 from datetime import datetime
 
 from fastapi import status
-from sqlalchemy import select
+from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.exc import IntegrityError, SQLAlchemyError
 
 from src.application.abstractions.repositories import IUserRepository
 from src.application.contracts import ILogger
 from src.application.domain.entities import UserEntity
+from src.application.domain.entities.organization import PartySearchEntity
 from src.application.domain.enums.account_type import AccountType
 from src.application.domain.exceptions import ApplicationException
 from src.infrastructure.database.models import UserModel
@@ -92,3 +93,69 @@ class UserRepository(IUserRepository):
         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
+
+    def _individual_search_filter(self, query: str):
+        pattern = f'%{query.strip()}%'
+        full_name = func.concat_ws(
+            ' ',
+            UserModel.last_name,
+            UserModel.first_name,
+            UserModel.middle_name,
+        )
+        return or_(
+            UserModel.id.ilike(pattern),
+            UserModel.email.ilike(pattern),
+            UserModel.last_name.ilike(pattern),
+            UserModel.first_name.ilike(pattern),
+            UserModel.middle_name.ilike(pattern),
+            UserModel.phone.ilike(pattern),
+            UserModel.passport_data.ilike(pattern),
+            UserModel.inn.ilike(pattern),
+            UserModel.erc20.ilike(pattern),
+            full_name.ilike(pattern),
+        )
+
+    def _to_search_entity(self, user: UserModel) -> PartySearchEntity:
+        name = ' '.join(
+            part for part in (user.last_name, user.first_name, user.middle_name)
+            if part
+        ) or None
+        return PartySearchEntity(
+            id=user.id,
+            account_type=user.account_type,
+            user_id=user.id,
+            email=user.email,
+            name=name,
+            inn=user.inn,
+            phone=user.phone,
+            status=None,
+            kyc_verified=user.kyc_verified,
+            created_at=user.created_at,
+        )
+
+    async def search_individuals(self, *, query: str, limit: int, offset: int) -> list[PartySearchEntity]:
+        stmt = (
+            select(UserModel)
+            .where(
+                UserModel.is_deleted.is_(False),
+                UserModel.account_type == AccountType.INDIVIDUAL,
+                self._individual_search_filter(query),
+            )
+            .order_by(UserModel.created_at.desc())
+            .limit(limit)
+            .offset(offset)
+        )
+        result = await self._session.execute(stmt)
+        return [self._to_search_entity(user) for user in result.scalars().all()]
+
+    async def count_individuals(self, *, query: str) -> int:
+        stmt = (
+            select(func.count(UserModel.id))
+            .where(
+                UserModel.is_deleted.is_(False),
+                UserModel.account_type == AccountType.INDIVIDUAL,
+                self._individual_search_filter(query),
+            )
+        )
+        result = await self._session.execute(stmt)
+        return int(result.scalar_one())