71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
import json
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import Engine
|
|
from sqlmodel import SQLModel, Field, Session, select
|
|
from typing_extensions import override
|
|
|
|
from core.domain.optimization.Optimization import Optimization
|
|
from core.domain.optimization.OptimizationVehicle import OptimizationVehicle
|
|
from core.domain.optimization.TransportMode import TransportMode
|
|
from core.repos.OptimizationVehicleRepo import OptimizationVehicleRepo
|
|
from core.types.Id import Id
|
|
|
|
|
|
@dataclass
|
|
class OptimizationVehicleSqlRepo(OptimizationVehicleRepo):
|
|
db: Engine
|
|
|
|
class Table(SQLModel, table=True):
|
|
__tablename__ = "optimization_vehicle"
|
|
|
|
id: str = Field(primary_key=True)
|
|
optimization_id: str = Field(foreign_key="optimization.id")
|
|
name: str
|
|
type: str
|
|
capacity: int
|
|
range: float # kilometers
|
|
min_quantity: int
|
|
max_quantity: int
|
|
delivery_time: float # hours
|
|
average_speed: float
|
|
max_speed: float
|
|
districts: str
|
|
|
|
def toDomain(self) -> OptimizationVehicle:
|
|
return OptimizationVehicle(
|
|
optimizationId=Id(value=uuid.UUID(self.optimization_id)),
|
|
name=self.name,
|
|
type=TransportMode(self.type),
|
|
capacity=self.capacity,
|
|
range=self.range,
|
|
minQuantity=self.min_quantity,
|
|
maxQuantity=self.max_quantity,
|
|
deliveryTime=self.delivery_time,
|
|
averageSpeed=self.average_speed,
|
|
maxSpeed=self.max_speed,
|
|
districts=self.districts,
|
|
id=Id(value=uuid.UUID(self.id))
|
|
)
|
|
|
|
@override
|
|
def getAll(self) -> list[OptimizationVehicle]:
|
|
with Session(self.db) as conn:
|
|
query = select(self.Table)
|
|
return [row.toDomain() for row in conn.exec(query).all()]
|
|
|
|
@override
|
|
def get(self, id: Id[OptimizationVehicle]) -> Optional[OptimizationVehicle]:
|
|
with Session(self.db) as conn:
|
|
query = select(self.Table).filter_by(id=id.value)
|
|
row = conn.exec(query).one_or_none()
|
|
return row.toDomain() if row is not None else None
|
|
|
|
@override
|
|
def getAllByOptimizationId(self, optimizationId: Id[Optimization]) -> list[OptimizationVehicle]:
|
|
with Session(self.db) as conn:
|
|
query = select(self.Table).filter_by(optimization_id=optimizationId.value)
|
|
return [row.toDomain() for row in conn.exec(query).all()]
|