Source code for ferrosoft.apps.ferrobase.services.update

#  Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
from abc import ABC, abstractmethod
from logging import Logger
from typing import Callable, List

from django.contrib.auth.models import Group, Permission
from django.utils.translation import gettext as _

from ferrosoft.apps.ferrobase.auth.roles import Registry
from ferrosoft.apps.ferrobase.models import Tenant
from ferrosoft.apps.ferrobase.models.registry import ModelRegistry
from ferrosoft.apps.ferrobase.tenant.context import active_tenant




class StatusWriter(ABC):
    """Writer used by updater functions to send success or error messages."""



    @abstractmethod
    def error(self, message: str):
        """Report an error encountered by an updater."""




    @abstractmethod
    def notice(self, message: str):
        """Report informational progress from an updater."""




    @abstractmethod
    def success(self, message: str):
        """Report a successful updater step."""






class WriterNoop(StatusWriter):
    """Discard all status messages (useful when running updaters silently)."""



    def error(self, message: str):
        """Drop the message."""




    def notice(self, message: str):
        """Drop the message."""




    def success(self, message: str):
        """Drop the message."""






class LoggingStatusWriter(StatusWriter):
    """Status writer forwarding messages to a :class:`logging.Logger`."""

    def __init__(self, logger: Logger):
        """Bind the writer to the destination ``logger``."""
        self.logger = logger



    def error(self, message: str):
        """Log ``message`` at ``ERROR`` level."""
        self.logger.error(message)




    def notice(self, message: str):
        """Log ``message`` at ``INFO`` level."""
        self.logger.info(message)




    def success(self, message: str):
        """Log ``message`` at ``INFO`` level."""
        self.logger.info(message)






class Updater:
    """Update database resources with data from current release.

    By default, only Ferrobase resources are updated. Apps need to
    call add_global_updater/add_tenant_updater to register their updaters."""

    _global_updaters: List[Callable[[StatusWriter, Tenant | None], None]] = []
    _tenant_updaters: List[Callable[[StatusWriter, Tenant | None], None]] = []



    @classmethod
    def add_global_updater(cls, *updaters: Callable[[StatusWriter], None]):
        """Register updater callables to run against the master database."""
        cls._global_updaters += updaters




    @classmethod
    def add_tenant_updater(cls, *updaters: Callable[[StatusWriter], None]):
        """Register updater callables to run inside each tenant database."""
        cls._tenant_updaters += updaters




    @classmethod
    def init(cls):
        """Seed the default updater registries on first call.

        Idempotent: re-invocations do nothing if updaters are already
        registered, so apps that add their own updaters before import are
        not overwritten.
        """
        if len(cls._global_updaters) == 0:
            cls._global_updaters = [
                create_registered_models,
                create_permission_sets,
            ]

        if len(cls._tenant_updaters) == 0:
            cls._tenant_updaters = [create_registered_models]




    def run(self, writer: StatusWriter):
        """Run all registered updaters: globals first, then per-tenant updaters."""
        self.run_global_updaters(writer)

        for tenant in Tenant.all_ordered():
            with active_tenant(tenant):
                writer.notice(_("Updating tenant %(name)s") % {"name": str(tenant)})
                self.run_tenant_updaters(writer, tenant)




    def run_global_updaters(self, writer: StatusWriter):
        """Invoke each registered global updater with ``tenant=None``."""
        for updater in self._global_updaters:
            updater(writer, None)




    def run_tenant_updaters(self, writer: StatusWriter, tenant: Tenant):
        """Invoke each registered tenant updater bound to ``tenant``."""
        for updater in self._tenant_updaters:
            updater(writer, tenant)






def create_permission_sets(writer: StatusWriter, tenant: Tenant | None):
    """Sync Django groups + permissions from the role registry.

    Only runs in the master database (``tenant is None``); permission sets
    are global and not duplicated per tenant.
    """
    if tenant is not None:
        # No need to create permission sets in tenant databases.
        return

    for perm_set in Registry.get_permission_sets().values():
        group, created = Group.objects.get_or_create(
            defaults={"name": perm_set.group}, name=perm_set.group
        )
        if created:
            writer.success(_("Created group %(name)s") % {"name": str(group)})

        permissions = list(Permission.objects.filter(codename__in=perm_set.permissions))
        # noinspection PyUnresolvedReferences
        group.permissions.set(permissions)
        writer.success(
            _("Updated permissions for group %(name)s") % {"name": str(group)}
        )





def create_registered_models(writer: StatusWriter, tenant: Tenant | None):
    """Persist the in-memory :class:`ModelRegistry` to the target database."""
    ModelRegistry.save_to_database(
        db_name=None if tenant is None else tenant.database_name
    )
    writer.success(_("Created registered models"))



Updater.init()