# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
from abc import ABC, abstractmethod
from itertools import groupby
from typing import Generator, Iterable
from uuid import UUID
from ferrosoft.apps.ferrobase.models import (
builtin,
SubscriptionPlan,
FeatureAssignment,
AppFeature,
)
from ferrosoft.apps.ferrobase.models.units import (
MeasurementUnit,
UnitCursor,
UnitRef,
UnitSystem,
)
from ferrosoft.apps.ferrobase.models.shared import Country, Language, TimeZone
[docs]
class BaseBackend(ABC):
model = None
save_to_master_db_only = False
objects = []
[docs]
def get_manager(self, db_name: str | None, model=None):
model = model or self.model
if model is None:
raise RuntimeError("Model class is not defined")
if isinstance(db_name, str):
return model.objects.db_manager(db_name)
return model.objects
[docs]
def save_to_database(self, db_name: str | None = None):
if db_name is not None and self.save_to_master_db_only:
return []
return self.get_manager(db_name).upsert(self.objects)
[docs]
class AppFeatureBackend(BaseBackend):
model = AppFeature
objects = builtin.APP_FEATURES.values()
save_to_master_db_only = True
[docs]
class CountryBackend(BaseBackend):
model = Country
objects = builtin.COUNTRIES
[docs]
class LanguageBackend(BaseBackend):
model = Language
objects = builtin.LANGUAGES
save_to_master_db_only = True
[docs]
class SubscriptionPlanRegistry(BaseBackend):
model = SubscriptionPlan
objects = builtin.SUBSCRIPTION_PLANS
save_to_master_db_only = True
def __init__(self):
super().__init__()
self.lookup = {obj.plan.pk: obj.plan for obj in self.objects}
[docs]
def get(self, pk: UUID) -> SubscriptionPlan | None:
return self.lookup.get(pk, None)
[docs]
def save_to_database(self, db_name: str | None = None):
if db_name is not None:
return
self.get_manager(db_name).upsert(map(lambda obj: obj.plan, self.objects))
assignments = [
assignment.with_plan(obj.plan)
for obj in self.objects
for assignment in obj.features
]
self.get_manager(db_name, FeatureAssignment).upsert(assignments)
[docs]
class UnitRegistry(BaseBackend):
"""Unit registry provides all supported measurement units.
It is storing units in-memory to avoid hitting database for lookups by ID or
unit cursor."""
model = MeasurementUnit
def __init__(self):
self.units = builtin.MEASUREMENT_UNITS
self.lookup_by_cursor = {
system: {unit.symbol: unit for unit in units}
for system, units in groupby(self.units, key=lambda unit: unit.system)
}
self.lookup_by_id = {unit.id: unit for unit in self.units}
[docs]
def save_to_database(self, db_name: str | None = None):
self.get_manager(db_name).upsert(self.units)
[docs]
def exists(self, needle: UnitRef) -> bool:
return self.get(needle) is not None
[docs]
def find(self, cursors: Iterable[UnitCursor]) -> Iterable[MeasurementUnit]:
return filter(
lambda value: value is not None,
map(
lambda cursor: self.lookup_by_cursor.get(cursor.system, {}).get(
cursor.symbol, None
),
cursors,
),
)
[docs]
def get(self, needle: UnitRef) -> MeasurementUnit:
if isinstance(needle, MeasurementUnit):
return needle
elif isinstance(needle, str):
unit = self.lookup_by_cursor.get(UnitSystem.METRIC, {}).get(needle, None)
elif isinstance(needle, UnitCursor):
unit = self.lookup_by_cursor.get(needle.system, {}).get(needle.symbol, None)
elif isinstance(needle, UUID):
unit = self.lookup_by_id.get(needle, None)
else:
raise ValueError("Cannot get measurement unit by %s" % type(needle))
if unit is None:
raise ValueError("Measurement unit is undefined: %s" % needle)
return unit
[docs]
class TimeZoneBackend(BaseBackend):
model = TimeZone
objects = builtin.TIMEZONES
save_to_master_db_only = True
[docs]
class UnifiedBackend(BaseBackend):
"""Model registries act as an in-memory storage for model instances.
This is used for models of fixed set size, which are defined once, such as
measurement units."""
AppFeature = AppFeatureBackend()
Country = CountryBackend()
Language = LanguageBackend()
SubscriptionPlan = SubscriptionPlanRegistry()
TimeZone = TimeZoneBackend()
Unit = UnitRegistry()
[docs]
def save_to_database(self, db_name: str | None = None):
for backend in self.backends():
backend.save_to_database(db_name)
[docs]
def backends(self) -> Generator[BaseBackend, None, None]:
for prop in dir(self):
if prop.startswith("_"):
continue
if isinstance(getattr(self, prop), BaseBackend):
yield getattr(self, prop)
ModelRegistry = UnifiedBackend()