o2o-negosium-original/negodata/backend/crud/card_crud.py

from abc import ABC, abstractmethod
from typing import Optional, Tuple

from sqlalchemy import select, func, and_, or_, update
from sqlalchemy.ext.asyncio import AsyncSession

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import users
from common.enums import ErrorType
from common.logger import LOG
from common.utils.gtime import GTime


# 협상카드 CRUD. nego_cards/wild_cards 두 테이블에 공통으로 쓰는 제네릭 구현.
class ICardCRUD(ABC):
    @abstractmethod
    async def search(self, cdb: AsyncSession, model, user_id, search, skip, limit) -> Tuple[ErrorType, list, int]:
        pass

    @abstractmethod
    async def get_by_id(self, cdb: AsyncSession, model, pk_col, card_id) -> Tuple[ErrorType, object]:
        pass

    @abstractmethod
    async def add(self, cdb: AsyncSession, card) -> ErrorType:
        pass

    @abstractmethod
    async def update(self, cdb: AsyncSession, model, pk_col, card_id, data: dict) -> ErrorType:
        pass

    @abstractmethod
    async def soft_delete(self, cdb: AsyncSession, model, pk_col, card_id) -> ErrorType:
        pass

    @abstractmethod
    async def user_name_map(self, cdb: AsyncSession, user_ids) -> Tuple[ErrorType, dict]:
        pass


class CardCRUD(ICardCRUD):
    async def search(
        self, cdb: AsyncSession, model, user_id, search: Optional[str], skip: int, limit: int
    ) -> Tuple[ErrorType, list, int]:
        try:
            # 내 개인 카드 + 전체(공용, user_id NULL) 카드. 남의 개인 카드는 제외.
            conditions = [
                model.deleted == False,  # noqa: E712
                or_(model.user_id == user_id, model.user_id.is_(None)),
            ]
            if search:
                conditions.append(
                    or_(
                        model.name.ilike(f"%{search}%"),
                        model.number.ilike(f"%{search}%"),
                        model.script.ilike(f"%{search}%"),
                    )
                )
            where = and_(*conditions)

            cnt_err, cnt_rows = await DB_SESSION_MNG.execute(cdb, select(func.count()).select_from(model).where(where))
            if cnt_err != ErrorType.SUCCESS:
                return cnt_err, [], 0
            total = int(cnt_rows[0] or 0) if cnt_rows else 0

            list_err, rows = await DB_SESSION_MNG.execute(
                cdb,
                select(model).where(where).order_by(model.created_at.desc()).offset(skip).limit(limit),
            )
            if list_err != ErrorType.SUCCESS:
                return list_err, [], 0
            return ErrorType.SUCCESS, list(rows), total
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, [], 0

    async def get_by_id(self, cdb: AsyncSession, model, pk_col, card_id) -> Tuple[ErrorType, object]:
        try:
            query = select(model).where(pk_col == card_id, model.deleted == False).limit(1)  # noqa: E712
            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query)
            if err_type != ErrorType.SUCCESS:
                return err_type, None
            if len(row_list) != 1:
                return ErrorType.DB_INVALID_KEY, None
            return ErrorType.SUCCESS, row_list[0]
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, None

    async def add(self, cdb: AsyncSession, card) -> ErrorType:
        try:
            return await DB_SESSION_MNG.insert(cdb, card)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED

    async def update(self, cdb: AsyncSession, model, pk_col, card_id, data: dict) -> ErrorType:
        try:
            if not data:
                return ErrorType.SUCCESS
            query = update(model).where(pk_col == card_id).values(**data)
            return await DB_SESSION_MNG.add(cdb, query)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED

    async def soft_delete(self, cdb: AsyncSession, model, pk_col, card_id) -> ErrorType:
        try:
            query = update(model).where(pk_col == card_id).values(deleted=True, updated_at=GTime.UTC())
            return await DB_SESSION_MNG.add(cdb, query)
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED

    async def user_name_map(self, cdb: AsyncSession, user_ids) -> Tuple[ErrorType, dict]:
        """user_id 목록 → {user_id: name}. 카드 목록 '작성자(등록자)' 표기용(company.users 조인).
        공용(user_id NULL) 카드는 호출 전에 걸러 넘긴다 → 맵에 없으면 작성자 없음."""
        try:
            if not user_ids:
                return ErrorType.SUCCESS, {}
            query = select(users.user_id, users.name).where(users.user_id.in_(user_ids))
            err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
            if err_type != ErrorType.SUCCESS:
                return err_type, {}
            return ErrorType.SUCCESS, {uid: name for uid, name in rows}
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, {}