From c0e6fcb1b313bcd779bd35d63dc469dae6e17e52 Mon Sep 17 00:00:00 2001 From: Gas Town Mayor Date: Fri, 28 Aug 2026 20:58:43 -0300 Subject: [PATCH] fix(models): create immutable defaults per instance --- .../_models/_base/flextclimodelsbase_part_01.py | 11 +++++++---- .../_models/_base/flextclimodelsbase_part_02.py | 8 +++++--- src/flext_cli/_models/_defaults.py | 10 ++++++++-- src/flext_cli/_models/docx_document.py | 5 ++--- src/flext_cli/_models/pipeline.py | 10 ++++++---- src/flext_cli/_models/pptx_presentation.py | 5 ++--- src/flext_cli/_protocols/domain.py | 4 +--- .../_utilities/_xlxx/xlsx_snapshot_structure.py | 17 ++++++----------- src/flext_cli/_utilities/pipeline.py | 6 +----- src/flext_cli/services/_prompts_support.py | 9 ++++++++- .../testsflextclipipeline_part_02.py | 17 ----------------- 11 files changed, 46 insertions(+), 56 deletions(-) diff --git a/src/flext_cli/_models/_base/flextclimodelsbase_part_01.py b/src/flext_cli/_models/_base/flextclimodelsbase_part_01.py index 82a07e13..6d5d1e27 100644 --- a/src/flext_cli/_models/_base/flextclimodelsbase_part_01.py +++ b/src/flext_cli/_models/_base/flextclimodelsbase_part_01.py @@ -6,7 +6,7 @@ from typing import Annotated, ClassVar from flext_cli import t -from flext_cli._models._defaults import EMPTY_JSON_MAPPING +from flext_cli._models._defaults import empty_json_mapping from flext_core import m, u @@ -84,9 +84,10 @@ class DisplayData(m.BaseModel): data: Annotated[ t.JsonMapping, m.Field( + default_factory=empty_json_mapping, description="Field-value pairs for display", ), - ] = EMPTY_JSON_MAPPING + ] @u.model_serializer def _serialize(self) -> t.JsonMapping: @@ -102,9 +103,10 @@ class LoadedConfig(m.BaseModel): content: Annotated[ t.JsonMapping, m.Field( + default_factory=empty_json_mapping, description="Loaded configuration content (dict or other JSON value)", ), - ] = EMPTY_JSON_MAPPING + ] class CliNormalizedJson(m.RootModel[t.JsonValue]): """Normalize raw JSON value with flat JSON serialization semantics. @@ -132,9 +134,10 @@ class NormalizedJsonList(m.BaseModel): default: Annotated[ t.JsonMapping, m.Field( + default_factory=empty_json_mapping, description="Default mapping if value is not a dict", ), - ] = EMPTY_JSON_MAPPING + ] @property def resolved(self) -> t.JsonMapping: diff --git a/src/flext_cli/_models/_base/flextclimodelsbase_part_02.py b/src/flext_cli/_models/_base/flextclimodelsbase_part_02.py index 8b17b3b8..718e064c 100644 --- a/src/flext_cli/_models/_base/flextclimodelsbase_part_02.py +++ b/src/flext_cli/_models/_base/flextclimodelsbase_part_02.py @@ -5,7 +5,7 @@ from typing import Annotated, ClassVar from flext_cli import c, t -from flext_cli._models._defaults import EMPTY_STR_MAPPING +from flext_cli._models._defaults import empty_str_mapping from flext_core import m, u @@ -53,15 +53,17 @@ class ProcessEnvironmentSpec(m.BaseModel): base_env: Annotated[ t.StrMapping, m.Field( + default_factory=empty_str_mapping, description="Base environment inherited from the current process", ), - ] = EMPTY_STR_MAPPING + ] overrides: Annotated[ t.StrMapping, m.Field( + default_factory=empty_str_mapping, description="Explicit environment overrides for the child process", ), - ] = EMPTY_STR_MAPPING + ] remove_keys: Annotated[ t.StrSequence, m.Field( diff --git a/src/flext_cli/_models/_defaults.py b/src/flext_cli/_models/_defaults.py index 1dc02465..39ec1a7a 100644 --- a/src/flext_cli/_models/_defaults.py +++ b/src/flext_cli/_models/_defaults.py @@ -7,8 +7,14 @@ from flext_cli import t -EMPTY_JSON_MAPPING: t.JsonMapping = MappingProxyType({}) -EMPTY_STR_MAPPING: t.StrMapping = MappingProxyType({}) +def empty_json_mapping() -> t.JsonMapping: + """Create an immutable empty JSON mapping for one model instance.""" + return MappingProxyType({}) + + +def empty_str_mapping() -> t.StrMapping: + """Create an immutable empty string mapping for one model instance.""" + return MappingProxyType({}) __all__: tuple[str, ...] = () diff --git a/src/flext_cli/_models/docx_document.py b/src/flext_cli/_models/docx_document.py index 12a0720a..0cc820a8 100644 --- a/src/flext_cli/_models/docx_document.py +++ b/src/flext_cli/_models/docx_document.py @@ -5,7 +5,7 @@ from typing import Annotated from flext_cli import t -from flext_cli._models._defaults import EMPTY_JSON_MAPPING +from flext_cli._models._defaults import empty_json_mapping from flext_core import m from .docx_styles import FlextCliModelsDocxStyles @@ -100,8 +100,7 @@ class DocxDocumentPlan(m.FrozenModel): default=(), strict=False, description="Document sections." ) core_properties: t.JsonMapping = m.Field( - default=EMPTY_JSON_MAPPING, - description="Core document properties.", + default_factory=empty_json_mapping, description="Core document properties." ) class DocxRenderRequest(m.FrozenModel): diff --git a/src/flext_cli/_models/pipeline.py b/src/flext_cli/_models/pipeline.py index 9ef55014..745ee27b 100644 --- a/src/flext_cli/_models/pipeline.py +++ b/src/flext_cli/_models/pipeline.py @@ -7,7 +7,7 @@ from typing import Annotated, ClassVar from flext_cli import c, p, t -from flext_cli._models._defaults import EMPTY_JSON_MAPPING +from flext_cli._models._defaults import empty_json_mapping from flext_core import m, u @@ -33,9 +33,10 @@ class PipelineStageContext(m.ContractModel): settings: Annotated[ t.JsonMapping, m.Field( + default_factory=empty_json_mapping, description="Immutable pipeline configuration", ), - ] = EMPTY_JSON_MAPPING + ] class PipelineStageSpec(m.ContractModel): """Declarative stage definition with dependency tracking.""" @@ -66,6 +67,7 @@ class PipelineStageSpec(m.ContractModel): Callable[[FlextCliModelsPipeline.PipelineStageContext], bool] | None, m.Field(description="Predicate — skip stage if returns True"), ] = None + class PipelineStageResult(m.ContractModel): """What a stage produces after execution.""" @@ -78,9 +80,9 @@ class PipelineStageResult(m.ContractModel): output: Annotated[ t.JsonMapping, m.Field( - description="Stage output payload", + default_factory=empty_json_mapping, description="Stage output payload" ), - ] = EMPTY_JSON_MAPPING + ] duration_ms: Annotated[ float, m.Field(description="Execution duration in milliseconds") ] = 0.0 diff --git a/src/flext_cli/_models/pptx_presentation.py b/src/flext_cli/_models/pptx_presentation.py index 04a1387a..ecf35f3e 100644 --- a/src/flext_cli/_models/pptx_presentation.py +++ b/src/flext_cli/_models/pptx_presentation.py @@ -5,7 +5,7 @@ from typing import Annotated from flext_cli import t -from flext_cli._models._defaults import EMPTY_JSON_MAPPING +from flext_cli._models._defaults import empty_json_mapping from flext_core import m @@ -23,8 +23,7 @@ class PptxPresentationPlan(m.FrozenModel): default=(), strict=False, description="Presentation slides." ) core_properties: t.JsonMapping = m.Field( - default=EMPTY_JSON_MAPPING, - description="Core document properties.", + default_factory=empty_json_mapping, description="Core document properties." ) class PptxRenderRequest(m.FrozenModel): diff --git a/src/flext_cli/_protocols/domain.py b/src/flext_cli/_protocols/domain.py index ae2bd261..f8fb69d3 100644 --- a/src/flext_cli/_protocols/domain.py +++ b/src/flext_cli/_protocols/domain.py @@ -13,9 +13,7 @@ class FlextCliProtocolsDomain: """CLI domain protocols layered on top of base callable contracts.""" - type ResultRouteHandler = Callable[ - ..., FlextCliProtocolsBase.ErasedCommandResult - ] + type ResultRouteHandler = Callable[..., FlextCliProtocolsBase.ErasedCommandResult] @runtime_checkable class JsonValueProcessor(Protocol): diff --git a/src/flext_cli/_utilities/_xlxx/xlsx_snapshot_structure.py b/src/flext_cli/_utilities/_xlxx/xlsx_snapshot_structure.py index b7e2680d..44aede30 100644 --- a/src/flext_cli/_utilities/_xlxx/xlsx_snapshot_structure.py +++ b/src/flext_cli/_utilities/_xlxx/xlsx_snapshot_structure.py @@ -64,7 +64,9 @@ def _snapshot_tables( f"Table snapshot failed: {detail}" ) return r[tuple[m.Cli.XlsxTableSnapshot, ...]].ok( - tuple(sorted(tables, key=FlextCliUtilitiesXlsxSnapshotStructure._table_name)) + tuple( + sorted(tables, key=FlextCliUtilitiesXlsxSnapshotStructure._table_name) + ) ) @staticmethod @@ -90,10 +92,7 @@ def _snapshot_rows( ) return r[tuple[m.Cli.XlsxRowDimensionSnapshot, ...]].ok( tuple( - sorted( - rows, - key=FlextCliUtilitiesXlsxSnapshotStructure._row_position, - ) + sorted(rows, key=FlextCliUtilitiesXlsxSnapshotStructure._row_position) ) ) @@ -124,8 +123,7 @@ def _snapshot_columns( return r[tuple[m.Cli.XlsxColumnDimensionSnapshot, ...]].ok( tuple( sorted( - columns, - key=FlextCliUtilitiesXlsxSnapshotStructure._column_position, + columns, key=FlextCliUtilitiesXlsxSnapshotStructure._column_position ) ) ) @@ -162,10 +160,7 @@ def _snapshot_names( ) return r[tuple[m.Cli.XlsxDefinedNameSnapshot, ...]].ok( tuple( - sorted( - names, - key=FlextCliUtilitiesXlsxSnapshotStructure._defined_name, - ) + sorted(names, key=FlextCliUtilitiesXlsxSnapshotStructure._defined_name) ) ) diff --git a/src/flext_cli/_utilities/pipeline.py b/src/flext_cli/_utilities/pipeline.py index c2233de8..c3f496e2 100644 --- a/src/flext_cli/_utilities/pipeline.py +++ b/src/flext_cli/_utilities/pipeline.py @@ -149,11 +149,7 @@ def _run_stage( return stage_result.model_copy(update={"duration_ms": duration_ms}) error = result.error or f"stage {spec.stage_id} failed" - log.error( - "stage_failed", - stage_id=spec.stage_id, - error=error, - ) + log.error("stage_failed", stage_id=spec.stage_id, error=error) return m.Cli.PipelineStageResult( stage_id=spec.stage_id, status=c.Cli.PipelineStageStatus.FAILED, diff --git a/src/flext_cli/services/_prompts_support.py b/src/flext_cli/services/_prompts_support.py index 03868bc6..416368e0 100644 --- a/src/flext_cli/services/_prompts_support.py +++ b/src/flext_cli/services/_prompts_support.py @@ -18,6 +18,13 @@ from collections.abc import Callable +class _GetpassReader: + """Invoke the platform password reader without descriptor binding.""" + + def __call__(self, prompt: str) -> str: + return getpass.getpass(prompt) + + class FlextCliPromptsSupport(s): """Support owner for prompt runtime state, logging, and input readers.""" @@ -28,7 +35,7 @@ class FlextCliPromptsSupport(s): _input_reader: t.Cli.PromptTextReader = m.PrivateAttr(input) - _password_reader: t.Cli.PromptTextReader = m.PrivateAttr(getpass.getpass) + _password_reader: t.Cli.PromptTextReader = m.PrivateAttr(_GetpassReader()) _test_env_override: bool | None = m.PrivateAttr(None) diff --git a/tests/unit/_cases/test_pipeline/testsflextclipipeline_part_02.py b/tests/unit/_cases/test_pipeline/testsflextclipipeline_part_02.py index 326c83f9..229c346c 100644 --- a/tests/unit/_cases/test_pipeline/testsflextclipipeline_part_02.py +++ b/tests/unit/_cases/test_pipeline/testsflextclipipeline_part_02.py @@ -92,22 +92,5 @@ def test_total_duration_tracked(self, tmp_path: Path) -> None: tm.ok(result) tm.that(result.value.total_duration_ms, gte=0.0) - def test_stage_raise_marks_pipeline_result_failed(self, tmp_path: Path) -> None: - """A stage handler that raises produces a failed overall pipeline result.""" - error_message = "intentional explosion" - - def exploding( - _ctx: p.Cli.PipelineStageContext, - ) -> p.Result[m.Cli.PipelineStageResult]: - raise ValueError(error_message) - - result = cli.pipeline( - [cli.stage("boom", handler=exploding)], context=cli.stage_context(tmp_path) - ) - - tm.fail(result) - tm.that(result.failure, eq=True) - tm.that(bool(result), eq=False) - __all__: list[str] = ["TestsFlextCliPipeline"]