refactor: type cli typer context state

This commit is contained in:
lda
2026-06-03 18:41:20 +07:00 Verified
parent 2364d0cfa3
commit 47e1e3af67
3 changed files with 108 additions and 18 deletions
+7 -6
View File
@@ -5,6 +5,7 @@ from typing import Annotated
import typer import typer
from .commands import artifacts, caps, deployments, docs, drafts, explain, runs, schema from .commands import artifacts, caps, deployments, docs, drafts, explain, runs, schema
from .context import CliTyperState
app = typer.Typer( app = typer.Typer(
name="wf", name="wf",
@@ -37,12 +38,12 @@ def root(
] = None, ] = None,
) -> None: ) -> None:
"""Run workflow platform commands.""" """Run workflow platform commands."""
ctx.obj = { ctx.obj = CliTyperState(
"config_path": config, config_path=config,
"force_local": local, force_local=local,
"rpc_url": url, rpc_url=url,
"rpc_timeout_seconds": timeout, rpc_timeout_seconds=timeout,
} )
app.add_typer(caps.app, name="cap") app.add_typer(caps.app, name="cap")
+44 -11
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
import json import json
from collections.abc import Mapping
import typer import typer
from pydantic import ValidationError from pydantic import ValidationError
@@ -39,11 +40,48 @@ class LocalCliContext:
handlers: WorkflowApi handlers: WorkflowApi
@dataclass(frozen=True, slots=True)
class CliTyperState:
"""Typed boundary for Typer's untyped `Context.obj` payload.
Typer/Click do not make `Context.obj` generic, so every command should read
root CLI options through this adapter instead of spelling dict keys locally.
"""
config_path: str = "wf_mcp.config.json"
force_local: bool = False
rpc_url: str | None = None
rpc_timeout_seconds: float | None = None
@classmethod
def from_context(cls, ctx: typer.Context) -> CliTyperState:
obj = ctx.obj
if isinstance(obj, cls):
return obj
if isinstance(obj, Mapping):
return cls.from_mapping(obj)
return cls()
@classmethod
def from_mapping(cls, obj: Mapping[object, object]) -> CliTyperState:
config_path = obj.get("config_path", cls.config_path)
rpc_url = obj.get("rpc_url")
timeout = obj.get("rpc_timeout_seconds")
return cls(
config_path=(
config_path if isinstance(config_path, str) else cls.config_path
),
force_local=bool(obj.get("force_local", cls.force_local)),
rpc_url=rpc_url if isinstance(rpc_url, str) else None,
rpc_timeout_seconds=(
float(timeout) if isinstance(timeout, float | int) else None
),
)
def config_path_from_context(ctx: typer.Context) -> str: def config_path_from_context(ctx: typer.Context) -> str:
"""Return the root --config path captured by the Typer callback.""" """Return the root --config path captured by the Typer callback."""
obj = ctx.obj if isinstance(ctx.obj, dict) else {} return CliTyperState.from_context(ctx).config_path
value = obj.get("config_path", "wf_mcp.config.json")
return value if isinstance(value, str) else "wf_mcp.config.json"
def load_cli_context( def load_cli_context(
@@ -136,20 +174,15 @@ def load_local_cli_context(
def force_local_from_context(ctx: typer.Context) -> bool: def force_local_from_context(ctx: typer.Context) -> bool:
obj = ctx.obj if isinstance(ctx.obj, dict) else {} return CliTyperState.from_context(ctx).force_local
return bool(obj.get("force_local", False))
def rpc_url_from_context(ctx: typer.Context) -> str | None: def rpc_url_from_context(ctx: typer.Context) -> str | None:
obj = ctx.obj if isinstance(ctx.obj, dict) else {} return CliTyperState.from_context(ctx).rpc_url
value = obj.get("rpc_url")
return value if isinstance(value, str) else None
def rpc_timeout_from_context(ctx: typer.Context) -> float | None: def rpc_timeout_from_context(ctx: typer.Context) -> float | None:
obj = ctx.obj if isinstance(ctx.obj, dict) else {} return CliTyperState.from_context(ctx).rpc_timeout_seconds
value = obj.get("rpc_timeout_seconds")
return value if isinstance(value, float | int) else None
def load_cli_context_from_typer(ctx: typer.Context) -> CliContext: def load_cli_context_from_typer(ctx: typer.Context) -> CliContext:
+57 -1
View File
@@ -1,10 +1,66 @@
from __future__ import annotations from __future__ import annotations
import click
import json import json
from pathlib import Path from pathlib import Path
import typer
from wf_api import WorkflowApi from wf_api import WorkflowApi
from wf_cli.context import load_cli_context from wf_cli.context import (
CliTyperState,
config_path_from_context,
force_local_from_context,
load_cli_context,
rpc_timeout_from_context,
rpc_url_from_context,
)
def _typer_context(obj: object | None) -> typer.Context:
ctx = typer.Context(click.Command("wf"))
ctx.obj = obj
return ctx
def test_cli_typer_state_reads_typed_context_object() -> None:
ctx = _typer_context(
CliTyperState(
config_path="remote.json",
force_local=True,
rpc_url="http://127.0.0.1:8000/rpc",
rpc_timeout_seconds=2.5,
)
)
assert config_path_from_context(ctx) == "remote.json"
assert force_local_from_context(ctx) is True
assert rpc_url_from_context(ctx) == "http://127.0.0.1:8000/rpc"
assert rpc_timeout_from_context(ctx) == 2.5
def test_cli_typer_state_accepts_legacy_dict_context_object() -> None:
ctx = _typer_context(
{
"config_path": "legacy.json",
"force_local": True,
"rpc_url": "http://localhost:9000/rpc",
"rpc_timeout_seconds": 3,
}
)
assert config_path_from_context(ctx) == "legacy.json"
assert force_local_from_context(ctx) is True
assert rpc_url_from_context(ctx) == "http://localhost:9000/rpc"
assert rpc_timeout_from_context(ctx) == 3.0
def test_cli_typer_state_defaults_for_missing_context_object() -> None:
ctx = _typer_context(None)
assert config_path_from_context(ctx) == "wf_mcp.config.json"
assert force_local_from_context(ctx) is False
assert rpc_url_from_context(ctx) is None
assert rpc_timeout_from_context(ctx) is None
def test_load_cli_context_builds_service_and_handlers(tmp_path: Path) -> None: def test_load_cli_context_builds_service_and_handlers(tmp_path: Path) -> None: