49 lines
2.1 KiB
Python
49 lines
2.1 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.entities.delivery import DeliveryEntity
|
|
from app.infrastructure.database.models.deliveries import Delivery
|
|
from app.infrastructure.database.repository.delivery_repository import DeliveryRepository
|
|
|
|
|
|
class DeliveryService:
|
|
def __init__(self, db: Session):
|
|
self.repository = DeliveryRepository(db)
|
|
|
|
def get_all_deliveries(self) -> List[DeliveryEntity]:
|
|
deliveries = self.repository.get_all()
|
|
return [DeliveryEntity(d.id, d.count, d.storage_accessories_id, d.step_id) for d in deliveries]
|
|
|
|
def get_delivery_by_id(self, delivery_id: int) -> Optional[DeliveryEntity]:
|
|
delivery = self.repository.get_by_id(delivery_id)
|
|
if delivery:
|
|
return DeliveryEntity(delivery.id, delivery.count, delivery.storage_accessories_id, delivery.step_id)
|
|
|
|
return None
|
|
|
|
def create_delivery(self, entity: DeliveryEntity) -> DeliveryEntity:
|
|
delivery_model = Delivery(
|
|
count=entity.count,
|
|
storage_accessories_id=entity.storage_accessories_id,
|
|
step_id=entity.step_id
|
|
)
|
|
created_delivery = self.repository.create(delivery_model)
|
|
return DeliveryEntity(created_delivery.id, created_delivery.count, created_delivery.storage_accessories_id,
|
|
created_delivery.step_id)
|
|
|
|
def update_delivery(self, delivery_id: int, entity: DeliveryEntity) -> Optional[DeliveryEntity]:
|
|
delivery_model = self.repository.get_by_id(delivery_id)
|
|
if delivery_model:
|
|
delivery_model.count = entity.count
|
|
delivery_model.storage_accessories_id = entity.storage_accessories_id
|
|
delivery_model.step_id = entity.step_id
|
|
self.repository.update(delivery_model)
|
|
return DeliveryEntity(delivery_model.id, delivery_model.count, delivery_model.storage_accessories_id,
|
|
delivery_model.step_id)
|
|
|
|
return None
|
|
|
|
def delete_delivery(self, delivery_id: int) -> bool:
|
|
return self.repository.delete(delivery_id) is not None
|