Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 3 additions & 4 deletions src/flext_cli/_constants/errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,6 @@ class FlextCliConstantsErrors:
)
ERR_USER_CANCELLED_CONFIRMATION: ClassVar[str] = "User cancelled confirmation"
ERR_INPUT_STREAM_ENDED: ClassVar[str] = "Input stream ended"
ERR_CONFIRMATION_FAILED_FMT: ClassVar[str] = "Confirmation failed: {error}"
ERR_NO_CHOICES: ClassVar[str] = "No choices provided"
ERR_INTERACTIVE_CHOICE_DISABLED: ClassVar[str] = (
"Interactive mode disabled for choice prompt"
Expand All @@ -107,9 +106,6 @@ class FlextCliConstantsErrors:
ERR_PASSWORD_TOO_SHORT_FMT: ClassVar[str] = (
"Password too short: minimum {min_length} characters"
)
ERR_PROMPT_FAILED_FMT: ClassVar[str] = "Prompt failed: {error}"
ERR_CHOICE_PROMPT_FAILED_FMT: ClassVar[str] = "Choice prompt failed: {error}"
ERR_PASSWORD_PROMPT_FAILED_FMT: ClassVar[str] = "Password prompt failed: {error}"

ERR_INVALID_COMMAND_NAME: ClassVar[str] = "Invalid command name"
ERR_COMMAND_FAILED: ClassVar[str] = "Command failed"
Expand All @@ -123,6 +119,9 @@ class FlextCliConstantsErrors:
ERR_CLI_DEFINITION_FIELD: ClassVar[str] = (
"command '{command}' model '{model}' field '{field}': {reason}"
)
ERR_FIELD_DEFAULT_NOT_CLI_VALUE_FMT: ClassVar[str] = (
"field '{field_name}' default {value!r} has no CLI option form"
)


__all__: t.MutableSequenceOf[str] = ["FlextCliConstantsErrors"]
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,14 @@ class FlextCliUtilitiesOptions(FlextCliUtilitiesOptionsPart01):
def field_default(
cls, field_name: str, field_info: m.FieldInfo, settings: t.Cli.ModelLike | None
) -> t.Cli.CliValue | None:
"""Resolve CLI default from settings first, then from model field metadata."""
"""Resolve CLI default from settings first, then from model field metadata.

``None`` is the typed absence of a default. Any other default without a
CLI form fails at command build with its cause: the default-source
validation error escapes unchanged, and a validated default that no
Typer option carries raises ``TypeError``. Structured defaults travel
through the JSON-option path.
"""
default_factory = getattr(field_info, "default_factory", None)
source_value = (
getattr(settings, field_name)
Expand All @@ -28,29 +35,21 @@ def field_default(
if callable(default_factory)
else getattr(field_info, "default", None)
)
if source_value is None:
return None
if cls.is_json_option(getattr(field_info, "annotation", None) or str):
# A JSON option's default is the JSON text its parser validates.
return None if source_value is None else u.to_json(source_value).decode()
try:
normalized_source = t.Cli.CLI_DEFAULT_SOURCE_ADAPTER.validate_python(
source_value
)
except c.EXC_VALIDATION_TYPE_VALUE:
normalized_source = None
if normalized_source is None:
return None
match normalized_source:
case _ if (
normalized_atom := cls.normalize_cli_atom(normalized_source)
) is not None:
normalized_default: t.Cli.CliValue | None = normalized_atom
case _ if cls.is_string_sequence(normalized_source):
normalized_default = t.Cli.STR_SEQUENCE_ADAPTER.validate_python(
normalized_source
return u.to_json(source_value).decode()
normalized_atom = cls.normalize_cli_atom(
t.Cli.CLI_DEFAULT_SOURCE_ADAPTER.validate_python(source_value)
)
if normalized_atom is None:
raise TypeError(
c.Cli.ERR_FIELD_DEFAULT_NOT_CLI_VALUE_FMT.format(
field_name=field_name, value=source_value
)
case _:
normalized_default = None
return normalized_default
)
return normalized_atom

@staticmethod
def build_option(
Expand Down
133 changes: 7 additions & 126 deletions src/flext_cli/_utilities/model_commands.py
Original file line number Diff line number Diff line change
@@ -1,110 +1,19 @@
"""Thin model-command adapters shared through ``u.Cli``.
"""Model-source helpers shared through ``u.Cli``.

NOTE (multi-agent): handlers are annotated as the structural type
``Callable[[M], t.JsonValue]`` (the exact contract of
``p.Cli.ModelCommandHandler``) because mypy degrades inherited nested
protocol classes reached through the ``p.Cli`` facade MRO to ``Any``
(pyrefly/pyright resolve them). Structural typing is the FLEXT default and
keeps all three checkers precise without casts.
Model-backed commands have one owner, ``cli.model_command``; model
derivation has one owner, ``cli.derive_model``. This module keeps only the
source-extraction primitive that ``cli.derive_model`` consumes.
"""

from __future__ import annotations

import inspect
from collections.abc import Callable, Mapping
from typing import Protocol, cast
from collections.abc import Mapping

from flext_cli import p, settings, t
from flext_core import m
from flext_cli import t


class FlextCliUtilitiesModelCommands:
"""Model command methods exposed directly on ``u.Cli``."""

class SignatureCarrier(Protocol):
"""A callable whose CLI parameters typer/click read from `__signature__`.

`__signature__` is a documented runtime attribute of function objects, but
it is absent from the static `FunctionType`, so assigning it directly is
rejected by the type checkers. Declaring it here states the contract the
generated command actually satisfies, with no suppression.
"""

__signature__: inspect.Signature
__annotations__: dict[str, object]

class Builder[M: t.Cli.ModelLike]:
"""Thin builder for direct model-backed command callables."""

def __init__(
self,
model_class: t.ModelClass[M],
handler: Callable[[M], t.JsonValue],
settings: t.Cli.ModelLike | None = None,
) -> None:
"""Store the canonical inputs for deferred command construction."""
super().__init__()
self.model_class = model_class
self.handler = handler
self.settings = settings

def _resolve_default(self, field_info: m.FieldInfo) -> t.Cli.CliValue | type:
if field_info.is_required():
return inspect.Parameter.empty
# NOTE (multi-agent): ``FieldInfo.get_default`` is typed ``Any``
# in pydantic; the declared union is the real runtime contract.
return cast(
"t.Cli.CliValue | type",
field_info.get_default(call_default_factory=True),
)

def build(self) -> t.Cli.CliCommand:
"""Build a direct callable with a real runtime signature."""
model_fields = self.model_class.model_fields
parameters = [
inspect.Parameter(
name=field_name,
kind=inspect.Parameter.KEYWORD_ONLY,
default=self._resolve_default(field_info),
annotation=getattr(field_info, "annotation", None) or str,
)
for field_name, field_info in model_fields.items()
if getattr(field_info, "exclude", None) is not True
]
signature = inspect.Signature(parameters)

# NOTE (multi-agent): ``self.settings`` (per-command, may be any
# model or None) falls back to the module settings singleton.
# Overrides apply only when the effective settings is a full
# Settings protocol (has update_global) — a plain model skips them.
effective_settings = (
self.settings if self.settings is not None else settings
)

def command(**kwargs: t.Cli.CliValue) -> t.JsonValue:
if isinstance(effective_settings, p.Cli.Settings):
settings_fields = effective_settings.model_dump()
applicable_overrides = {
field_name: field_value
for field_name, field_value in kwargs.items()
if field_name in settings_fields
}
if applicable_overrides:
effective_settings.update_global(**applicable_overrides)
model = self.model_class.model_validate(kwargs)
return self.handler(model)

# typer/click read the CLI parameters off `__signature__`; the
# protocol above declares that contract so no cast is suppressed.
typed_command = cast(
"FlextCliUtilitiesModelCommands.SignatureCarrier", command
)
typed_command.__signature__ = signature
typed_command.__annotations__ = {
parameter.name: parameter.annotation for parameter in parameters
}
typed_command.__annotations__["return"] = t.JsonValue
return command
"""Model-source methods exposed directly on ``u.Cli``."""

@staticmethod
def model_source_data(
Expand All @@ -123,33 +32,5 @@ def model_source_data(
}
return t.Cli.JSON_MAPPING_ADAPTER.validate_python(filtered_payload)

@classmethod
def derive_model[M: t.Cli.ModelLike](
cls,
model_cls: type[M],
*sources: t.Cli.ModelSource,
overrides: t.ScalarMapping | None = None,
) -> M:
"""Derive a target model from ordered model/mapping sources."""
merged: t.MutableJsonMapping = {}
for source in sources:
merged.update(cls.model_source_data(model_cls, source))
if overrides is not None:
merged.update(cls.model_source_data(model_cls, overrides))
validated: M = model_cls.model_validate(merged)
return validated

@staticmethod
def build_model_command[M: t.Cli.ModelLike](
model_class: t.ModelClass[M],
handler: Callable[[M], t.JsonValue],
settings: t.Cli.ModelLike | None = None,
) -> t.Cli.CliCommand:
"""Build a model command through the canonical CLI service."""
# NOTE (multi-agent): All model-class ingress uses t.ModelClass.
return FlextCliUtilitiesModelCommands.Builder(
model_class=model_class, handler=handler, settings=settings
).build()


__all__: list[str] = ["FlextCliUtilitiesModelCommands"]
55 changes: 4 additions & 51 deletions src/flext_cli/services/_prompts_support.py
Original file line number Diff line number Diff line change
@@ -1,22 +1,16 @@
"""Prompt service support primitives.

NOTE (multi-agent): mro-i6nq.13 — moved here from the removed
``_prompts_parts/flextcliprompts_support.py`` so the whole numbered
``_prompts_parts`` package could be eliminated. Adds the ``_guarded`` DRY
helper that collapses the repeated ``try/except CLI_SAFE_EXCEPTIONS -> _fatal
-> r.fail(fmt)`` idiom into one canonical ``u.guard_result`` boundary.
Input-port failures are not caught here: the reader's exception escapes the
prompt method with its cause. Only declared prompt outcomes return ``r``.
"""

from __future__ import annotations

import getpass
from typing import TYPE_CHECKING, Annotated, Self
from typing import Annotated, Self

from flext_cli import c, m, p, r, s, settings, t, u

if TYPE_CHECKING:
from collections.abc import Callable


class _PromptInputReaderDefault:
"""Resolve the built-in input reader without storing a method descriptor."""
Expand Down Expand Up @@ -76,44 +70,6 @@ def _is_test_env(self) -> bool:
"""
return u.Cli.cli_test_env(settings)

def _guarded[TResult](
self,
operation: str,
message: str,
work: Callable[[], p.Result[TResult]],
*,
consequence: str,
error_format: str,
) -> p.Result[TResult]:
"""Run a Result-returning prompt operation behind one exception boundary.

Collapses the canonical ``try/except CLI_SAFE_EXCEPTIONS -> _fatal ->
r.fail(fmt)`` idiom shared by every interactive prompt method into a
single ``u.guard_result`` call plus structured fatal logging.
"""
guarded = u.guard_result(
work, catch=c.Cli.CLI_SAFE_EXCEPTIONS, op_name=operation
)
if guarded.success:
return guarded
exc = guarded.error or operation
self._fatal(operation, message, Exception(exc), consequence)
return r[TResult].fail(error_format.format(error=exc))

def _fatal(
self, operation: str, message: str, exc: Exception, consequence: str
) -> None:
self._log(
c.LogLevel.ERROR,
f"FATAL ERROR during {operation} - operation aborted",
operation=operation,
prompt_message=message,
error=str(exc),
error_type=type(exc).__name__,
consequence=consequence,
severity="critical",
)

def _log(self, log_level: str, message: str, **context: t.LogValue) -> None:
match log_level:
case c.LogLevel.DEBUG:
Expand All @@ -126,10 +82,7 @@ def _log(self, log_level: str, message: str, **context: t.LogValue) -> None:
self.logger.info(message, **context)

def _print_message(
self,
message: str,
log_level: str,
message_format: str,
self, message: str, log_level: str, message_format: str
) -> p.Result[bool]:
# Fail loud: a logger failure propagates with its cause.
self._log(log_level, message_format.format(message=message))
Expand Down
Loading
Loading