fix(typing): stop 3.14 DI wire NameError in MySQL session manager

archipy/adapters/mysql/sqlalchemy/session_managers.py

@@ -1,5 +1,7 @@
 """MySQL SQLAlchemy session manager implementations."""
 
+from __future__ import annotations
+
 from typing import TYPE_CHECKING, override
 
 from sqlalchemy import URL

features/atomic_transactions.feature

@@ -205,6 +205,18 @@ Feature: SQLAlchemy Atomic Transactions
       | sqlite  |
       | mysql   |
 
+  @unit
+  Scenario Outline: SQLAlchemy session manager callables stay inspectable at runtime
+    When I inspect callables of the <db_type> SQLAlchemy session manager module
+    Then all session manager callable signatures resolve without NameError
+
+    Examples:
+      | db_type   |
+      | postgres  |
+      | sqlite    |
+      | mysql     |
+      | starrocks |
+
   # StarRocks SQL transaction limitations (shared-nothing allin1): nested ArchiPy
   # blocks share one txn and hit multi-insert; no multiple same-table inserts;
   # no further DML on a table already modified in the same txn.

features/steps/atomic_transaction_steps.py

@@ -5,6 +5,8 @@
 """
 
 import asyncio
+import importlib
+import inspect
 import logging
 import os
 import tempfile
@@ -547,6 +549,32 @@ def verify_session_usable():
     assert result, "Session is not usable after transaction rollback"
 
 
+@when("I inspect callables of the {db_type} SQLAlchemy session manager module")
+def step_when_inspect_session_manager_module_callables(context, db_type):
+    scenario_context = get_current_scenario_context(context)
+    module = importlib.import_module(f"archipy.adapters.{db_type}.sqlalchemy.session_managers")
+    failures: list[str] = []
+    for name, obj in vars(module).items():
+        members = vars(obj).items() if isinstance(obj, type) else [(None, obj)]
+        for member_name, member in members:
+            func = member.__func__ if isinstance(member, (staticmethod, classmethod)) else member
+            if not inspect.isfunction(func):
+                continue
+            label = name if member_name is None else f"{name}.{member_name}"
+            try:
+                inspect.signature(func)
+            except NameError as e:
+                failures.append(f"{label}: {e}")
+    scenario_context.store("session_manager_inspect_failures", failures)
+
+
+@then("all session manager callable signatures resolve without NameError")
+def step_then_session_manager_signatures_resolve(context):
+    scenario_context = get_current_scenario_context(context)
+    failures = scenario_context.get("session_manager_inspect_failures")
+    assert not failures, "Session manager annotation inspection failed:\n" + "\n".join(failures)
+
+
 @when("nested atomic transactions are attempted on StarRocks")
 def step_when_nested_atomic_attempted_on_starrocks(context):
     """Attempt nested StarRocks atomic blocks and capture the limitation error."""