120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
from sqlalchemy import update
|
|
from sqlalchemy.orm import joinedload, selectinload
|
|
|
|
from modules.fulfillment_base.models import (
|
|
BarcodeTemplate,
|
|
BarcodeTemplateAttribute,
|
|
BarcodeTemplateSize,
|
|
)
|
|
from modules.fulfillment_base.schemas.barcode_template import (
|
|
CreateBarcodeTemplateSchema,
|
|
UpdateBarcodeTemplateSchema,
|
|
)
|
|
from repositories.mixins import *
|
|
|
|
|
|
class BarcodeTemplateRepository(
|
|
RepCrudMixin[
|
|
BarcodeTemplate, CreateBarcodeTemplateSchema, UpdateBarcodeTemplateSchema
|
|
],
|
|
):
|
|
session: AsyncSession
|
|
entity_class = BarcodeTemplate
|
|
entity_not_found_msg = "Шаблон штрихкода не найден"
|
|
|
|
def _process_get_all_stmt(self, stmt: Select) -> Select:
|
|
return (
|
|
stmt.options(
|
|
selectinload(BarcodeTemplate.attributes),
|
|
joinedload(BarcodeTemplate.size),
|
|
)
|
|
.where(BarcodeTemplate.is_deleted.is_(False))
|
|
.order_by(BarcodeTemplate.id)
|
|
)
|
|
|
|
def _process_get_by_id_stmt(self, stmt: Select) -> Select:
|
|
return stmt.options(
|
|
selectinload(BarcodeTemplate.attributes),
|
|
joinedload(BarcodeTemplate.size),
|
|
)
|
|
|
|
async def _get_size_by_id(self, size_id: int) -> Optional[BarcodeTemplateSize]:
|
|
stmt = select(BarcodeTemplateSize).where(BarcodeTemplateSize.id == size_id)
|
|
result = await self.session.scalars(stmt)
|
|
return result.one_or_none()
|
|
|
|
async def _get_attrs_by_ids(
|
|
self, attrs_ids: list[int]
|
|
) -> list[BarcodeTemplateAttribute]:
|
|
stmt = select(BarcodeTemplateAttribute).where(
|
|
BarcodeTemplateAttribute.id.in_(attrs_ids)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def create(self, data: CreateBarcodeTemplateSchema) -> int:
|
|
if data.is_default is not None and data.is_default:
|
|
await self._turn_off_defaults()
|
|
|
|
data_dict = data.model_dump()
|
|
data_dict["size"] = await self._get_size_by_id(data.size.id)
|
|
data_dict["attributes"] = await self._get_attrs_by_ids(
|
|
[a.id for a in data.attributes]
|
|
)
|
|
|
|
obj = BarcodeTemplate(**data_dict)
|
|
self.session.add(obj)
|
|
await self.session.commit()
|
|
await self.session.refresh(obj)
|
|
return obj.id
|
|
|
|
async def _turn_off_defaults(self):
|
|
stmt = (
|
|
update(BarcodeTemplate)
|
|
.where(BarcodeTemplate.is_default.is_(True))
|
|
.values({"is_default": False})
|
|
)
|
|
await self.session.execute(stmt)
|
|
|
|
async def _set_first_as_default(self, with_commit: bool = False):
|
|
stmt = select(BarcodeTemplate).limit(1)
|
|
result = await self.session.execute(stmt)
|
|
obj = result.scalar()
|
|
if not obj:
|
|
return
|
|
obj.is_default = True
|
|
self.session.add(obj)
|
|
if with_commit:
|
|
await self.session.commit()
|
|
await self.session.refresh(obj)
|
|
|
|
async def update(
|
|
self, template: BarcodeTemplate, data: UpdateBarcodeTemplateSchema
|
|
) -> BarcodeTemplate:
|
|
if data.size is not None:
|
|
data.size = await self._get_size_by_id(data.size.id)
|
|
if data.attributes is not None:
|
|
data.attributes = await self._get_attrs_by_ids(
|
|
[a.id for a in data.attributes]
|
|
)
|
|
if data.is_default is not None:
|
|
if data.is_default:
|
|
await self._turn_off_defaults()
|
|
else:
|
|
await self._set_first_as_default()
|
|
return await self._apply_update_data_to_model(template, data, True)
|
|
|
|
async def _before_delete(self, template: BarcodeTemplate):
|
|
if template.is_default:
|
|
await self._set_first_as_default()
|
|
|
|
async def get_attributes(self) -> list[BarcodeTemplateAttribute]:
|
|
stmt = select(BarcodeTemplateAttribute)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_sizes(self) -> list[BarcodeTemplateSize]:
|
|
stmt = select(BarcodeTemplateSize)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|