38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
|
|
from models.product import Product
|
|
from schemas.base import PaginationSchema
|
|
from services.base import BaseService
|
|
from schemas.product import *
|
|
|
|
|
|
class ProductService(BaseService):
|
|
async def create(self, request: ProductCreateRequest) -> ProductCreateResponse:
|
|
existing_product_query = await self.session.execute(
|
|
select(Product)
|
|
.where(Product.client_id == request.client_id,
|
|
Product.article == request.article)
|
|
)
|
|
existing_product = existing_product_query.first()
|
|
if existing_product:
|
|
raise HTTPException(status_code=403, detail="Product already exists")
|
|
product = Product(**request.dict())
|
|
self.session.add(product)
|
|
await self.session.commit()
|
|
return ProductCreateResponse(product_id=product.id)
|
|
|
|
async def get_by_client_id(self, client_id: int, pagination: PaginationSchema) -> ProductGetResponse:
|
|
query = await self.session.execute(
|
|
select(Product)
|
|
.where(Product.client_id == client_id)
|
|
.offset(pagination.page * pagination.items_per_page)
|
|
.limit(pagination.items_per_page)
|
|
)
|
|
products: list[ProductSchema] = []
|
|
for product in query.scalars().all():
|
|
products.append(
|
|
ProductSchema.model_validate(product)
|
|
)
|
|
return ProductGetResponse(products=products)
|