refactor: decouple ContextKey from Context using a structural protocol and reformat code for consistency

Signed-off-by: tercel <tercel.yi@gmail.com>

Author: tercel

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "apcore"
-version = "0.15.1"
+version = "0.16.0"
 description = "Schema-driven module standard for AI-perceivable interfaces"
 readme = "README.md"
 requires-python = ">=3.11"

src/apcore/acl.py

@@ -84,7 +84,9 @@ def register_condition(cls, key: str, handler: ACLConditionHandler) -> None:
 
     @classmethod
     def _evaluate_conditions(
-        cls, conditions: dict[str, Any], context: Context,
+        cls,
+        conditions: dict[str, Any],
+        context: Context,
     ) -> bool:
         """Evaluate all conditions with AND logic. Fail-closed on unknown."""
         for key, value in conditions.items():
@@ -100,7 +102,8 @@ def _evaluate_conditions(
             if inspect.isawaitable(result):
                 result.close()  # prevent "coroutine never awaited" warning
                 _logger.warning(
-                    "Async condition %r in sync context — treated as unsatisfied. Use async_check().", key,
+                    "Async condition %r in sync context — treated as unsatisfied. Use async_check().",
+                    key,
                 )
                 return False
             if not result:
@@ -109,7 +112,9 @@ def _evaluate_conditions(
 
     @classmethod
     async def _evaluate_conditions_async(
-        cls, conditions: dict[str, Any], context: Context,
+        cls,
+        conditions: dict[str, Any],
+        context: Context,
     ) -> bool:
         """Async variant. Awaits async handlers, calls sync handlers directly."""
         for key, value in conditions.items():

src/apcore/builtin_steps.py

@@ -17,12 +17,6 @@
 import pydantic
 
 from apcore.context import Context
-from apcore.errors import (
-    ACLDeniedError,
-    ModuleNotFoundError,
-    SchemaValidationError,
-)
-from apcore.executor import REDACTED_VALUE, redact_sensitive
 from apcore.pipeline import (
     BaseStep,
     ExecutionStrategy,
@@ -363,9 +357,7 @@ async def execute(self, ctx: PipelineContext) -> StepResult:
                 output = await module.execute(inputs, ctx.context)
             else:
                 loop = asyncio.get_event_loop()
-                output = await loop.run_in_executor(
-                    None, module.execute, inputs, ctx.context
-                )
+                output = await loop.run_in_executor(None, module.execute, inputs, ctx.context)
             ctx.output = output
         except Exception as exc:
             return StepResult(action="abort", explanation=f"Execution error: {exc}")

src/apcore/context.py

@@ -143,9 +143,7 @@ def serialize(self) -> dict[str, Any]:
             result["identity"] = None
         if self.redacted_inputs is not None:
             result["redacted_inputs"] = dict(self.redacted_inputs)
-        result["data"] = {
-            k: v for k, v in self.data.items() if not k.startswith("_")
-        }
+        result["data"] = {k: v for k, v in self.data.items() if not k.startswith("_")}
         return result
 
     @classmethod
@@ -162,8 +160,7 @@ def deserialize(cls, data: dict[str, Any]) -> Context:
         version = data.get("_context_version", 1)
         if version > 1:
             _logger.warning(
-                "Unknown _context_version %d (expected 1). "
-                "Proceeding with best-effort deserialization.",
+                "Unknown _context_version %d (expected 1). " "Proceeding with best-effort deserialization.",
                 version,
             )
 

src/apcore/context_key.py

@@ -3,15 +3,19 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Generic, TypeVar
-
-from apcore.context import Context
+from typing import Any, Generic, Protocol, TypeVar
 
 T = TypeVar("T")
 
 _MISSING = object()
 
 
+class _ContextLike(Protocol):
+    """Structural protocol for any object with a ``data`` mapping."""
+
+    data: dict[str, Any]
+
+
 @dataclass(frozen=True)
 class ContextKey(Generic[T]):
     """Typed key for context.data with namespace isolation.
@@ -22,20 +26,20 @@ class ContextKey(Generic[T]):
 
     name: str
 
-    def get(self, ctx: Context, default: T | None = None) -> T | None:  # type: ignore[type-var]
+    def get(self, ctx: _ContextLike, default: T | None = None) -> T | None:  # type: ignore[type-var]
         """Return the value for this key, or *default* if absent."""
         value = ctx.data.get(self.name, _MISSING)
         return default if value is _MISSING else value  # type: ignore[return-value]
 
-    def set(self, ctx: Context, value: T) -> None:
+    def set(self, ctx: _ContextLike, value: T) -> None:
         """Store *value* under this key in context.data."""
         ctx.data[self.name] = value
 
-    def delete(self, ctx: Context) -> None:
+    def delete(self, ctx: _ContextLike) -> None:
         """Remove this key from context.data (no-op if absent)."""
         ctx.data.pop(self.name, None)
 
-    def exists(self, ctx: Context) -> bool:
+    def exists(self, ctx: _ContextLike) -> bool:
         """Return True if this key is present in context.data."""
         return self.name in ctx.data
 

src/apcore/executor.py

@@ -216,6 +216,7 @@ def __init__(
         )
         if strategy is None:
             from apcore.builtin_steps import build_standard_strategy
+
             self._strategy = build_standard_strategy(**strategy_kwargs)
         elif isinstance(strategy, str):
             self._strategy = self._resolve_strategy_name(strategy, **strategy_kwargs)
@@ -416,9 +417,7 @@ def call(
                         errors=_convert_validation_errors(e),
                     ) from e
 
-                REDACTED_OUTPUT.set(ctx, redact_sensitive(
-                    output, module.output_schema.model_json_schema()
-                ))
+                REDACTED_OUTPUT.set(ctx, redact_sensitive(output, module.output_schema.model_json_schema()))
 
             # Step 10 -- Middleware After
             output = self._middleware_manager.execute_after(module_id, inputs, output, ctx)
@@ -922,9 +921,7 @@ async def call_async(
                         errors=_convert_validation_errors(e),
                     ) from e
 
-                REDACTED_OUTPUT.set(ctx, redact_sensitive(
-                    output, module.output_schema.model_json_schema()
-                ))
+                REDACTED_OUTPUT.set(ctx, redact_sensitive(output, module.output_schema.model_json_schema()))
 
             # Step 10 -- Middleware After (async-aware)
             output = await self._middleware_manager.execute_after_async(module_id, inputs, output, ctx)
@@ -1048,9 +1045,7 @@ async def stream(
                             errors=_convert_validation_errors(e),
                         ) from e
 
-                    REDACTED_OUTPUT.set(ctx, redact_sensitive(
-                        output, module.output_schema.model_json_schema()
-                    ))
+                    REDACTED_OUTPUT.set(ctx, redact_sensitive(output, module.output_schema.model_json_schema()))
 
                 # Step 10 -- Middleware After (async-aware)
                 output = await self._middleware_manager.execute_after_async(module_id, effective_inputs, output, ctx)
@@ -1073,9 +1068,7 @@ async def stream(
                             errors=_convert_validation_errors(e),
                         ) from e
 
-                    REDACTED_OUTPUT.set(ctx, redact_sensitive(
-                        accumulated, module.output_schema.model_json_schema()
-                    ))
+                    REDACTED_OUTPUT.set(ctx, redact_sensitive(accumulated, module.output_schema.model_json_schema()))
 
                 # Step 10 -- Middleware After on accumulated result (async-aware)
                 accumulated = await self._middleware_manager.execute_after_async(
@@ -1226,9 +1219,7 @@ def call_with_trace(
 
         if loop is None:
             return asyncio.run(engine.run(effective_strategy, pipe_ctx))
-        return self._run_in_new_thread(
-            engine.run(effective_strategy, pipe_ctx), module_id, None
-        )
+        return self._run_in_new_thread(engine.run(effective_strategy, pipe_ctx), module_id, None)
 
     async def call_async_with_trace(
         self,
@@ -1263,7 +1254,8 @@ async def call_async_with_trace(
         return await engine.run(effective_strategy, pipe_ctx)
 
     def _effective_strategy(
-        self, strategy: ExecutionStrategy | str | None,
+        self,
+        strategy: ExecutionStrategy | str | None,
     ) -> ExecutionStrategy:
         """Return the strategy to use for a call, resolving strings."""
         if strategy is None:

src/apcore/pipeline.py

@@ -182,7 +182,9 @@ class PipelineEngine:
     """Executes a pipeline strategy step by step."""
 
     async def run(
-        self, strategy: ExecutionStrategy, ctx: PipelineContext,
+        self,
+        strategy: ExecutionStrategy,
+        ctx: PipelineContext,
     ) -> tuple[Any, PipelineTrace]:
         """Run all steps in the strategy, returning the final output and trace."""
         trace = PipelineTrace(
@@ -200,13 +202,15 @@ async def run(
             try:
                 result = await step.execute(ctx)
             except Exception as exc:
-                trace.steps.append(StepTrace(
-                    name=step.name,
-                    duration_ms=(time.monotonic() - step_start) * 1000,
-                    result=StepResult(action="abort", explanation=str(exc)),
-                    skipped=False,
-                    decision_point=False,
-                ))
+                trace.steps.append(
+                    StepTrace(
+                        name=step.name,
+                        duration_ms=(time.monotonic() - step_start) * 1000,
+                        result=StepResult(action="abort", explanation=str(exc)),
+                        skipped=False,
+                        decision_point=False,
+                    )
+                )
                 trace.total_duration_ms = (time.monotonic() - start) * 1000
                 raise
 
@@ -235,13 +239,15 @@ async def run(
                         target_idx = j
                         break
                     # Record skipped steps in trace
-                    trace.steps.append(StepTrace(
-                        name=steps[j].name,
-                        duration_ms=0,
-                        result=StepResult(action="continue"),
-                        skipped=True,
-                        decision_point=False,
-                    ))
+                    trace.steps.append(
+                        StepTrace(
+                            name=steps[j].name,
+                            duration_ms=0,
+                            result=StepResult(action="continue"),
+                            skipped=True,
+                            decision_point=False,
+                        )
+                    )
                 if target_idx is None:
                     raise StepNotFoundError(
                         f"skip_to target '{target}' not found",
@@ -253,9 +259,7 @@ async def run(
 
         trace.success = True
         trace.total_duration_ms = (time.monotonic() - start) * 1000
-        final_output = (
-            ctx.validated_output if ctx.validated_output is not None else ctx.output
-        )
+        final_output = ctx.validated_output if ctx.validated_output is not None else ctx.output
         return final_output, trace
 
 

tests/test_acl_conditions.py

@@ -10,13 +10,9 @@
 
 from apcore.acl import ACL, ACLRule
 from apcore.acl_handlers import (
-    ACLConditionHandler,
-    AsyncACLConditionHandler,
     SyncACLConditionHandler,
     _IdentityTypesHandler,
     _MaxCallDepthHandler,
-    _NotHandler,
-    _OrHandler,
     _RolesHandler,
 )
 from apcore.context import Context, Identity
@@ -190,18 +186,24 @@ class TestOrHandler:
     def test_or_passes_when_any_match(self) -> None:
         """AC-011: $or evaluates with OR logic."""
         ctx = _make_context(identity_type="user", roles=["admin"])
-        acl = _make_acl_with_condition("$or", [
-            {"roles": ["admin"]},
-            {"identity_types": ["service"]},
-        ])
+        acl = _make_acl_with_condition(
+            "$or",
+            [
+                {"roles": ["admin"]},
+                {"identity_types": ["service"]},
+            ],
+        )
         assert acl.check("caller", "target", context=ctx) is True
 
     def test_or_fails_when_none_match(self) -> None:
         ctx = _make_context(identity_type="user", roles=["viewer"])
-        acl = _make_acl_with_condition("$or", [
-            {"roles": ["admin"]},
-            {"identity_types": ["service"]},
-        ])
+        acl = _make_acl_with_condition(
+            "$or",
+            [
+                {"roles": ["admin"]},
+                {"identity_types": ["service"]},
+            ],
+        )
         assert acl.check("caller", "target", context=ctx) is False
 
     def test_or_empty_list_returns_false(self) -> None:
@@ -217,10 +219,13 @@ def test_or_non_list_returns_false(self) -> None:
 
     def test_or_skips_non_dict_elements(self) -> None:
         ctx = _make_context(roles=["admin"])
-        acl = _make_acl_with_condition("$or", [
-            "invalid_string",
-            {"roles": ["admin"]},
-        ])
+        acl = _make_acl_with_condition(
+            "$or",
+            [
+                "invalid_string",
+                {"roles": ["admin"]},
+            ],
+        )
         assert acl.check("caller", "target", context=ctx) is True
 
 
@@ -244,18 +249,24 @@ class TestNestedCompound:
     def test_nested_or_with_and(self) -> None:
         """AC-032: Nested compound conditions."""
         ctx = _make_context(identity_type="service", call_chain=["a", "b"])
-        acl = _make_acl_with_condition("$or", [
-            {"roles": ["admin"]},
-            {"identity_types": ["service"], "max_call_depth": 5},
-        ])
+        acl = _make_acl_with_condition(
+            "$or",
+            [
+                {"roles": ["admin"]},
+                {"identity_types": ["service"], "max_call_depth": 5},
+            ],
+        )
         assert acl.check("caller", "target", context=ctx) is True
 
     def test_nested_or_with_and_fails_when_depth_exceeded(self) -> None:
         ctx = _make_context(identity_type="service", call_chain=["a"] * 10)
-        acl = _make_acl_with_condition("$or", [
-            {"roles": ["admin"]},
-            {"identity_types": ["service"], "max_call_depth": 5},
-        ])
+        acl = _make_acl_with_condition(
+            "$or",
+            [
+                {"roles": ["admin"]},
+                {"identity_types": ["service"], "max_call_depth": 5},
+            ],
+        )
         assert acl.check("caller", "target", context=ctx) is False
 
 

tests/test_builtin_steps.py

@@ -25,7 +25,6 @@
     BaseStep,
     ExecutionStrategy,
     PipelineContext,
-    StepResult,
 )
 
 

tests/test_context_serialization.py

@@ -98,9 +98,7 @@ class TestContextDeserialize:
     def test_deserialize_roundtrip(self) -> None:
         """Serialize then deserialize preserves fields."""
         ctx = Context.create(executor=None)
-        ctx.identity = Identity(
-            id="user-1", type="user", roles=("admin",), attrs={"org": "acme"}
-        )
+        ctx.identity = Identity(id="user-1", type="user", roles=("admin",), attrs={"org": "acme"})
         ctx.data["app.counter"] = 42
         serialized = ctx.serialize()
         restored = Context.deserialize(serialized)