Refactoring: types, mixin improvements, clarity

Author: alganet

apywire/__init__.py

@@ -12,13 +12,15 @@
     WiringError,
 )
 from .runtime import Accessor, AioAccessor, Spec, SpecEntry, WiringRuntime
+from .threads import ThreadSafeMixin
 from .wiring import WiringBase
 
 Wiring = WiringRuntime
 
 __all__ = [
     "Spec",
     "SpecEntry",
+    "ThreadSafeMixin",
     "Wiring",
     "WiringRuntime",
     "WiringCompiler",

apywire/compiler.py

@@ -604,12 +604,12 @@ def compile(self, *, aio: bool = False, thread_safe: bool = False) -> str:
             modules.add("apywire.exceptions")
         for module in sorted(modules):
             if module == "apywire.threads":
-                # Import CompiledThreadSafeMixin from threads
+                # Import ThreadSafeMixin from threads
                 body.append(
                     ast.ImportFrom(
                         module="apywire.threads",
                         names=[
-                            ast.alias(name="CompiledThreadSafeMixin"),
+                            ast.alias(name="ThreadSafeMixin"),
                         ],
                         level=0,
                     )
@@ -658,25 +658,23 @@ def compile(self, *, aio: bool = False, thread_safe: bool = False) -> str:
                 type_params=[],
             )
             class_body.insert(0, init_def)
-        for name, (
-            module_name,
-            class_name,
-            factory_method,
-            data,
-        ) in self._parsed.items():
+        for name, entry in self._parsed.items():
             # Skip synthetic auto-promoted constants
             # These require runtime interpolation and can't be pre-computed
-            if module_name == SYNTHETIC_CONST and class_name == "str":
+            if (
+                entry.module_name == SYNTHETIC_CONST
+                and entry.class_name == "str"
+            ):
                 continue
 
             # Regular wired object
             class_body.append(
                 self._compile_property(
                     name,
-                    module_name,
-                    class_name,
-                    factory_method,
-                    cast(_ResolvedSpecMapping, data),
+                    entry.module_name,
+                    entry.class_name,
+                    entry.factory_method,
+                    cast(_ResolvedSpecMapping, entry.data),
                     aio=aio,
                     thread_safe=thread_safe,
                 )
@@ -696,14 +694,12 @@ def compile(self, *, aio: bool = False, thread_safe: bool = False) -> str:
 
         class_body = [ast.Pass()] if not class_body else class_body
         # When using thread_safe compiled output we will rely on the
-        # CompiledThreadSafeMixin and _LockUnavailableError imported from
+        # ThreadSafeMixin and _LockUnavailableError imported from
         # apywire.threads rather than embedding helper code.
         # Build class definition
         class_bases: list[ast.expr] = []
         if thread_safe:
-            class_bases.append(
-                ast.Name(id="CompiledThreadSafeMixin", ctx=ast.Load())
-            )
+            class_bases.append(ast.Name(id="ThreadSafeMixin", ctx=ast.Load()))
         class_def = ast.ClassDef(
             name="Compiled",
             bases=class_bases,

apywire/runtime.py

@@ -18,7 +18,7 @@
     UnknownPlaceholderError,
     WiringError,
 )
-from apywire.threads import CompiledThreadSafeMixin
+from apywire.threads import ThreadSafeMixin
 from apywire.wiring import (
     Spec,
     SpecEntry,
@@ -48,7 +48,7 @@ class _Constructor(Protocol):
     def __call__(self, *args: object, **kwargs: object) -> object: ...
 
 
-class WiringRuntime(WiringBase, CompiledThreadSafeMixin):
+class WiringRuntime(WiringBase, ThreadSafeMixin):
     """Runtime container for wired objects.
 
     This class handles the runtime resolution and instantiation of wired
@@ -79,21 +79,23 @@ def __init__(
             max_lock_attempts=max_lock_attempts,
             lock_retry_sleep=lock_retry_sleep,
         )
+        if self._thread_safe:
+            self._init_thread_safety(max_lock_attempts, lock_retry_sleep)
 
     def _init_thread_safety(
         self,
         max_lock_attempts: int = 10,
         lock_retry_sleep: float = 0.01,
     ) -> None:
         """Initialize thread safety mixin."""
-        CompiledThreadSafeMixin._init_thread_safety(
+        ThreadSafeMixin._init_thread_safety(
             self, max_lock_attempts, lock_retry_sleep
         )
 
     def _get_resolving_stack(self) -> list[str]:
         """Return the resolving stack for the current context."""
         if self._thread_safe:
-            return CompiledThreadSafeMixin._get_resolving_stack(self)
+            return ThreadSafeMixin._get_resolving_stack(self)
         return self._resolving_stack
 
     def __getattr__(self, name: str) -> Accessor:
@@ -124,13 +126,13 @@ def _instantiate_attr(
     ) -> object:
         """Instantiate an attribute using the configured strategy.
 
-        If thread_safe is True, uses the CompiledThreadSafeMixin implementation
+        If thread_safe is True, uses the ThreadSafeMixin implementation
         which handles optimistic locking and global fallback.
         If thread_safe is False, uses a simple direct instantiation.
         """
         if self._thread_safe:
             # Use the mixin's implementation which handles locking
-            return CompiledThreadSafeMixin._instantiate_attr(self, name, maker)
+            return ThreadSafeMixin._instantiate_attr(self, name, maker)
 
         # Non-thread-safe path: simple check and set
         if name in self._values:
@@ -206,26 +208,31 @@ def _instantiate_impl(self, name: str) -> _RuntimeValue:
                     f"Unknown placeholder '{name}' referenced."
                 )
 
-            module_name, class_name, factory_method, data = self._parsed[name]
+            entry = self._parsed[name]
 
             # Check for synthetic auto-promoted constant
-            if module_name == SYNTHETIC_CONST and class_name == "str":
+            if (
+                entry.module_name == SYNTHETIC_CONST
+                and entry.class_name == "str"
+            ):
                 # This is an auto-promoted constant with string interpolation
-                value = self._format_string_constant(data, context=name)
+                value = self._format_string_constant(entry.data, context=name)
                 self._values[name] = value
                 return value
 
-            module = importlib.import_module(module_name)
-            cls = cast(_Constructor, getattr(module, class_name))
+            module = importlib.import_module(entry.module_name)
+            cls = cast(_Constructor, getattr(module, entry.class_name))
 
             # If a factory method is specified, get it from the class
-            if factory_method:
-                constructor = cast(_Constructor, getattr(cls, factory_method))
+            if entry.factory_method:
+                constructor = cast(
+                    _Constructor, getattr(cls, entry.factory_method)
+                )
             else:
                 constructor = cls
 
             # Resolve arguments
-            kwargs = self._resolve_runtime(data, context=name)
+            kwargs = self._resolve_runtime(entry.data, context=name)
 
             try:
                 if isinstance(kwargs, dict):

apywire/threads.py

@@ -19,7 +19,7 @@
 from apywire.exceptions import LockUnavailableError
 
 
-class ThreadLocalState(threading.local):
+class _ThreadLocalState(threading.local):
     """Typed thread-local storage for wiring resolution state.
 
     This class provides properly typed attributes to avoid mypy Any propagation
@@ -32,12 +32,14 @@ class ThreadLocalState(threading.local):
         held_locks: List of locks currently held by this thread.
     """
 
-    resolving_stack: list[str]
-    mode: Literal["optimistic", "global"] | None
-    held_locks: list[threading.RLock]
+    def __init__(self) -> None:
+        super().__init__()
+        self.resolving_stack: list[str] = []
+        self.mode: Literal["optimistic", "global"] | None = None
+        self.held_locks: list[threading.RLock] = []
 
 
-class CompiledThreadSafeMixin:
+class ThreadSafeMixin:
     """Mixin that provides thread-safety helpers for wiring containers.
 
     This mixin is used by both runtime `Wiring` and compiled `Compiled`
@@ -63,7 +65,7 @@ def _init_thread_safety(
         self._inst_lock = threading.RLock()
         self._attr_locks: dict[str, threading.RLock] = {}
         self._attr_locks_lock = threading.Lock()
-        self._local: ThreadLocalState = ThreadLocalState()
+        self._local: _ThreadLocalState = _ThreadLocalState()
         self._max_lock_attempts = max_lock_attempts
         self._lock_retry_sleep = lock_retry_sleep
 
@@ -74,18 +76,12 @@ def _get_attribute_lock(self, name: str) -> threading.RLock:
             return self._attr_locks[name]
 
     def _get_resolving_stack(self) -> list[str]:
-        if not hasattr(self._local, "resolving_stack"):
-            self._local.resolving_stack = []
         return self._local.resolving_stack
 
     def _get_held_locks(self) -> list[threading.RLock]:
-        if not hasattr(self._local, "held_locks"):
-            self._local.held_locks = []
         return self._local.held_locks
 
     def _release_held_locks(self) -> None:
-        if not hasattr(self._local, "held_locks"):
-            return
         for lock in reversed(self._local.held_locks):
             lock.release()
         self._local.held_locks.clear()
@@ -156,7 +152,7 @@ def _instantiate_attr(
             return value
 
         lock = self._get_attribute_lock(name)
-        mode: str | None = getattr(self._local, "mode", None)
+        mode = self._local.mode
         if mode is None:
             # Optimistic locking: Try to acquire per-attribute lock without
             # blocking. If successful, instantiate in optimistic mode.
@@ -192,13 +188,7 @@ def _instantiate_attr(
                         return inst
                 finally:
                     # Clean up optimistic mode when exiting top-level call
-                    if (
-                        cast(
-                            Literal["optimistic", "global"] | None,
-                            getattr(self._local, "mode", None),
-                        )
-                        == "optimistic"
-                    ):
+                    if self._local.mode == "optimistic":
                         self._local.mode = None
 
             # fallback to global

apywire/wiring.py

@@ -19,7 +19,7 @@
 import re
 from collections import deque
 from types import EllipsisType
-from typing import TypeAlias, cast
+from typing import NamedTuple, TypeAlias, cast
 
 from apywire.constants import (
     PLACEHOLDER_END,
@@ -29,7 +29,6 @@
     SYNTHETIC_CONST,
 )
 from apywire.exceptions import CircularWiringError, UnknownPlaceholderError
-from apywire.threads import ThreadLocalState
 
 _ConstantValue: TypeAlias = (
     str | bytes | bool | int | float | complex | EllipsisType | None
@@ -91,20 +90,24 @@ def __init__(self, name: str) -> None:
 )
 Spec: TypeAlias = dict[str, _SpecMapping | _ConstantValue]
 
-# Type aliases for parsed spec entries
-_ParsedEntry: TypeAlias = tuple[
-    str, str, str | None, _ResolvedSpecMapping | str | _WiredRef
-]  # (module, class, factory_method, data)
-_UnresolvedParsedEntry: TypeAlias = tuple[
-    str, str, str, str | None, _SpecMapping
-]  # (module, class, name, factory_method, data)
 
+class _ParsedEntry(NamedTuple):
+    """A parsed wiring entry ready for instantiation."""
 
-class _ThreadLocalState(ThreadLocalState):
-    """Thread-local state for wiring resolution.
+    module_name: str
+    class_name: str
+    factory_method: str | None
+    data: _ResolvedSpecMapping | str | _WiredRef
 
-    Inherits from ThreadLocalState to get properly typed attributes.
-    """
+
+class _UnresolvedParsedEntry(NamedTuple):
+    """An intermediate parsed entry before placeholder resolution."""
+
+    module_name: str
+    class_name: str
+    name: str
+    factory_method: str | None
+    data: _SpecMapping
 
 
 class WiringBase:
@@ -150,13 +153,13 @@ def __init__(
 
         # Parse wired objects first
         self._parsed: dict[str, _ParsedEntry] = {
-            name: (
-                module_name,
-                class_name,
-                factory_method,
-                cast(_ResolvedSpecMapping, self._resolve(value)),
+            entry.name: _ParsedEntry(
+                entry.module_name,
+                entry.class_name,
+                entry.factory_method,
+                cast(_ResolvedSpecMapping, self._resolve(entry.data)),
             )
-            for module_name, class_name, name, factory_method, value in parsed
+            for entry in parsed
         }
 
         # Classify constants by placeholder type
@@ -227,19 +230,13 @@ def __init__(
         # Create synthetic parsed entries for auto-promoted constants
         for key, value in const_with_wired_refs.items():
             # Create a synthetic entry that will format the string at runtime
-            self._parsed[key] = (
+            self._parsed[key] = _ParsedEntry(
                 SYNTHETIC_CONST,  # Synthetic module marker
                 "str",  # Will use string formatting
                 None,  # No factory method
                 cast(str | _WiredRef, self._resolve(value)),
             )
-        if self._thread_safe:
-            # Thread-safe mode: use mixin helpers for lock state.
-            # Note: Derived classes must implement _init_thread_safety
-            # if they support thread safety (like WiringRuntime).
-            if hasattr(self, "_init_thread_safety"):
-                self._init_thread_safety(max_lock_attempts, lock_retry_sleep)
-        else:
+        if not self._thread_safe:
             # Non-thread-safe mode: use simple list for resolving stack
             self._resolving_stack: list[str] = []
 
@@ -276,7 +273,7 @@ def _parse_spec_entry(
                 f"invalid spec key '{key}': missing module qualification"
             )
 
-        return (
+        return _UnresolvedParsedEntry(
             module_name,
             class_name,
             name,