feat: barcode templates

modules/fulfillment_base/repositories/__init__.py

@@ -5,3 +5,4 @@ from .product import ProductRepository as ProductRepository
 from .service import ServiceRepository as ServiceRepository
 from .services_kit import ServicesKitRepository as ServicesKitRepository
 from .service_category import ServiceCategoryRepository as ServiceCategoryRepository
+from .barcode_template import BarcodeTemplateRepository as BarcodeTemplateRepository

modules/fulfillment_base/repositories/barcode_template.py

@@ -0,0 +1,113 @@
+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)
+        )
+
+    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())