feat: temp barcode templates

2024-05-03 17:07:36 +03:00
parent 4f62bb2468
commit 99f0308a8a
10 changed files with 341 additions and 1 deletions
--- a/services/barcode.py
+++ b/services/barcode.py
@@ -0,0 +1,147 @@
+from sqlalchemy import select, update
+
+from models import BarcodeTemplate, BarcodeTemplateAttribute, BarcodeTemplateAttributeLink
+from schemas.barcode import (GetBarcodeTemplateByIdRequest,
+                             GetBarcodeTemplateByIdResponse,
+                             BarcodeTemplateCreateResponse,
+                             BarcodeTemplateCreateRequest, CreateBarcodeTemplateAttributeRequest,
+                             BarcodeTemplateUpdateResponse, BarcodeTemplateUpdateRequest)
+from services.base import BaseService
+
+
+class BarcodeService(BaseService):
+
+    # region Template
+    async def get_barcode_template_by_id(self,
+                                         request: GetBarcodeTemplateByIdRequest) -> GetBarcodeTemplateByIdResponse:
+        stmt = select(BarcodeTemplate).filter(BarcodeTemplate.id == request.id)
+        query = await self.session.execute(
+            stmt
+        )
+        return query.scalar()
+
+    async def create_barcode_template(self, request: BarcodeTemplateCreateRequest) -> BarcodeTemplateCreateResponse:
+        try:
+            if request.is_default:
+                stmt = select(BarcodeTemplate).filter(BarcodeTemplate.is_default == True)
+                query = await self.session.execute(stmt)
+                if query.scalar():
+                    raise ValueError('Стандартный шаблон уже существует')
+
+            # prevent duplicate template
+            stmt = select(BarcodeTemplate).filter(BarcodeTemplate.name == request.name)
+            query = await self.session.execute(stmt)
+            if query.scalar():
+                raise ValueError('Шаблон с таким именем уже существует')
+
+            # create template then add attributes
+            template = BarcodeTemplate(name=request.name, is_default=request.is_default)
+            self.session.add(template)
+            await self.session.flush()
+
+            # get all attributes from database
+            stmt = select(BarcodeTemplateAttribute).filter(
+                BarcodeTemplateAttribute.id.in_(request.attribute_ids))
+            query = await self.session.execute(stmt)
+            attributes = query.scalars().all()
+
+            # add attributes to template
+            for attribute in attributes:
+                template_attribute_link = BarcodeTemplateAttributeLink(
+                    barcode_template_id=template.id,
+                    attribute_id=attribute.id
+                )
+                self.session.add(template_attribute_link)
+                await self.session.flush()
+            await self.session.commit()
+            return BarcodeTemplateCreateResponse(message='Шаблон успешно создан',
+                                                 ok=True,
+                                                 id=template.id)
+        except Exception as e:
+            await self.session.rollback()
+            return BarcodeTemplateCreateResponse(message=str(e),
+                                                 ok=False,
+                                                 id=-1)
+
+    async def update_barcode_template(self, request: BarcodeTemplateUpdateRequest) -> BarcodeTemplateUpdateResponse:
+        try:
+            stmt = select(BarcodeTemplate).filter(BarcodeTemplate.id == request.id)
+            query = await self.session.execute(stmt)
+            template = query.scalar()
+            if not template:
+                raise ValueError('Шаблон не найден')
+
+            if request.is_default:
+                stmt = select(BarcodeTemplate).filter(BarcodeTemplate.is_default == True)
+                query = await self.session.execute(stmt)
+                default_template = query.scalar()
+                if default_template and default_template.id != request.id:
+                    raise ValueError('Стандартный шаблон уже существует')
+
+            # update template then add attributes with dict and value syntax
+            request_dict = request.dict()
+            del request_dict['id']
+            del request_dict['attribute_ids']
+            await self.session.execute(
+                update(BarcodeTemplate)
+                .where(BarcodeTemplate.id == request.id)
+                .values(**request_dict)
+            )
+
+            # difference deleted and new
+            template_attributes = set([attribute.attribute_id for attribute in template.attributes])
+            request_attributes = set(request.attribute_ids)
+            new_attributes = request_attributes.difference(template_attributes)
+            deleted_attributes = template_attributes.difference(request_attributes)
+
+            # delete attributes
+            for attribute in template.attributes:
+                if attribute.attribute_id not in deleted_attributes:
+                    continue
+                await self.session.delete(attribute)
+
+            # add new attributes
+            for new_attribute in new_attributes:
+                template_attribute_link = BarcodeTemplateAttributeLink(
+                    barcode_template_id=template.id,
+                    attribute_id=new_attribute
+                )
+                self.session.add(template_attribute_link)
+                await self.session.flush()
+
+            await self.session.commit()
+            return BarcodeTemplateUpdateResponse(message='Шаблон успешно обновлен',
+                                                 ok=True)
+        except Exception as e:
+            await self.session.rollback()
+            return BarcodeTemplateUpdateResponse(message=str(e),
+                                                 ok=False)
+
+    # endregion
+
+    # region Template attributes
+    async def get_all_barcode_template_attributes(self):
+        stmt = select(BarcodeTemplateAttribute)
+        query = await self.session.execute(stmt)
+        return query.scalars().all()
+
+    async def create_barcode_template_attribute(self, request: CreateBarcodeTemplateAttributeRequest):
+        try:
+            # prevent duplicate attribute
+            stmt = select(BarcodeTemplateAttribute).filter(BarcodeTemplateAttribute.name == request.name)
+            query = await self.session.execute(stmt)
+            if query.scalar():
+                raise ValueError('Атрибут с таким именем уже существует')
+
+            attribute = BarcodeTemplateAttribute(**request.dict())
+            self.session.add(attribute)
+            await self.session.commit()
+            return BarcodeTemplateCreateResponse(message='Атрибут успешно создан',
+                                                 ok=True,
+                                                 id=attribute.id)
+        except Exception as e:
+            await self.session.rollback()
+            return BarcodeTemplateCreateResponse(message=str(e),
+                                                 ok=False,
+                                                 id=-1)
+    # endregion