lavkach3

1
from typing import Any, Optional
2

3
from sqlalchemy import select
4

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
12

13
class ProductService(BaseService[Product, ProductCreateScheme, ProductUpdateScheme, ProductFilter]):
14
    def __init__(self, request: Request):
15
        super(ProductService, self).__init__(request, Product, ProductCreateScheme, ProductUpdateScheme)
16

17
    @permit('product_edit')
18
    async def update(self, id: Any, obj: UpdateSchemaType) -> Optional[ModelType]:
19
        return await super(ProductService, self).update(id, obj)
20

21
    @permit('product_list')
22
    async def list(self, _filter: FilterSchemaType, size: int):
23
        return await super(ProductService, self).list(_filter, size)
24

25
    @permit('product_create')
26
    async def create(self, obj: CreateSchemaType) -> ModelType:
27
        return await super(ProductService, self).create(obj)
28

29
    @permit('product_delete')
30
    async def delete(self, id: Any) -> None:
31
        return await super(ProductService, self).delete(id)
32

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()
39
        if not entity:
40
            raise ModuleException(status_code=404, enum=ProductErrors.PRODUCT_NOT_FOUND)
41
        return entity
42