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

"""Tenant provisioning: PostgreSQL role/database creation, migrations and fixtures."""

import logging
import secrets
import string
import uuid
from dataclasses import dataclass
from datetime import datetime

from django.core import management
from django.db import connection
from uuid_extensions import uuid7

from ferrosoft.apps.emiflow.catalog.database import import_builtin_fixtures
from ferrosoft.apps.ferrobase.models import Tenant, Organization
from ferrosoft.apps.ferrobase.services.update import Updater, LoggingStatusWriter
from ferrosoft.apps.ferrobase.tenant.context import active_tenant
from ferrosoft.ops import sql
from ferrosoft.ops.sql import tenant_database_name

LANG_CODE_PLACEHOLDER = "LANGUAGE_CODE"
SETTINGS_DATABASE_HOST = "FERROSOFT_TENANT_DATABASE_HOST"
DEFAULT_DATABASE_PORT = 5432
STANDARD_FIXTURES = [
    "ferrobase/%s/directories" % LANG_CODE_PLACEHOLDER,
    "ferrobase/%s/material" % LANG_CODE_PLACEHOLDER,
    "emiflow/default-settings",
]


logger = logging.getLogger(__name__)




@dataclass
class TenantOptions:
    id: str | None
    """ID of the tenant or None to autogenerate"""

    name: str
    active: bool
    expiry_date: datetime | None
    language: str
    """Two letter language code"""
    organization: Organization
    """Organization to add tenant to"""
    fixtures: list[str]
    """Fixtures to import. The string LANGUAGE_CODE is replaced with language code"""

    db_name: str | None = None
    """Override default database name (not tested)."""
    db_host: str | None = None
    """Override default database host (not tested)."""
    db_username: str | None = None
    """Override default database username (not tested)."""
    db_port: str | None = None
    """Override default database port (not tested)."""
    db_password: str | None = None
    """Override default database password (not tested)."""



def _subst_fixture_placeholders(fixtures: list, language: str) -> list:
    """Replace ``LANGUAGE_CODE`` in fixture paths with ``language``."""
    return [item.replace(LANG_CODE_PLACEHOLDER, language) for item in fixtures]

def _random_password():
    """Return a 36-character alphanumeric password from :mod:`secrets`."""
    alphabet = string.ascii_letters + string.digits
    return ''.join(secrets.choice(alphabet) for i in range(36))




def add_tenant(options: TenantOptions) -> Tenant:
    """Provision a new tenant database and register it in the master DB.

    Must be called with the default (master) database connection active. The
    function:

    1. Inserts the :class:`Tenant` row as inactive so subsequent failures do
       not yield a half-provisioned, user-visible tenant.
    2. Creates a PostgreSQL role and a tenant-specific database owned by it.
    3. Runs ``migrate`` and the configured fixtures inside the new database.
    4. Runs the registered tenant updaters and built-in fixture imports.
    5. Flips the tenant's ``active`` flag to ``options.active`` once
       everything succeeded.

    Args:
        options: Provisioning parameters; an empty ``id`` triggers
            UUIDv7 generation, and an empty ``db_password`` causes a
            random 36-char password to be generated.

    Returns:
        Tenant: The newly created tenant record.
    """
    from django.conf import settings

    default_db_host = getattr(settings, SETTINGS_DATABASE_HOST, "localhost")

    with connection.cursor() as cursor:
        # The default connection is the master database, and the user is
        # expected to create users.
        cursor.execute("SELECT session_user")
        result = cursor.fetchone()
        connection_role = result[0]

        # Accept user-supplied ID or generate it.
        tenant_id = uuid7() if not options.id else uuid.UUID(options.id)

        # While the options allow custom db_xxx values, it has not been tested.
        # Currently only the defaults are known to work.
        db_name = options.db_name or tenant_database_name(tenant_id)

        tenant = Tenant.objects.create(
            id=tenant_id,
            # Create inactive, so tenant is not usable if any subsequent commands fails.
            active=False,
            name=options.name,
            organization=options.organization,
            expiry_date=options.expiry_date,
            database_hostname=options.db_host or default_db_host,
            database_username=options.db_username or db_name,
            database_port=options.db_port or DEFAULT_DATABASE_PORT,
            database_password=options.db_password or _random_password(),
            database_name=db_name,
        )

        cursor.execute(
            sql.create_tenant_role_command(
                tenant.database_username,
                tenant.database_password,
                connection_role,
            )
        )

        cursor.execute(
            sql.create_tenant_database_command(
                tenant.database_name,
                tenant.database_username,
            )
        )

        with active_tenant(tenant):
            management.call_command(
                "migrate",
                database=tenant.database_name,
                no_input=True,
            )

            fixtures = _subst_fixture_placeholders(options.fixtures, options.language)
            management.call_command("loaddata", fixtures, database=tenant.database_name)
            Updater().run_tenant_updaters(LoggingStatusWriter(logger), tenant)
            import_builtin_fixtures(options.language)

        # Activate tenant only after all commands succeeded, so that the user
        # cannot visit the page using half-finished tenant.
        tenant.active = options.active
        tenant.save(update_fields=["active"])
        return tenant