"""Fluent builders for assembling transport chains and related rows.
Each :class:`ModelBuilder` subclass wraps a Django model instance and
exposes chainable setter methods so that complex transport chains can
be constructed declaratively. Builders compose via
:class:`BuilderOfBuilders` (a chain holds builders for its elements
and freight items), and can either ``save()`` rows one-by-one or
defer everything into a :class:`BulkCreator` for one bulk upsert per
model class.
"""
from abc import ABC, abstractmethod
from datetime import date
from decimal import Decimal
from typing import List, Type
from uuid import UUID
from django.db.models import Model
from ferrosoft.apps.emiflow.models import (
TransportChain,
BookingStatus,
TransportChainElement,
DistanceType,
FreightItem,
TCEGeoPoint,
TCEGeoOrigin,
TCEGeoDestination,
TransportChainType,
Incoterm,
OperationEpitype,
TransportOperation,
)
from ferrosoft.apps.ferrobase.models import BusinessPartner, Material, Country
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferromaps.services.addresses import Address, Coordinates
[docs]
def resolve_model(needle, model_type):
"""Resolve ``needle`` to an instance of ``model_type``.
Args:
needle: ``str`` to look up by ``name``, ``UUID`` to look up by
primary key, an instance of ``model_type`` (returned
unchanged), or ``None`` (returned as ``None``).
model_type: Django model class.
Returns:
The resolved model instance, or ``None`` when ``needle`` is
``None``.
Raises:
ValueError: When ``needle`` is none of the accepted types.
model_type.DoesNotExist: When the lookup finds no row.
"""
if needle is None:
return None
if isinstance(needle, model_type):
return needle
elif isinstance(needle, str):
return model_type.objects.get(name=needle)
elif isinstance(needle, UUID):
return model_type.objects.get(pk=needle)
else:
raise ValueError("Cannot lookup %s with %s" % (model_type, type(needle)))
[docs]
class BulkCreator:
"""Collects unsaved model instances and upserts them in declared order.
The model classes are grouped by name; on :meth:`upsert` the
creator iterates through ``create_order`` so foreign keys are
satisfied (e.g. ``TransportChain`` before ``TransportChainElement``)
and each model's manager-level ``upsert`` is invoked once with
all instances of that class.
"""
def __init__(self, create_order: list[Type[Model]]):
"""Configure the model class order used by :meth:`upsert`."""
self.models = {}
self.create_order = create_order
[docs]
def add_model(self, model: Model):
"""Add ``model`` to the pending bulk for its class."""
model_name = type(model).__name__
if model_name not in self.models:
self.models[model_name] = []
self.models[model_name].append(model)
[docs]
def upsert(self):
"""Run ``Manager.upsert(...)`` for each pending model class in declared order.
Returns:
dict[str, list]: A mapping of class name to the list of
upserted instances. Classes with no pending rows are
omitted.
Raises:
RuntimeError: When a model class's manager has no
``upsert`` method.
"""
results = {}
for model_class in self.create_order:
name = model_class.__name__
if name in self.models and len(self.models[name]) > 0:
upsert = getattr(model_class.objects, "upsert", None)
if not callable(upsert):
raise RuntimeError(
"Manager for %s model has no upsert method" % name
)
results[name] = upsert(self.models[name])
return results
[docs]
class ModelBuilder(ABC):
"""Abstract fluent builder around a single Django model instance.
Subclasses expose chainable setter methods returning ``self`` and
must implement :meth:`save` to persist the instance individually
or :meth:`add_bulk` to defer persistence into a
:class:`BulkCreator`.
"""
[docs]
@abstractmethod
def save(self, *args, **kwargs):
"""Persist the built model instance and return it."""
raise NotImplementedError
[docs]
def add_bulk(self, bulk: BulkCreator):
"""Register the built instance with ``bulk`` for deferred persistence."""
raise NotImplementedError
[docs]
class BuilderOfBuilders(ModelBuilder):
"""Aggregate of child builders that fans ``save``/``add_bulk`` out to each."""
def __init__(self):
self.builders: List[ModelBuilder] = []
[docs]
def append(self, builder: ModelBuilder) -> ModelBuilder:
"""Add ``builder`` to the aggregate and return it for chaining."""
self.builders.append(builder)
return builder
[docs]
def save(self, *args, **kwargs):
"""Save every child builder."""
for builder in self.builders:
builder.save(*args, **kwargs)
[docs]
def add_bulk(self, bulk: BulkCreator):
"""Register every child's instance with ``bulk``."""
for builder in self.builders:
builder.add_bulk(bulk)
def __len__(self):
return len(self.builders)
[docs]
class MaterialBuilder(ModelBuilder):
"""Fluent builder around a :class:`Material`."""
def __init__(self, material: Material):
self.material = material
[docs]
@classmethod
def new(cls) -> "MaterialBuilder":
"""Start a fresh builder around an empty :class:`Material`."""
return cls(Material())
[docs]
def code(self, code: str) -> "MaterialBuilder":
self.material.code = code
return self
[docs]
def name(self, name: str) -> "MaterialBuilder":
self.material.name = name
return self
[docs]
def save(self, *args, **kwargs) -> Material:
self.material.save(*args, **kwargs)
return self.material
[docs]
def add_bulk(self, bulk: BulkCreator) -> Material:
bulk.add_model(self.material)
return self.material
[docs]
class GeoPointBuilder(ModelBuilder):
"""Fluent builder around a :class:`TCEGeoPoint` (origin or destination)."""
def __init__(self, point: TCEGeoPoint):
self.point = point
[docs]
@classmethod
def new(cls) -> "GeoPointBuilder":
"""Start a fresh builder around an empty :class:`TCEGeoPoint`."""
return cls(TCEGeoPoint())
[docs]
def address(self, address: Address) -> "GeoPointBuilder":
self.point.first_name = address.first_name
self.point.last_name = address.last_name
self.point.company = address.company
self.point.additional_lines = address.additional_lines
self.point.street = address.street
self.point.house_no = address.house_no
self.point.city = address.city
self.point.postal_code = address.postal_code
self.point.country = Country.get_by_iso_code(address.country)
return self
[docs]
def coordinates(self, coords: Coordinates) -> "GeoPointBuilder":
self.point.latitude = coords.lat
self.point.longitude = coords.lon
return self
[docs]
def save(self, *args, **kwargs) -> TCEGeoPoint:
self.point.save(*args, **kwargs)
return self.point
[docs]
def add_bulk(self, bulk: BulkCreator) -> TCEGeoPoint:
bulk.add_model(self.point)
return self.point
[docs]
class TransportChainElementBuilder(ModelBuilder):
"""Fluent builder for one :class:`TransportChainElement` (with optional geo-points)."""
def __init__(self, element: TransportChainElement):
self.element = element
self._point_builders = BuilderOfBuilders()
[docs]
@classmethod
def new(cls) -> "TransportChainElementBuilder":
"""Start a fresh builder around an empty :class:`TransportChainElement`."""
return cls(TransportChainElement())
[docs]
def chain(self, chain: TransportChain) -> "TransportChainElementBuilder":
self.element.chain = chain
return self
[docs]
def epitype(self, epitype: OperationEpitype) -> "TransportChainElementBuilder":
self.element.epitype = epitype
return self
[docs]
def position(self, position: int) -> "TransportChainElementBuilder":
self.element.position = position
return self
[docs]
def operation(
self, operation: TransportOperation
) -> "TransportChainElementBuilder":
self.element.operation = operation
return self
[docs]
def activity_distance_type(
self, dtype: DistanceType
) -> "TransportChainElementBuilder":
self.element.activity_distance_type = dtype
return self
[docs]
def activity_distance(
self, distance: DecimalWithUnit
) -> "TransportChainElementBuilder":
self.element.activity_distance_value = distance.value
self.element.activity_distance_unit = distance.unit
return self
[docs]
def distance_adjustment_factor(
self, daf: Decimal
) -> "TransportChainElementBuilder":
self.element.distance_adjustment_factor = daf
return self
[docs]
def add_geo_origin(self) -> GeoPointBuilder:
"""Attach a new origin geo-point and return its builder for chaining."""
builder = GeoPointBuilder(TCEGeoOrigin(tce=self.element))
self._point_builders.append(builder)
return builder
[docs]
def add_geo_destination(self) -> GeoPointBuilder:
"""Attach a new destination geo-point and return its builder for chaining."""
builder = GeoPointBuilder(TCEGeoDestination(tce=self.element))
self._point_builders.append(builder)
return builder
[docs]
def save(self, *args, **kwargs) -> TransportChainElement:
self.element.save(*args, **kwargs)
self._point_builders.save(*args, **kwargs)
return self.element
[docs]
def add_bulk(self, bulk: BulkCreator) -> TransportChainElement:
bulk.add_model(self.element)
self._point_builders.add_bulk(bulk)
return self.element
[docs]
class FreightItemBuilder(ModelBuilder):
"""Fluent builder for one :class:`FreightItem`."""
def __init__(self, item: FreightItem):
self.item = item
[docs]
@classmethod
def new(cls) -> "FreightItemBuilder":
"""Start a fresh builder around an empty :class:`FreightItem`."""
return cls(FreightItem())
[docs]
def chain(self, chain: TransportChain) -> "FreightItemBuilder":
self.item.chain = chain
return self
[docs]
def weight(self, weight: DecimalWithUnit) -> "FreightItemBuilder":
self.item.weight_value = weight.value
self.item.weight_unit = weight.unit
return self
[docs]
def material(self, material: str | UUID | Material) -> "FreightItemBuilder":
self.item.material = resolve_model(material, Material)
return self
[docs]
def save(self, *args, **kwargs) -> FreightItem:
self.item.save(*args, **kwargs)
return self.item
[docs]
def add_bulk(self, bulk: BulkCreator) -> FreightItem:
bulk.add_model(self.item)
return self.item
[docs]
class TransportChainBuilder(ModelBuilder):
"""Top-level fluent builder that composes the whole chain.
Owns nested builders for chain elements (auto-numbered by
insertion order) and freight items. ``save`` cascades to all
children so the entire object graph can be persisted from a single
call.
"""
def __init__(self, chain: TransportChain):
self.chain = chain
self.element_builders = BuilderOfBuilders()
self.other_builders = BuilderOfBuilders()
[docs]
@classmethod
def new(cls) -> "TransportChainBuilder":
"""Start a fresh builder around an empty :class:`TransportChain`."""
return cls(TransportChain())
[docs]
def reference(self, reference: str) -> "TransportChainBuilder":
self.chain.reference = reference
return self
[docs]
def booking_date(self, booking_date: date) -> "TransportChainBuilder":
self.chain.booking_date = booking_date
return self
[docs]
def status(self, status: BookingStatus) -> "TransportChainBuilder":
self.chain.status = status
return self
[docs]
def transport_type(
self, transport_type: TransportChainType
) -> "TransportChainBuilder":
self.chain.transport_type = transport_type
return self
[docs]
def incoterms(self, incoterms: Incoterm | None) -> "TransportChainBuilder":
self.chain.incoterms = incoterms
return self
[docs]
def simulation(self, simulation: bool) -> "TransportChainBuilder":
self.chain.simulation = simulation
return self
[docs]
def consignor(
self, partner: str | UUID | BusinessPartner
) -> "TransportChainBuilder":
self.chain.consignor = resolve_model(partner, BusinessPartner)
return self
[docs]
def consignee(
self, partner: str | UUID | BusinessPartner
) -> "TransportChainBuilder":
self.chain.consignee = resolve_model(partner, BusinessPartner)
return self
[docs]
def add_element(self) -> TransportChainElementBuilder:
"""Append a new chain element (auto-numbered by insertion order) and return its builder."""
builder = TransportChainElementBuilder(
TransportChainElement(
chain=self.chain,
position=len(self.element_builders) + 1,
)
)
self.element_builders.append(builder)
return builder
[docs]
def add_freight_item(self) -> FreightItemBuilder:
"""Append a new freight item to the chain and return its builder."""
builder = FreightItemBuilder(FreightItem(chain=self.chain))
self.other_builders.append(builder)
return builder
[docs]
def save(self, *args, **kwargs) -> TransportChain:
self.chain.save(*args, **kwargs)
self.element_builders.save(*args, **kwargs)
self.other_builders.save(*args, **kwargs)
return self.chain
[docs]
def add_bulk(self, bulk: BulkCreator) -> TransportChain:
bulk.add_model(self.chain)
self.element_builders.add_bulk(bulk)
self.other_builders.add_bulk(bulk)
return self.chain