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, company_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

    @abstractmethod
    async def owner_company_id(self, cdb: AsyncSession, owner_user_id) -> Tuple[ErrorType, Optional[object]]:
        pass


class CardCRUD(ICardCRUD):
    async def search(
        self, cdb: AsyncSession, model, company_id, search: Optional[str], skip: int, limit: int
    ) -> Tuple[ErrorType, list, int]:
        try:
            # 회사 카드(같은 회사 유저 등록) + 전체(공용, user_id NULL) 카드. 타 회사 카드는 제외.
            company_users = select(users.user_id).where(users.company_id == company_id)
            conditions = [
                model.deleted == False,  # noqa: E712
                or_(model.user_id.in_(company_users), 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, {}

    async def owner_company_id(self, cdb: AsyncSession, owner_user_id) -> Tuple[ErrorType, Optional[object]]:
        """카드 소유자(user_id) → 소속 company_id. 개인 카드의 회사 스코프 접근 판정용."""
        try:
            query = select(users.company_id).where(users.user_id == owner_user_id).limit(1)
            err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
            if err_type != ErrorType.SUCCESS:
                return err_type, None
            return ErrorType.SUCCESS, rows[0] if rows else None
        except Exception as ex:
            LOG.e_no_callstack(ex)
            return ErrorType.DB_RUN_FAILED, None