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

from abc import ABC, abstractmethod

from psycopg import sql

from ferrosoft.apps.ferrobase.models import IdSequenceConfig, IDSequence
from ferrosoft.apps.ferrobase.tenant.utils import tenant_db_connection




class IdGenerator(ABC):
    """Abstract sequence-based identifier generator.

    Concrete subclasses bind a sequence to a backing store and produce
    formatted IDs combining the configured prefix and zero-padded sequence
    number.
    """



    @classmethod
    @abstractmethod
    def create(cls, name: IDSequence) -> "IdGenerator":
        """Create or load the generator for the sequence identified by ``name``."""
        raise NotImplementedError




    @abstractmethod
    def next_id(self) -> str:
        """Return the next formatted identifier from the underlying sequence."""
        raise NotImplementedError




    @classmethod
    def get_instance(cls, name: IDSequence):
        """Return the default :class:`IdGeneratorPostgres` for ``name``."""
        return IdGeneratorPostgres.create(name)






class IdGeneratorPostgres(IdGenerator):
    """PostgreSQL-backed ID generator using ``CREATE SEQUENCE``."""

    def __init__(self, config: IdSequenceConfig):
        """Bind the generator to the persisted :class:`IdSequenceConfig`."""
        self.config = config

    @classmethod
    def _sequence_name(cls, name: str) -> str:
        """Return the canonical sequence name ``ferrobase_idseq_<name_lower>``."""
        return "ferrobase_idseq_%s" % name.lower()



    @classmethod
    def create(cls, name: IDSequence) -> "IdGeneratorPostgres":
        """Persist a configuration if missing and ensure the SQL sequence exists.

        The :class:`IdSequenceConfig` row is created in the master database
        with a default prefix equal to ``name.label`` and seven digits of
        padding.  The corresponding PostgreSQL sequence is created in the
        tenant database via ``CREATE SEQUENCE IF NOT EXISTS``.
        """
        sequence_name = cls._sequence_name(name.name)
        config = IdSequenceConfig.objects.filter(name=name.value).first()
        if not config:
            config = IdSequenceConfig.objects.create(
                name=name, prefix=name.label, padding_amount=7
            )

        with tenant_db_connection().cursor() as cursor:
            query = sql.SQL(
                """CREATE SEQUENCE IF NOT EXISTS {}
                    START 1
                    INCREMENT 1
                    MINVALUE 1
                    CACHE 1;
                """
            ).format(sql.Identifier(sequence_name))
            cursor.execute(query)

        return cls(config)




    def next_id(self) -> str:
        """Allocate the next sequence value and format it as ``prefix + padded_number``."""
        sequence_name = self._sequence_name(IDSequence(self.config.name).name)
        with tenant_db_connection().cursor() as cursor:
            cursor.execute("SELECT nextval(%s)", [sequence_name])
            next_val = cursor.fetchone()[0]

        return "%s%s" % (
            self.config.prefix,
            str(next_val).zfill(self.config.padding_amount),
        )