"""
Models shared between apps.
"""
import datetime
import logging
from datetime import timedelta
from decimal import Decimal
from django.conf import settings
from django.db import models
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from django.utils.translation import pgettext_lazy
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.ferrobase.models import (
ISOCatalogEntry,
CoordinatesMixin,
CoordinateDecimalField,
ISOCatalogEntryManager,
)
from ferrosoft.apps.ferrobase.models.base import (
AddressMixin,
CommonLimit,
SelectOptionsMixin,
UUIDModel,
)
from ferrosoft.apps.ferromaps.services.addresses import CoordinatesNotFound
from ferrosoft.decimal import zero
[docs]
class ColorTheme(models.TextChoices):
"""User-selectable color theme for the web interface."""
LIGHT = "L", pgettext_lazy("Color Theme", "Light")
DARK = "D", pgettext_lazy("Color Theme", "Dark")
AUTO = "A", pgettext_lazy("Color Theme", "Auto")
[docs]
class ColorThemeDiscovery:
"""Resolves the effective color theme for a request.
Combines the authenticated user's saved preference with the browser's
reported theme (stored in the ``fpagenttheme`` cookie by the front-end
JavaScript) to avoid a flash of the wrong theme on page load.
"""
_COOKIE_AGENT_THEME = "fpagenttheme"
[docs]
@classmethod
def guess(cls, request) -> ColorTheme:
"""Return the best color theme for the given request.
For users with ``AUTO`` preference, falls back to the agent (browser)
theme if one is available.
Args:
request: The current HTTP request.
Returns:
ColorTheme: Resolved theme value.
"""
user_theme = cls.user_theme(request)
agent_theme = cls.agent_theme(request)
# In case of AUTO, consider the color theme preferred by the agent.
# This avoids visual flickering, when library.js sets the preferred theme.
if user_theme == ColorTheme.AUTO and agent_theme != ColorTheme.AUTO:
return agent_theme
return user_theme
[docs]
@staticmethod
def user_theme(request):
return (
ColorTheme(request.user.color_theme)
if request.user.is_authenticated
else ColorTheme.AUTO
)
[docs]
@classmethod
def agent_theme(cls, request) -> ColorTheme:
match request.COOKIES.get(cls._COOKIE_AGENT_THEME, None):
case "light":
return ColorTheme.LIGHT
case "dark":
return ColorTheme.DARK
case _:
return ColorTheme.AUTO
[docs]
class IDSequence(models.IntegerChoices):
"""Enumeration of entity types that have configurable human-readable ID sequences."""
TRANSPORT_CHAIN = 1, "TC"
CONSIGNMENT = 2, "CO"
SHIPMENT = 3, "SH"
SUPPORT_TICKET = 4, "SUP"
TREATMENT = 5, "TMT"
TREATMENT_OPERATION = 6, "TMP"
REPORT = 7, "RP"
TRANSPORT_OPERATION = 8, "TT"
COMPANY_SITE = 9, "CS"
[docs]
class Account(UUIDModel):
"""
Account names a physical location.
Attributes:
name (str): A descriptive name
internal (bool): True, if the account represents a location owned by the user, else False
address (str): Physical address
latitude (Decimal): Physical coordinates (latitude)
longitude (Decimal): Physical coordinates (longitude)
"""
class Meta:
indexes = [
models.Index(fields=["internal"]),
models.Index(fields=["name"]),
]
name = models.CharField(
max_length=255,
verbose_name=_("Name"),
)
internal = models.BooleanField(
default=False,
verbose_name=_("Internal"),
)
address = models.TextField(
default="",
verbose_name=_("Address"),
)
latitude = CoordinateDecimalField(
verbose_name=_("Latitude"),
)
longitude = CoordinateDecimalField(
verbose_name=_("Longitude"),
)
[docs]
@classmethod
def externals(cls):
return cls.objects.filter(internal=False).order_by("name")
[docs]
@classmethod
def internals(cls):
return cls.objects.filter(internal=True).order_by("name")
def __str__(self):
return self.name
[docs]
def validate(self):
errors = []
if self.address == 0:
errors.append(_("Empty address"))
if self.latitude == Decimal(0) or self.longitude == Decimal(0):
errors.append(_("Invalid coordinates"))
return errors
[docs]
class BusinessPartner(AddressMixin, CoordinatesMixin, UUIDModel):
class Meta:
verbose_name = _("Business Partner")
verbose_name_plural = _("Business Partners")
indexes = [
models.Index(fields=["name"]),
models.Index(fields=["street"]),
models.Index(fields=["city"]),
models.Index(fields=["postal_code"]),
]
name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Name"),
)
def __str__(self):
return self.name
[docs]
def get_absolute_url(self):
from django.urls import reverse
return reverse("ferrobase:businesspartner_update", kwargs={"pk": self.pk})
[docs]
@classmethod
def get_ordered(cls):
return cls.objects.order_by("name")
[docs]
@classmethod
def get_by_name(cls, name: str) -> "BusinessPartner":
return cls.objects.get(name=name)
[docs]
def save(self, *args, **kwargs):
if self.latitude == zero and self.longitude == zero:
try:
location = self.address.geocode()
self.latitude = CoordinateDecimalField.round(location.coords.lat)
self.longitude = CoordinateDecimalField.round(location.coords.lon)
except CoordinatesNotFound:
# Silently ignore coordinates not found. This should be only
# due to garbage data.
pass
except Exception as e:
logging.error("could not save coordinates: {}".format(e))
super().save(*args, **kwargs)
[docs]
class Country(ISOCatalogEntry):
"""
Country defines a country.
Since this is a ISO catalog entry, iso_code should conform to ISO codes,
although this is not enforced.
"""
class Meta:
verbose_name = _("Country")
[docs]
class LanguageManager(ISOCatalogEntryManager):
[docs]
def get_supported(self):
return self.filter(iso_code__in=Language.SUPPORTED_LANGUAGES)
[docs]
class Language(ISOCatalogEntry):
"""
Language defines a written language.
Since this is a ISO catalog entry, iso_code should conform to ISO codes,
although this is not enforced.
Attributes:
SUPPORTED_LANGUAGES: List of language codes for which translations are kept up to date.
"""
class Meta:
verbose_name = _("Language")
objects = LanguageManager()
SUPPORTED_LANGUAGES = [
"de",
"en",
]
[docs]
@classmethod
def get_supported_language(cls, iso_code: str) -> str:
"""
Get ISO code of supported language.
Args:
iso_code: Requested language. If it is a supported language, it is returned.
Returns: Supported language code. Defaults to `settings.LANGUAGE_CODE` or if that is undefined to "en".
"""
from django.conf import settings
if iso_code in cls.SUPPORTED_LANGUAGES:
return iso_code
return getattr(settings, "LANGUAGE_CODE", "en")
[docs]
class LocalUser(UUIDModel):
"""
LocalUser represents the user in tenant database.
Since User model resides in the global tenant-independent database,
tenant-dependent models cannot reference User. Instead, they must
reference LocalUser.
The 1-1 relationship from LocalUser to User is loosely enforced through
the username field.
"""
username = models.CharField(
verbose_name=_("username"),
max_length=150,
unique=True,
)
def __str__(self):
return self.username
[docs]
@classmethod
def load(cls, user: settings.AUTH_USER_MODEL) -> "LocalUser":
"""
Find LocalUser corresponding to this user. If LocalUser does not
exist, it is created implicitly.
Args:
user: Tenant-independent User model instance
Returns: LocalUser
"""
local = cls.objects.filter(username=user.username).first()
if not local:
local = cls.objects.create(username=user.username)
return local
[docs]
class Salutation(UUIDModel):
name = models.CharField(
max_length=int(CommonLimit.TINY_CHAR), verbose_name=_("Name")
)
def __str__(self):
return self.name
[docs]
@classmethod
def get_ordered(cls):
return cls.objects.order_by("name")
[docs]
class ModelIsLocked(Exception):
"""Raised when a ``LockableModelMixin`` instance is locked by another user."""
def __init__(self, user, until: datetime.datetime):
super().__init__(
"locked by %s until %s" % (str(user), until.isoformat(sep=" "))
)
[docs]
class LockableModelMixin(models.Model):
"""
Mixing adding fields to create a lockable model.
Locking means prohibiting write operations. Enforcement is done by view code.
Views must acquire locks when selecting a model. On frontend side, a web socket should be used to periodically
call to the backend to renew the lock.
"""
LOCK_DURATION = timedelta(minutes=1)
locked_until = models.DateTimeField(
verbose_name=_("Locked at"),
blank=True,
null=True,
default=None,
)
locked_by = models.ForeignKey(
LocalUser,
on_delete=models.SET_NULL,
blank=True,
null=True,
default=None,
)
def _lock_model(self, user: LocalUser):
"""
Set locked_until and locked_by fields.
Locked_until is set to current time plus LOCK_DURATION. To extend the duration, call lock_model repeatedly
(usually it should be done via acquire_lock).
Args:
user (LocalUser): User recorded in the lock.
"""
self.locked_until = now_utc() + self.LOCK_DURATION
self.locked_by = user
self.save()
[docs]
def acquire_lock(self, user: LocalUser):
"""
Lock model for given user.
If the model is already locked with a locked_until time lesser or equal to the current time,
and locked_by being unequal to the given user, the exception ModelIsLocked is thrown.
Otherwise, the model is locked via lock_model method.
Args:
user (LocalUser): User acquiring the lock
"""
is_time_locked = (
self.locked_until is not None and now_utc() <= self.locked_until
)
is_same_user = self.locked_by is not None and self.locked_by.pk == user.pk
if is_time_locked and not is_same_user:
# noinspection PyTypeChecker
raise ModelIsLocked(self.locked_by, self.locked_until)
self._lock_model(user)
[docs]
class MaterialManager(models.Manager):
"""Manager for ``Material`` with a bulk upsert keyed on ``code``."""
[docs]
def upsert(self, records):
return self.bulk_create(
records,
update_conflicts=True,
update_fields=["name"],
unique_fields=["code"],
)
[docs]
class Material(SelectOptionsMixin, UUIDModel):
"""
Material represents a product.
Attributes:
code (str): A short identifier (free-form)
name (str): A descriptive name
"""
class Meta:
verbose_name = _("Material")
indexes = [
models.Index(
fields=["name"],
name="fb_material_name",
)
]
objects = MaterialManager()
code = models.CharField(
verbose_name=_("Code"),
max_length=100,
unique=True,
)
name = models.CharField(
verbose_name=_("Name"),
max_length=255,
)
def __str__(self):
return "%s - %s" % (self.code, self.name)
[docs]
@classmethod
def order_by_field(cls):
return "code"
[docs]
def get_absolute_url(self):
return reverse("ferrobase:material_update", kwargs={"pk": self.pk})
[docs]
class CompanySite(SelectOptionsMixin, AddressMixin, UUIDModel):
"""Physical site of the tenant's company (warehouse, production facility, etc.).
Access can be restricted to specific ``LocalUser`` instances via
``allowed_users``. A user who is not in ``allowed_users`` cannot see or
use this site in the UI.
"""
class Meta:
verbose_name = _("Company Site")
verbose_name_plural = _("Company Sites")
indexes = [
models.Index(
fields=["name"],
name="fb_storage_location_name",
)
]
code = models.CharField(
verbose_name=_("Code"),
unique=True,
)
name = models.CharField(
verbose_name=_("Name"),
)
allowed_users = models.ManyToManyField(
LocalUser,
blank=True,
related_name="company_sites",
verbose_name=_("Users"),
)
def __str__(self):
return "%s - %s" % (self.code, self.name)
[docs]
@classmethod
def options_for_user(cls, user: LocalUser, only_enabled: bool = None):
if only_enabled:
return user.company_sites.all().values_list("pk", "name")
enabled_locations = [location.pk for location in user.company_sites.all()]
return [
(
str(location.pk),
str(location),
location.pk in enabled_locations,
)
for location in cls.objects.order_by("code")
]
[docs]
@classmethod
def set_locations_for_user(cls, location_ids: list, user: LocalUser):
"""
Replace locations set for user with new ones.
"""
locations = [cls.objects.get(pk=location_id) for location_id in location_ids]
user.company_sites.set(locations, clear=True)
[docs]
def get_absolute_url(self):
return reverse("ferrobase:companysite_update", kwargs={"pk": self.pk})
[docs]
def is_user_allowed(self, user: LocalUser):
return self.allowed_users.filter(pk=user.pk).exists()
[docs]
class TimeZoneManager(models.Manager):
"""Manager for ``TimeZone`` that provides a conflict-safe bulk upsert."""
[docs]
def upsert(self, objects):
self.bulk_create(objects, ignore_conflicts=True)
[docs]
class TimeZone(UUIDModel):
"""IANA timezone identifier stored per-tenant for user preferences."""
objects = TimeZoneManager()
identifier = models.CharField(
verbose_name=_("Timezone identifier"),
max_length=int(CommonLimit.TINY_CHAR),
unique=True,
)
def __str__(self):
return self.identifier
[docs]
class IdSequenceConfig(UUIDModel):
"""Configuration for auto-generating human-readable sequential IDs.
Each ``IDSequence`` entry can have its own prefix and zero-padding amount
so that, for example, transport chains get IDs like ``TC-0001234``.
"""
class Meta:
verbose_name = _("ID Sequence")
verbose_name_plural = _("ID Sequences")
name = models.PositiveIntegerField(
unique=True,
choices=IDSequence.choices,
verbose_name=_("Sequence Name"),
)
prefix = models.CharField(
max_length=10,
verbose_name=_("Prefix"),
)
padding_amount = models.PositiveIntegerField(
default=7,
verbose_name=_("Padding Amount"),
)
def __str__(self):
return self.name
[docs]
def get_absolute_url(self):
return reverse("ferrobase:idsequenceconfig_update", args=[self.pk])