2
from typing import Any, Optional
3
from typing import TYPE_CHECKING
5
from sqlalchemy import select
6
from sqlalchemy.ext.asyncio import AsyncSession
8
from app.inventory.quant.models.quants_models import Quant
9
from app.inventory.quant.schemas.quants_schemas import QuantCreateScheme, QuantUpdateScheme, QuantFilter
10
from core.permissions import permit
11
from core.service.base import BaseService, UpdateSchemaType, ModelType, FilterSchemaType, CreateSchemaType
12
from starlette.requests import Request
17
class QuantService(BaseService[Quant, QuantCreateScheme, QuantUpdateScheme, QuantFilter]):
18
def __init__(self, request:Request):
19
super(QuantService, self).__init__(request, Quant, QuantCreateScheme, QuantUpdateScheme)
22
async def update(self, id: Any, obj: UpdateSchemaType) -> Optional[ModelType]:
23
return await super(QuantService, self).update(id, obj)
26
async def list(self, _filter: FilterSchemaType, size: int):
27
return await super(QuantService, self).list(_filter, size)
29
@permit('quant_create')
30
async def create(self, obj: CreateSchemaType, commit=False) -> ModelType:
31
return await super(QuantService, self).create(obj, commit)
33
@permit('quant_delete')
34
async def delete(self, id: Any) -> None:
35
return await super(QuantService, self).delete(id)
37
async def get_available_quants(
39
product_id: uuid.UUID,
42
exclude_id: uuid.UUID = None,
43
location_class_ids: [uuid.UUID] = None,
44
location_ids: [uuid.UUID] = None,
45
location_type_ids: [uuid.UUID] = None,
46
lot_ids: [uuid.UUID] = None,
47
partner_id: uuid.UUID = None,
50
Метод получения квантов по параметрам
52
query = select(self.model)
54
query = query.where(self.model.id == id)
57
query = query.where(self.model.product_id == product_id)
59
query = query.where(self.model.id != exclude_id)
61
query = query.where(self.model.store_id == store_id)
62
if location_class_ids:
63
query = query.where(self.model.location_class.in_(location_class_ids))
65
query = query.where(self.model.location_id.in_(location_ids))
67
query = query.where(self.model.location_type_id.in_(location_type_ids))
69
query = query.where(self.model.lot_id.in_(lot_ids))
71
"""Если не указываем партнера, то партнер None тк тут не подходит логика --> Любой"""
72
query = query.where(self.model.partner_id == partner_id)
74
executed_data = await self.session.execute(query)
75
return executed_data.scalars().all()