Scoping and Lifecycle Management

SINGLETON: _SingletonScopeId
    """
    Scope where a single instance is created and shared across all requests.
    Default scope for implicit bindings.
    """

PROTOTYPE: _PrototypeScopeId
    """
    Scope where a new instance is created for each request.
    """

class Scope(object):
    def provide(self, binding_key, default_provider_fn):
        """
        Provides an instance according to the scope's lifecycle policy.

        Parameters:
        - binding_key: unique identifier for the binding
        - default_provider_fn: function that creates new instances

        Returns:
        Instance according to scope policy (cached, new, etc.)
        """

class SingletonScope(object):
    def provide(self, binding_key, default_provider_fn):
        """
        Returns cached instance if exists, creates and caches new instance otherwise.
        Thread-safe implementation using re-entrant locks.
        """

class PrototypeScope(object):
    def provide(self, binding_key, default_provider_fn):
        """
        Always returns a new instance by calling default_provider_fn().
        """

import pinject

class DatabaseConnection(object):
    def __init__(self):
        self.connection_id = id(self)
        print(f"Creating connection {self.connection_id}")

class ServiceA(object):
    def __init__(self, database_connection):
        self.db = database_connection

class ServiceB(object):
    def __init__(self, database_connection):
        self.db = database_connection

obj_graph = pinject.new_object_graph()

# Both services get the same connection instance (singleton is default)
service_a = obj_graph.provide(ServiceA)  # "Creating connection ..."
service_b = obj_graph.provide(ServiceB)  # No additional creation message

print(service_a.db.connection_id == service_b.db.connection_id)  # True

import pinject

class RequestHandler(object):
    def __init__(self):
        self.request_id = id(self)

class MyBindingSpec(pinject.BindingSpec):
    def configure(self, bind):
        # Each request gets a new handler instance
        bind('request_handler').to_class(RequestHandler, in_scope=pinject.PROTOTYPE)

class WebService(object):
    def __init__(self, request_handler):
        self.handler = request_handler

obj_graph = pinject.new_object_graph(binding_specs=[MyBindingSpec()])

# Each service gets a different handler instance
service1 = obj_graph.provide(WebService)
service2 = obj_graph.provide(WebService)

print(service1.handler.request_id == service2.handler.request_id)  # False

import pinject
import threading
import time

class ThreadLocalScope(pinject.Scope):
    def __init__(self):
        self._local = threading.local()
    
    def provide(self, binding_key, default_provider_fn):
        if not hasattr(self._local, 'instances'):
            self._local.instances = {}
        
        if binding_key not in self._local.instances:
            self._local.instances[binding_key] = default_provider_fn()
        
        return self._local.instances[binding_key]

class SessionData(object):
    def __init__(self):
        self.thread_id = threading.current_thread().ident
        self.created_at = time.time()

THREAD_LOCAL = "thread_local"

class ScopingBindingSpec(pinject.BindingSpec):
    @pinject.provides('session_data', in_scope=THREAD_LOCAL)
    def provide_session_data(self):
        return SessionData()

obj_graph = pinject.new_object_graph(
    binding_specs=[ScopingBindingSpec()],
    id_to_scope={THREAD_LOCAL: ThreadLocalScope()}
)

class UserService(object):
    def __init__(self, session_data):
        self.session = session_data

# Same thread gets same session, different threads get different sessions
user_service = obj_graph.provide(UserService)
print(f"Session for thread {user_service.session.thread_id}")

import pinject
import weakref

class WeakReferenceScope(pinject.Scope):
    """Custom scope that holds weak references to instances."""
    
    def __init__(self):
        self._instances = {}
    
    def provide(self, binding_key, default_provider_fn):
        # Check if we have a live weak reference
        if binding_key in self._instances:
            instance = self._instances[binding_key]()
            if instance is not None:
                return instance
        
        # Create new instance and store weak reference
        instance = default_provider_fn()
        self._instances[binding_key] = weakref.ref(instance)
        return instance

class CacheService(object):
    def __init__(self):
        self.data = {}
        print("CacheService created")

WEAK_REF_SCOPE = "weak_reference"

class CacheBindingSpec(pinject.BindingSpec):
    def configure(self, bind):
        bind('cache_service').to_class(
            CacheService, 
            in_scope=WEAK_REF_SCOPE
        )

obj_graph = pinject.new_object_graph(
    binding_specs=[CacheBindingSpec()],
    id_to_scope={WEAK_REF_SCOPE: WeakReferenceScope()}
)

# Instance is created and weakly cached
cache1 = obj_graph.provide(CacheService)  # "CacheService created"
cache2 = obj_graph.provide(CacheService)  # Same instance, no creation message

print(cache1 is cache2)  # True

# If we delete references and force garbage collection,
# next request creates a new instance
del cache1, cache2
import gc
gc.collect()

cache3 = obj_graph.provide(CacheService)  # "CacheService created" (new instance)

import pinject

class DatabaseConnection(object):
    def __init__(self):
        self.connection_id = id(self)

class RequestContext(object):
    def __init__(self, database_connection):
        self.db = database_connection

REQUEST_SCOPE = "request"
APPLICATION_SCOPE = "application"

class RequestScope(pinject.Scope):
    def provide(self, binding_key, default_provider_fn):
        # Always create new instance for request scope
        return default_provider_fn()

class MyBindingSpec(pinject.BindingSpec):
    def configure(self, bind):
        bind('database_connection').to_class(
            DatabaseConnection, 
            in_scope=APPLICATION_SCOPE
        )
        bind('request_context').to_class(
            RequestContext, 
            in_scope=REQUEST_SCOPE
        )

def validate_scope_usage(from_scope, to_scope):
    """Validate that request scope can use application scope."""
    if from_scope == REQUEST_SCOPE and to_scope == APPLICATION_SCOPE:
        return True
    return from_scope == to_scope

obj_graph = pinject.new_object_graph(
    binding_specs=[MyBindingSpec()],
    id_to_scope={
        APPLICATION_SCOPE: pinject.Scope(),  # Singleton-like
        REQUEST_SCOPE: RequestScope()
    },
    is_scope_usable_from_scope=validate_scope_usage
)

# Request context gets new instance, but shares database connection
context1 = obj_graph.provide(RequestContext)
context2 = obj_graph.provide(RequestContext)

print(context1 is context2)  # False (different request contexts)
print(context1.db is context2.db)  # True (shared database connection)