fakz9/Crm-Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: product endpoints changes for products table

Browse Source
This commit is contained in:
AlexSserb
2025-10-08 22:30:43 +04:00
parent 7d6155ff6c
commit 6b0f8a1aa5
6 changed files with 127 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
modules/fulfillment_base/repositories/deal_product.py
View File

@ -3,7 +3,7 @@ from typing import Optional
from sqlalchemy import Select, select
from sqlalchemy.orm import joinedload, selectinload
from modules.fulfillment_base.models import DealProductService
from modules.fulfillment_base.models import DealProductService, Product
from modules.fulfillment_base.models.deal_product import DealProduct
from modules.fulfillment_base.schemas.deal_product import (
UpdateDealProductSchema,
@ -25,7 +25,7 @@ class DealProductRepository(
deal_id = args[0]
return (
stmt.options(
joinedload(DealProduct.product),
joinedload(DealProduct.product).selectinload(Product.barcodes),
selectinload(DealProduct.product_services).joinedload(
DealProductService.service
),

79
modules/fulfillment_base/repositories/product.py
View File

@ -1,15 +1,16 @@
from modules.fulfillment_base.models import Product
from sqlalchemy import or_, delete
from sqlalchemy.orm import selectinload
from modules.fulfillment_base.models import Product, ProductBarcode
from modules.fulfillment_base.schemas.product import (
CreateProductSchema,
UpdateProductSchema,
)
from repositories.mixins import *
from schemas.base import PaginationSchema
class ProductRepository(
BaseRepository,
RepGetAllMixin[Product],
RepDeleteMixin[Product],
RepCreateMixin[Product, CreateProductSchema],
RepUpdateMixin[Product, UpdateProductSchema],
@ -18,17 +19,73 @@ class ProductRepository(
entity_class = Product
entity_not_found_msg = "Товар не найден"
def _process_get_all_stmt_with_args(self, stmt: Select, *args) -> Select:
search_input = args[0]
pagination: PaginationSchema = args[1]
async def get_all(
self,
page: Optional[int],
items_per_page: Optional[int],
client_id: Optional[int],
search_input: Optional[str],
) -> tuple[list[Product], int]:
stmt = (
select(Product)
.options(selectinload(Product.barcodes))
.where(Product.is_deleted.is_(False))
)
if client_id:
stmt = stmt.where(Product.client_id == client_id)
if search_input:
stmt = stmt.where(Product.name.ilike(f"%{search_input}%"))
if pagination.items_per_page and pagination.page:
stmt = self._apply_pagination(
stmt, pagination.page, pagination.items_per_page
stmt = stmt.where(
or_(
Product.name.ilike(f"%{search_input}%"),
Product.barcodes.any(
ProductBarcode.barcode.ilike(f"%{search_input}%")
),
Product.article.ilike(f"%{search_input}%"),
Product.factory_article.ilike(f"%{search_input}%"),
)
)
return stmt
total_items = len((await self.session.execute(stmt)).all())
if page and items_per_page:
stmt = self._apply_pagination(stmt, page, items_per_page)
result = await self.session.execute(stmt)
return list(result.scalars().all()), total_items
def _process_get_by_id_stmt(self, stmt: Select) -> Select:
return stmt.options(selectinload(Product.barcodes))
async def _after_create(self, product: Product, data: CreateProductSchema) -> None:
new_barcodes = [
ProductBarcode(product_id=product.id, barcode=barcode)
for barcode in data.barcodes
]
self.session.add_all(new_barcodes)
async def _update_barcodes(self, product: Product, new_barcodes: list[str]):
new_barcodes_set: set[str] = set(new_barcodes)
old_barcodes_set: set[str] = set(obj.barcode for obj in product.barcodes)
barcodes_to_add = new_barcodes_set - old_barcodes_set
barcodes_to_delete = old_barcodes_set - new_barcodes_set
del_stmt = delete(ProductBarcode).where(
ProductBarcode.product_id == product.id,
ProductBarcode.barcode.in_(barcodes_to_delete),
)
await self.session.execute(del_stmt)
new_barcodes = [
ProductBarcode(product_id=product.id, barcode=barcode)
for barcode in barcodes_to_add
]
self.session.add_all(new_barcodes)
await self.session.commit()
await self.session.refresh(product)
async def update(self, product: Product, data: UpdateProductSchema) -> Product:
if data.barcodes is not None:
await self._update_barcodes(product, data.barcodes)
del data.barcodes
return await self._apply_update_data_to_model(product, data, True)

3
modules/fulfillment_base/routers/product.py
View File

@ -15,9 +15,10 @@ router = APIRouter(tags=["product"])
async def get_products(
session: SessionDependency,
pagination: PaginationDependency,
client_id: Optional[int] = Query(alias="clientId", default=None),
search_input: Optional[str] = Query(alias="searchInput", default=None),
):
return await ProductService(session).get_all(search_input, pagination)
return await ProductService(session).get_all(pagination, client_id, search_input)
@router.post(

19
modules/fulfillment_base/schemas/product.py
View File

@ -1,6 +1,10 @@
from typing import Optional
from schemas.base import BaseSchema, BaseResponse
from pydantic import field_validator
from modules.fulfillment_base.models import ProductBarcode
from modules.fulfillment_base.schemas.barcode_template import BarcodeTemplateSchema
from schemas.base import BaseSchema, BaseResponse, PaginationInfoSchema
# region Entity
@ -16,26 +20,38 @@ class CreateProductSchema(BaseSchema):
name: str
article: str
factory_article: str
client_id: int
barcode_template_id: int
brand: Optional[str]
color: Optional[str]
composition: Optional[str]
size: Optional[str]
additional_info: Optional[str]
barcodes: list[str]
class ProductSchema(CreateProductSchema):
id: int
barcode_template: BarcodeTemplateSchema
@field_validator("barcodes", mode="before")
def barcodes_to_list(cls, v: Optional[list[ProductBarcode]]):
if isinstance(v, list):
return [barcode.barcode for barcode in v]
return v
class UpdateProductSchema(BaseSchema):
name: Optional[str] = None
article: Optional[str] = None
factory_article: Optional[str] = None
barcode_template_id: Optional[int] = None
brand: Optional[str] = None
color: Optional[str] = None
composition: Optional[str] = None
size: Optional[str] = None
additional_info: Optional[str] = None
barcodes: Optional[list[str]] = None
images: list[ProductImageSchema] | None = []
@ -60,6 +76,7 @@ class UpdateProductRequest(BaseSchema):
class GetProductsResponse(BaseSchema):
items: list[ProductSchema]
pagination_info: PaginationInfoSchema
class CreateProductResponse(BaseResponse):

28
modules/fulfillment_base/services/product.py
View File

@ -1,15 +1,17 @@
import math
from modules.fulfillment_base.models import Product
from modules.fulfillment_base.repositories import ProductRepository
from modules.fulfillment_base.schemas.product import (
CreateProductRequest,
ProductSchema,
UpdateProductRequest,
UpdateProductRequest, GetProductsResponse,
)
from schemas.base import PaginationSchema, PaginationInfoSchema
from services.mixins import *
class ProductService(
ServiceGetAllMixin[Product, ProductSchema],
ServiceCreateMixin[Product, CreateProductRequest, ProductSchema],
ServiceUpdateMixin[Product, UpdateProductRequest],
ServiceDeleteMixin[Product],
@ -22,5 +24,27 @@ class ProductService(
def __init__(self, session: AsyncSession):
self.repository = ProductRepository(session)
async def get_all(
self,
pagination: PaginationSchema,
*filters,
) -> GetProductsResponse:
products, total_items = await self.repository.get_all(
pagination.page,
pagination.items_per_page,
*filters,
)
total_pages = 1
if pagination.items_per_page:
total_pages = math.ceil(total_items / pagination.items_per_page)
return GetProductsResponse(
items=[ProductSchema.model_validate(product) for product in products],
pagination_info=PaginationInfoSchema(
total_pages=total_pages, total_items=total_items
),
)
async def is_soft_delete(self, product: ProductSchema) -> bool:
return True