minaChoi/o2o-negosium-original
1
0
Fork 0
Code Pull Requests Activity

[feat] negodata/backend: 상품·협력사 목록 서버 페이지네이션 + 검색/필터(카테고리·priority) + 상품 카테고리 distinct 엔드포인트

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mina Choi 2026-06-18 11:00:15 +09:00
parent 4587be5844
commit 1fef405e88
8 changed files with 269 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
negodata/backend/crud/item_crud.py
View File

@ -17,6 +17,18 @@ class IItemCRUD(ABC):
async def search(self, cdb: AsyncSession, company_id, search, category, skip, limit) -> Tuple[ErrorType, list, int]:
pass
@abstractmethod
async def distinct_categories(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, list]:
pass
@abstractmethod
async def code_exists(self, cdb: AsyncSession, company_id, code) -> Tuple[ErrorType, bool]:
pass
@abstractmethod
async def existing_codes(self, cdb: AsyncSession, company_id, codes: list) -> Tuple[ErrorType, list]:
pass
@abstractmethod
async def get_by_id(self, cdb: AsyncSession, item_id) -> Tuple[ErrorType, items]:
pass
@ -62,6 +74,61 @@ class ItemCRUD(IItemCRUD):
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED, [], 0
# 회사의 모든 상품에서 카테고리를 "이름 단위"로 중복 제거해 (이름, 대표 타입) 목록을 돌려준다.
async def distinct_categories(self, cdb: AsyncSession, company_id) -> Tuple[ErrorType, list]:
try:
query = (
select(items.category, func.min(items.category_type))
.where(
items.company_id == company_id,
items.deleted == False, # noqa: E712
items.category.isnot(None),
items.category != "",
)
.group_by(items.category)
.order_by(items.category)
)
err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
if err_type != ErrorType.SUCCESS:
return err_type, []
return ErrorType.SUCCESS, list(rows)
except Exception as ex:
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED, []
async def code_exists(self, cdb: AsyncSession, company_id, code) -> Tuple[ErrorType, bool]:
try:
query = select(func.count()).select_from(items).where(
items.company_id == company_id,
items.code == code,
items.deleted == False, # noqa: E712
)
err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
if err_type != ErrorType.SUCCESS:
return err_type, False
cnt = int(rows[0] or 0) if rows else 0
return ErrorType.SUCCESS, cnt > 0
except Exception as ex:
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED, False
async def existing_codes(self, cdb: AsyncSession, company_id, codes: list) -> Tuple[ErrorType, list]:
try:
if not codes:
return ErrorType.SUCCESS, []
query = select(items.code).where(
items.company_id == company_id,
items.code.in_(codes),
items.deleted == False, # noqa: E712
)
err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
if err_type != ErrorType.SUCCESS:
return err_type, []
return ErrorType.SUCCESS, [c for c in rows if c is not None]
except Exception as ex:
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED, []
async def get_by_id(self, cdb: AsyncSession, item_id) -> Tuple[ErrorType, items]:
try:
query = select(items).where(items.item_id == item_id, items.deleted == False).limit(1) # noqa: E712

47
negodata/backend/crud/supplier_crud.py
View File

@ -14,7 +14,15 @@ from common.utils.gtime import GTime
# 협력사 CRUD. 모든 조회/변경은 company_id 로 스코프된다(멀티테넌트).
class ISupplierCRUD(ABC):
@abstractmethod
async def search(self, cdb: AsyncSession, company_id, search, skip, limit) -> Tuple[ErrorType, list, int]:
async def search(self, cdb: AsyncSession, company_id, search, priority, skip, limit) -> Tuple[ErrorType, list, int]:
pass
@abstractmethod
async def code_exists(self, cdb: AsyncSession, company_id, code) -> Tuple[ErrorType, bool]:
pass
@abstractmethod
async def existing_codes(self, cdb: AsyncSession, company_id, codes: list) -> Tuple[ErrorType, list]:
pass
@abstractmethod
@ -36,7 +44,7 @@ class ISupplierCRUD(ABC):
class SupplierCRUD(ISupplierCRUD):
async def search(
self, cdb: AsyncSession, company_id, search: Optional[str], skip: int, limit: int
self, cdb: AsyncSession, company_id, search: Optional[str], priority: Optional[str], skip: int, limit: int
) -> Tuple[ErrorType, list, int]:
try:
conditions = [suppliers.deleted == False, suppliers.company_id == company_id] # noqa: E712
@ -48,6 +56,8 @@ class SupplierCRUD(ISupplierCRUD):
suppliers.manager_name.ilike(f"%{search}%"),
)
)
if priority:
conditions.append(suppliers.priority == priority)
where = and_(*conditions)
cnt_err, cnt_rows = await DB_SESSION_MNG.execute(cdb, select(func.count()).select_from(suppliers).where(where))
@ -66,6 +76,39 @@ class SupplierCRUD(ISupplierCRUD):
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED, [], 0
async def code_exists(self, cdb: AsyncSession, company_id, code) -> Tuple[ErrorType, bool]:
try:
query = select(func.count()).select_from(suppliers).where(
suppliers.company_id == company_id,
suppliers.code == code,
suppliers.deleted == False, # noqa: E712
)
err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
if err_type != ErrorType.SUCCESS:
return err_type, False
cnt = int(rows[0] or 0) if rows else 0
return ErrorType.SUCCESS, cnt > 0
except Exception as ex:
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED, False
async def existing_codes(self, cdb: AsyncSession, company_id, codes: list) -> Tuple[ErrorType, list]:
try:
if not codes:
return ErrorType.SUCCESS, []
query = select(suppliers.code).where(
suppliers.company_id == company_id,
suppliers.code.in_(codes),
suppliers.deleted == False, # noqa: E712
)
err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
if err_type != ErrorType.SUCCESS:
return err_type, []
return ErrorType.SUCCESS, [c for c in rows if c is not None]
except Exception as ex:
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED, []
async def get_by_id(self, cdb: AsyncSession, supplier_id) -> Tuple[ErrorType, suppliers]:
try:
query = select(suppliers).where(suppliers.supplier_id == supplier_id, suppliers.deleted == False).limit(1) # noqa: E712

20
negodata/backend/router/v1/item/item.py
View File

@ -2,15 +2,18 @@ from uuid import UUID
from fastapi import APIRouter, Depends, File, Query, UploadFile
from common.models.gmodel import UserInfo
from common.models.gmodel import PageParams, UserInfo
from router.v1.validator.dependencies import IsValidAccessToken, RemoveNoneResponse
from services.item_service import ItemService
from .protocol import (
Req_CheckCodes,
Req_CreateItem,
Req_UpdateItem,
Res_CheckCodes,
Res_DeleteItem,
Res_ExcelUpload,
Res_Item,
Res_ItemCategories,
Res_ItemList,
Res_LowestPriceResult,
Res_LowestPriceTrigger,
@ -26,10 +29,14 @@ async def list_items(
user_info: UserInfo = Depends(IsValidAccessToken),
search: str | None = Query(None, description="상품명/상품코드 검색"),
category: str | None = Query(None, description="카테고리 필터"),
page: int = Query(1, ge=1),
size: int = Query(20, ge=1, le=100),
pg: PageParams = Depends(),
):
return RemoveNoneResponse(await service.list_items(user_info.company_id, search, category, page, size))
return RemoveNoneResponse(await service.list_items(user_info.company_id, search, category, pg))
@router.get(path="/categories", response_model=Res_ItemCategories, summary="상품 카테고리 목록(distinct)")
async def list_item_categories(service: ItemService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
return RemoveNoneResponse(await service.list_categories(user_info.company_id))
@router.post(path="/create", response_model=Res_Item, summary="상품 등록")
@ -39,6 +46,11 @@ async def create_item(req: Req_CreateItem, service: ItemService = Depends(), use
)
@router.post(path="/check-codes", response_model=Res_CheckCodes, summary="코드 중복 사전검사(업로드 즉시)")
async def check_item_codes(req: Req_CheckCodes, service: ItemService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
return RemoveNoneResponse(await service.check_codes(user_info.company_id, req.codes))
@router.post(path="/upload-excel", response_model=Res_ExcelUpload, summary="엑셀 일괄 등록(스텁)")
async def upload_items_excel(
service: ItemService = Depends(),

24
negodata/backend/router/v1/item/protocol.py
View File

@ -4,7 +4,7 @@ from typing import Optional
from pydantic import ConfigDict
from common.models.gmodel import Res_WebPacketProtocol, WebPacketProtocol
from common.models.gmodel import Res_PageProtocol, Res_WebPacketProtocol, WebPacketProtocol
class ItemProtocol(WebPacketProtocol):
@ -82,11 +82,25 @@ class Res_Item(Res_WebPacketProtocol):
item: Optional[ItemData] = None
class Res_ItemList(Res_WebPacketProtocol):
class Res_ItemList(Res_PageProtocol):
items: list[ItemData] = []
total: int = 0
page: int = 0
size: int = 0
class ItemCategory(WebPacketProtocol):
name: str
category_type: int = 1
class Res_ItemCategories(Res_WebPacketProtocol):
categories: list[ItemCategory] = []
class Req_CheckCodes(ItemProtocol):
codes: list[str] = []
class Res_CheckCodes(Res_WebPacketProtocol):
existing: list[str] = [] # codes 중 이미 DB(같은 회사)에 존재하는 코드들
class Res_DeleteItem(Res_WebPacketProtocol):

15
negodata/backend/router/v1/supplier/protocol.py
View File

@ -4,7 +4,7 @@ from typing import Optional
from pydantic import ConfigDict
from common.models.gmodel import Res_WebPacketProtocol, WebPacketProtocol
from common.models.gmodel import Res_PageProtocol, Res_WebPacketProtocol, WebPacketProtocol
class SupplierProtocol(WebPacketProtocol):
@ -49,11 +49,16 @@ class Res_Supplier(Res_WebPacketProtocol):
supplier: Optional[SupplierData] = None
class Res_SupplierList(Res_WebPacketProtocol):
class Res_SupplierList(Res_PageProtocol):
suppliers: list[SupplierData] = []
total: int = 0
page: int = 0
size: int = 0
class Req_CheckCodes(SupplierProtocol):
codes: list[str] = []
class Res_CheckCodes(Res_WebPacketProtocol):
existing: list[str] = [] # codes 중 이미 DB(같은 회사)에 존재하는 코드들
class Res_DeleteSupplier(Res_WebPacketProtocol):

17
negodata/backend/router/v1/supplier/supplier.py
View File

@ -2,12 +2,14 @@ from uuid import UUID
from fastapi import APIRouter, Depends, File, Query, UploadFile
from common.models.gmodel import UserInfo
from common.models.gmodel import PageParams, UserInfo
from router.v1.validator.dependencies import IsValidAccessToken, RemoveNoneResponse
from services.supplier_service import SupplierService
from .protocol import (
Req_CheckCodes,
Req_CreateSupplier,
Req_UpdateSupplier,
Res_CheckCodes,
Res_DeleteSupplier,
Res_ExcelUpload,
Res_Supplier,
@ -23,10 +25,10 @@ async def list_suppliers(
service: SupplierService = Depends(),
user_info: UserInfo = Depends(IsValidAccessToken),
search: str | None = Query(None, description="협력사명/코드/담당자명 검색"),
page: int = Query(1, ge=1),
size: int = Query(20, ge=1, le=100),
priority: str | None = Query(None, description="우선순위 필터(HIGH/MEDIUM/LOW)"),
pg: PageParams = Depends(),
):
return RemoveNoneResponse(await service.list_suppliers(user_info.company_id, search, page, size))
return RemoveNoneResponse(await service.list_suppliers(user_info.company_id, search, priority, pg))
@router.post(path="/create", response_model=Res_Supplier, summary="협력사 등록")
@ -38,6 +40,13 @@ async def create_supplier(
)
@router.post(path="/check-codes", response_model=Res_CheckCodes, summary="코드 중복 사전검사(업로드 즉시)")
async def check_supplier_codes(
req: Req_CheckCodes, service: SupplierService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)
):
return RemoveNoneResponse(await service.check_codes(user_info.company_id, req.codes))
@router.post(path="/upload-excel", response_model=Res_ExcelUpload, summary="협력사 엑셀 일괄 등록(스텁)")
async def upload_suppliers_excel(
service: SupplierService = Depends(),

67
negodata/backend/services/item_service.py
View File

@ -6,8 +6,17 @@ from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import items
from common.enums import DBWRType, ErrorType
from common.logger import LOG
from common.models.gmodel import PageParams
from crud.item_crud import IItemCRUD, ItemCRUD
from router.v1.item.protocol import ItemData, Res_DeleteItem, Res_Item, Res_ItemList
from router.v1.item.protocol import (
ItemCategory,
ItemData,
Res_CheckCodes,
Res_DeleteItem,
Res_Item,
Res_ItemCategories,
Res_ItemList,
)
class ItemService:
@ -29,15 +38,14 @@ class ItemService:
return ErrorType.ITEM_NOT_FOUND, None
return ErrorType.SUCCESS, item
async def list_items(self, company_id: str, search, category, page: int, size: int) -> Res_ItemList:
res = Res_ItemList(page=page, size=size)
async def list_items(self, company_id: str, search, category, pg: PageParams) -> Res_ItemList:
res = Res_ItemList(page=pg.page, size=pg.size)
company_uuid = uuid.UUID(company_id)
skip = (page - 1) * size
err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
items.DBType(),
DBWRType.DB_READ.value,
lambda s: self.item_crud.search(s, company_uuid, search, category, skip, size),
lambda s: self.item_crud.search(s, company_uuid, search, category, pg.skip, pg.size),
)
if err_type != ErrorType.SUCCESS:
res.result.SetResult(err_type)
@ -46,6 +54,21 @@ class ItemService:
res.total = total
return res
async def list_categories(self, company_id: str) -> Res_ItemCategories:
"""회사 전체 상품에서 distinct 카테고리(이름+타입)를 돌려준다. 카테고리 목록은 백엔드가 책임진다."""
res = Res_ItemCategories()
company_uuid = uuid.UUID(company_id)
err_type, rows = await DB_SESSION_MNG.execute_lambda(
items.DBType(),
DBWRType.DB_READ.value,
lambda s: self.item_crud.distinct_categories(s, company_uuid),
)
if err_type != ErrorType.SUCCESS:
res.result.SetResult(err_type)
return res
res.categories = [ItemCategory(name=r[0], category_type=r[1] if r[1] is not None else 1) for r in rows]
return res
async def get_item(self, company_id: str, item_id: str) -> Res_Item:
res = Res_Item()
err_type, item = await self._fetch_owned(uuid.UUID(company_id), uuid.UUID(item_id))
@ -55,9 +78,41 @@ class ItemService:
res.item = ItemData.model_validate(item)
return res
async def check_codes(self, company_id: str, codes: list) -> Res_CheckCodes:
"""업로드 즉시 호출: codes 중 같은 회사 DB 에 이미 있는 코드를 돌려준다(미리보기 사전검사)."""
res = Res_CheckCodes()
company_uuid = uuid.UUID(company_id)
err_type, existing = await DB_SESSION_MNG.execute_lambda(
items.DBType(),
DBWRType.DB_READ.value,
lambda s: self.item_crud.existing_codes(s, company_uuid, codes),
)
if err_type != ErrorType.SUCCESS:
res.result.SetResult(err_type)
return res
res.existing = list(existing)
return res
async def create_item(self, company_id: str, user_id: str, data: dict) -> Res_Item:
res = Res_Item()
item = items(**data, company_id=uuid.UUID(company_id), user_id=uuid.UUID(user_id))
company_uuid = uuid.UUID(company_id)
# DB 중복코드 검증: 같은 회사에 동일 code 가 이미 있으면 거부(프론트는 받아온 목록만 보므로 여기서 최종 차단).
code = data.get("code")
if code:
dup_err, exists = await DB_SESSION_MNG.execute_lambda(
items.DBType(),
DBWRType.DB_READ.value,
lambda s: self.item_crud.code_exists(s, company_uuid, code),
)
if dup_err != ErrorType.SUCCESS:
res.result.SetResult(dup_err)
return res
if exists:
res.result.SetResult(ErrorType.ITEM_CODE_DUPLICATE)
return res
item = items(**data, company_id=company_uuid, user_id=uuid.UUID(user_id))
err_type = await DB_SESSION_MNG.execute_lambda_run(
[items.DBType()],
[lambda s: self.item_crud.add_item(s, item)],

44
negodata/backend/services/supplier_service.py
View File

@ -6,8 +6,9 @@ from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import suppliers
from common.enums import DBWRType, ErrorType
from common.logger import LOG
from common.models.gmodel import PageParams
from crud.supplier_crud import ISupplierCRUD, SupplierCRUD
from router.v1.supplier.protocol import Res_DeleteSupplier, Res_Supplier, Res_SupplierList, SupplierData
from router.v1.supplier.protocol import Res_CheckCodes, Res_DeleteSupplier, Res_Supplier, Res_SupplierList, SupplierData
class SupplierService:
@ -29,15 +30,14 @@ class SupplierService:
return ErrorType.SUPPLIER_NOT_FOUND, None
return ErrorType.SUCCESS, supplier
async def list_suppliers(self, company_id: str, search, page: int, size: int) -> Res_SupplierList:
res = Res_SupplierList(page=page, size=size)
async def list_suppliers(self, company_id: str, search, priority, pg: PageParams) -> Res_SupplierList:
res = Res_SupplierList(page=pg.page, size=pg.size)
company_uuid = uuid.UUID(company_id)
skip = (page - 1) * size
err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
suppliers.DBType(),
DBWRType.DB_READ.value,
lambda s: self.supplier_crud.search(s, company_uuid, search, skip, size),
lambda s: self.supplier_crud.search(s, company_uuid, search, priority, pg.skip, pg.size),
)
if err_type != ErrorType.SUCCESS:
res.result.SetResult(err_type)
@ -55,9 +55,41 @@ class SupplierService:
res.supplier = SupplierData.model_validate(supplier)
return res
async def check_codes(self, company_id: str, codes: list) -> Res_CheckCodes:
"""업로드 즉시 호출: codes 중 같은 회사 DB 에 이미 있는 코드를 돌려준다(미리보기 사전검사)."""
res = Res_CheckCodes()
company_uuid = uuid.UUID(company_id)
err_type, existing = await DB_SESSION_MNG.execute_lambda(
suppliers.DBType(),
DBWRType.DB_READ.value,
lambda s: self.supplier_crud.existing_codes(s, company_uuid, codes),
)
if err_type != ErrorType.SUCCESS:
res.result.SetResult(err_type)
return res
res.existing = list(existing)
return res
async def create_supplier(self, company_id: str, user_id: str, data: dict) -> Res_Supplier:
res = Res_Supplier()
supplier = suppliers(**data, company_id=uuid.UUID(company_id), user_id=uuid.UUID(user_id))
company_uuid = uuid.UUID(company_id)
# DB 중복코드 검증: 같은 회사에 동일 code 가 이미 있으면 거부(프론트는 받아온 목록만 보므로 여기서 최종 차단).
code = data.get("code")
if code:
dup_err, exists = await DB_SESSION_MNG.execute_lambda(
suppliers.DBType(),
DBWRType.DB_READ.value,
lambda s: self.supplier_crud.code_exists(s, company_uuid, code),
)
if dup_err != ErrorType.SUCCESS:
res.result.SetResult(dup_err)
return res
if exists:
res.result.SetResult(ErrorType.SUPPLIER_CODE_DUPLICATE)
return res
supplier = suppliers(**data, company_id=company_uuid, user_id=uuid.UUID(user_id))
err_type = await DB_SESSION_MNG.execute_lambda_run(
[suppliers.DBType()],
[lambda s: self.supplier_crud.add_supplier(s, supplier)],