import hashlib
import os
import uuid
from datetime import datetime, timedelta
from decimal import Decimal
from enum import Enum
from functools import wraps
from typing import Tuple
from uuid import UUID
from django.db import models
from django.db.models import QuerySet
from django.utils.translation import gettext_lazy as _
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.ferrobase.models.fields import CoordinateDecimalField
from ferrosoft.apps.ferromaps.services.addresses import Address
[docs]
class PrefetchedFieldMissing(RuntimeError):
"""Raised when a method requires a prefetched related field that was not loaded."""
def __init__(self, field_name: str):
super().__init__("Prefetched field %s is missing" % field_name)
[docs]
def prefetched_field(field_name: str, expected_type):
"""Decorator that guards a method behind a prefetch-loaded related field.
Raises ``PrefetchedFieldMissing`` when the named attribute on the instance
is not an instance of ``expected_type``, which typically means the queryset
did not call ``prefetch_related`` before invoking the method.
Args:
field_name: Attribute name to inspect on the model instance.
expected_type: Type the attribute must be an instance of.
Returns:
Callable: Decorated function that validates the prefetched field before
delegating to the original function.
"""
def inner(function):
@wraps(function)
def wrapper(self, *args, **kwargs):
value = getattr(self, field_name, None)
if isinstance(value, expected_type):
return function(value)
raise PrefetchedFieldMissing(field_name)
return wrapper
return inner
[docs]
class CommonLimit(Enum):
"""
Defines common limits for VARCHAR columns.
Its usage is DISCOURAGED, because PostgreSQL does not require an arbitrary
limit on VARCHAR columns. There is also no performance benefit.
"""
TINY_CHAR = 100
NORMAL_CHAR = 1000
BIG_CHAR = 2000
def __int__(self):
return self.value
[docs]
class UUIDModel(models.Model):
"""Abstract base model with a UUID v4 primary key.
All concrete models in Ferrobase should inherit from this class rather than
Django's default integer-keyed ``Model`` so that primary keys are
opaque, globally unique, and safe to expose in URLs.
"""
id = models.UUIDField(
primary_key=True,
default=uuid.uuid4,
editable=False,
)
[docs]
@classmethod
def get_by_id(cls, pk: UUID):
return cls.objects.get(pk=pk)
[docs]
class AccessKeyMixin(UUIDModel):
"""Mixin that adds a time-limited, single-use access key to a model.
The key is a SHA-256 hex digest generated from ``settings.SECRET_KEY``
plus OS random bytes, stored in ``access_key``. It expires after
``_DEFAULT_VALIDITY`` (1 day by default). Used for password-reset and
account-activation tokens.
"""
_RANDOM_INPUT_LENGTH = 512
"""How many bytes to read from the OS random source when generating the key."""
_DEFAULT_VALIDITY = timedelta(days=1)
"""Lifetime of a newly generated access key."""
access_key = models.CharField(
max_length=64,
unique=True,
verbose_name=_("Access Key"),
)
access_key_expiry = models.DateTimeField(
verbose_name=_("Access key expiry"),
)
[docs]
@classmethod
def new_access_key_params(cls, expiry: datetime | None = None) -> dict:
"""Generate a new access key and expiry date."""
from django.conf import settings
valid_until = expiry or (now_utc() + cls._DEFAULT_VALIDITY)
h = hashlib.new("sha256")
# Add salt
h.update(settings.SECRET_KEY.encode())
h.update(os.urandom(cls._RANDOM_INPUT_LENGTH))
key = h.hexdigest()
return {
"access_key": key,
"access_key_expiry": valid_until,
}
[docs]
@classmethod
def get_if_not_expired(cls, access_key: str):
obj = cls.objects.filter(access_key=access_key).first()
return obj if obj and obj.access_key_expiry > now_utc() else None
[docs]
class AddressMixin(UUIDModel):
"""
Add address fields to a model.
"""
salutation = models.ForeignKey(
"ferrobase.Salutation",
on_delete=models.RESTRICT,
verbose_name=_("Salutation"),
blank=True,
null=True,
)
first_name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("First name"),
blank=True,
default="",
)
last_name = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Last name"),
blank=True,
default="",
)
company = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Company"),
blank=True,
default="",
)
additional_lines = models.TextField(
verbose_name=_("Additional lines"),
blank=True,
default="",
)
street = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("Street"),
blank=True,
null=True,
)
house_no = models.CharField(
max_length=int(CommonLimit.TINY_CHAR),
verbose_name=_("House no."),
blank=True,
default="",
null=True,
)
city = models.CharField(
max_length=int(CommonLimit.NORMAL_CHAR),
verbose_name=_("City"),
blank=True,
null=True,
)
postal_code = models.CharField(
max_length=int(CommonLimit.TINY_CHAR),
verbose_name=_("Postal code"),
blank=True,
null=True,
)
country = models.ForeignKey(
"ferrobase.Country",
on_delete=models.RESTRICT,
verbose_name=_("Country"),
blank=True,
null=True,
)
@property
def address(self):
return Address(
first_name=self.first_name,
last_name=self.last_name,
salutation=str(self.salutation or ""),
company=self.company,
additional_lines=self.additional_lines,
street=self.street,
house_no=self.house_no,
city=self.city,
postal_code=self.postal_code,
country="" if self.country is None else self.country.english_name,
)
[docs]
def address_lines(self):
self.address.format_lines()
[docs]
class ReadOnlyMixin(UUIDModel):
"""Models can be marked read-only to prevent updates and deletions.
Generic views try to get this attribute from model instances to determine
whether to enable certain functionality."""
read_only = models.BooleanField(
verbose_name=_("Read-only"),
blank=True,
default=False,
)
[docs]
class CoordinatesMixin(UUIDModel):
latitude = CoordinateDecimalField(
verbose_name=_("Latitude"),
)
longitude = CoordinateDecimalField(
verbose_name=_("Longitude"),
)
@property
def coordinates(self) -> Tuple[Decimal, Decimal]:
return self.latitude, self.longitude
[docs]
class IntegerChoicesMixin(models.IntegerChoices):
"""Additional methods for IntegerChoices."""
[docs]
@classmethod
def do_parse(cls, value):
"""Parse value as an instance of the IntegerChoices class or raise ValueError."""
if isinstance(value, int):
return cls(value)
elif isinstance(value, str):
if value == "":
raise ValueError("Empty string")
return cls(int(value))
else:
raise ValueError("Invalid value: %s" % type(value))
[docs]
@classmethod
def try_parse(cls, value, default=None):
"""Parse value as an instance of the IntegerChoices class or return None."""
try:
return cls.do_parse(value)
except ValueError:
return default
[docs]
class ISOCatalogEntryManager(models.Manager):
"""Manager for ``ISOCatalogEntry`` models.
Returns querysets ordered by ``iso_code`` by default and provides an
``upsert`` method for bulk-synchronising catalog data from release fixtures.
"""
[docs]
def get_queryset(self):
return super().get_queryset().order_by("iso_code")
[docs]
def upsert(self, objects):
self.bulk_create(
objects,
update_conflicts=True,
update_fields=[
"english_name",
"localized_name",
],
unique_fields=["iso_code"],
)
[docs]
class ISOCatalogEntry(UUIDModel):
"""Abstract base for ISO-coded catalog entities (countries, languages, etc.).
Subclasses store an ``iso_code`` (unique), an ``english_name``, and a
``localized_name``. The manager keeps entries sorted by ``iso_code``.
"""
[docs]
class Meta:
abstract = True
objects = ISOCatalogEntryManager()
iso_code = models.CharField(
verbose_name=_("ISO Code"),
max_length=int(CommonLimit.TINY_CHAR), # see remark in Language
unique=True,
)
english_name = models.CharField(
verbose_name=_("Name (English)"),
max_length=int(CommonLimit.NORMAL_CHAR),
)
localized_name = models.CharField(
verbose_name=_("Name (localized)"),
max_length=int(CommonLimit.NORMAL_CHAR),
)
def __str__(self):
return "%s - %s" % (self.iso_code, self.localized_name)
[docs]
@classmethod
def get_by_iso_code(cls, code: str):
return cls.objects.get(iso_code=code.upper())
[docs]
@classmethod
def get_ordered(cls) -> QuerySet:
return cls.objects.order_by("english_name")
[docs]
@classmethod
def id_mapping(cls) -> dict:
"""
Create mapping from ISO code to internal ID.
Returns: dict
"""
return {
country["iso_code"]: country["id"]
for country in cls.objects.values("id", "iso_code")
}
[docs]
class ObjectLookupCache:
"""In-memory lookup cache for a queryset of model instances.
Builds a dictionary indexed by a composite key derived from each object
via ``key_parts_func``. Useful for bulk import operations that need fast
repeated lookups without hitting the database on every iteration.
"""
[docs]
@staticmethod
def lookup_key(key_parts):
return "-".join([str(it) for it in key_parts])
def __init__(self, queryset, key_parts_func):
self.key_parts_func = key_parts_func
self.objects = {self.lookup_key(key_parts_func(obj)): obj for obj in queryset}
def __getitem__(self, key):
if isinstance(key, tuple):
return self.objects.get(self.lookup_key(key))
elif isinstance(key, str):
return self.objects.get(key)
else:
raise ValueError("Unsupported key type %s" % type(key))
[docs]
def get(self, key, default=None):
try:
return self[key]
except KeyError:
return default
[docs]
def update(self, objects):
for obj in objects:
self.objects[self.lookup_key(self.key_parts_func(obj))] = obj
[docs]
class SelectOptionsMixin:
"""Mixin that adds a ``select_options`` class method for use in HTML select widgets.
Returns a list of ``(pk_str, label)`` tuples ordered by the field named in
``order_by_field`` (defaults to ``"name"``).
"""
[docs]
@classmethod
def select_options(cls):
return [
(str(obj.pk), str(obj))
for obj in cls.objects.order_by(cls.order_by_field())
]
[docs]
@classmethod
def order_by_field(cls):
return "name"