Trixy7/trixy_core/service/scopes/singleton.py

# -*- coding: utf-8 -*-
"""
Singleton-Scope für Services.

Erstellt genau eine Instanz pro Service, die während der
gesamten Lebensdauer der Anwendung wiederverwendet wird.
"""

from __future__ import annotations

import threading
from typing import Callable, Generic, TypeVar

from trixy_core.service.scopes.scope import ServiceScope, ScopeType


T = TypeVar("T")


class SingletonScope(ServiceScope[T], Generic[T]):
    """
    Singleton-Scope für Services.

    Garantiert, dass nur eine Instanz eines Services existiert.
    Thread-safe durch Locking.

    Example:
        scope = SingletonScope(lambda: MyService())
        instance1 = scope.get_instance("my_service")
        instance2 = scope.get_instance("my_service")
        assert instance1 is instance2
    """

    scope_type = ScopeType.SINGLETON

    def __init__(self, factory: Callable[[], T]) -> None:
        """
        Initialisiert den Singleton-Scope.

        Args:
            factory: Factory-Funktion zur Instanz-Erstellung.
        """
        super().__init__(factory)
        self._instance: T | None = None
        self._lock = threading.Lock()
        self._initialized = False

    @property
    def has_instance(self) -> bool:
        """Prüft, ob bereits eine Instanz erstellt wurde."""
        return self._initialized

    @property
    def instance(self) -> T | None:
        """Gibt die Instanz zurück (ohne sie zu erstellen)."""
        return self._instance

    def get_instance(self, key: str) -> T:
        """
        Holt oder erstellt die Singleton-Instanz.

        Thread-safe durch Double-Checked Locking.

        Args:
            key: Service-Schlüssel (wird für Singleton ignoriert).

        Returns:
            Die Singleton-Instanz.
        """
        if self._initialized:
            return self._instance  # type: ignore

        with self._lock:
            if not self._initialized:
                self._instance = self._factory()
                self._initialized = True

        return self._instance  # type: ignore

    def reset(self) -> None:
        """
        Setzt den Singleton zurück.

        Die nächste Anfrage erstellt eine neue Instanz.
        """
        with self._lock:
            if self._instance is not None:
                # Dispose aufrufen falls vorhanden
                if hasattr(self._instance, "dispose"):
                    try:
                        self._instance.dispose()
                    except Exception:
                        pass
                elif hasattr(self._instance, "close"):
                    try:
                        self._instance.close()
                    except Exception:
                        pass

            self._instance = None
            self._initialized = False

    def dispose(self) -> None:
        """Disposed den Singleton."""
        self.reset()


class LazySingletonScope(SingletonScope[T], Generic[T]):
    """
    Lazy-Singleton-Scope.

    Erstellt die Instanz erst beim ersten Zugriff.
    Identisch mit SingletonScope, aber mit explizitem Namen.
    """

    pass


class EagerSingletonScope(SingletonScope[T], Generic[T]):
    """
    Eager-Singleton-Scope.

    Erstellt die Instanz sofort bei der Scope-Erstellung.
    """

    def __init__(self, factory: Callable[[], T]) -> None:
        """
        Initialisiert den Eager-Singleton-Scope.

        Args:
            factory: Factory-Funktion zur Instanz-Erstellung.
        """
        super().__init__(factory)
        # Instanz sofort erstellen
        self._instance = self._factory()
        self._initialized = True