1
from typing import Any, Optional
3
from sqlalchemy import select
5
from app.basic.product.enums.exceptions_product_enums import ProductErrors
6
from app.basic.product.models.product_models import Product
7
from app.basic.product.schemas import ProductCreateScheme, ProductUpdateScheme, ProductFilter
8
from core.exceptions.module import ModuleException
9
from core.permissions import permit
10
from core.service.base import BaseService, UpdateSchemaType, ModelType, FilterSchemaType, CreateSchemaType
11
from starlette.requests import Request
13
class ProductService(BaseService[Product, ProductCreateScheme, ProductUpdateScheme, ProductFilter]):
14
def __init__(self, request: Request):
15
super(ProductService, self).__init__(request, Product, ProductCreateScheme, ProductUpdateScheme)
17
@permit('product_edit')
18
async def update(self, id: Any, obj: UpdateSchemaType) -> Optional[ModelType]:
19
return await super(ProductService, self).update(id, obj)
21
@permit('product_list')
22
async def list(self, _filter: FilterSchemaType, size: int):
23
return await super(ProductService, self).list(_filter, size)
25
@permit('product_create')
26
async def create(self, obj: CreateSchemaType) -> ModelType:
27
return await super(ProductService, self).create(obj)
29
@permit('product_delete')
30
async def delete(self, id: Any) -> None:
31
return await super(ProductService, self).delete(id)
33
@permit('product_by_barcode')
34
async def product_by_barcode(self, barcode: str):
35
query = select(self.model)
36
query = query.where(self.model.barcode_list.contains([barcode]))
37
result = await self.session.execute(query)
38
entity = result.scalars().first()
40
raise ModuleException(status_code=404, enum=ProductErrors.PRODUCT_NOT_FOUND)