msgspec_config

146class DataModel(msgspec.Struct, metaclass=DataModelMeta):
147    """Base typed settings model with optional datasource composition.
148
149    ``DataModel`` instances may be built from ordered datasource patches plus
150    constructor keyword overrides. Source-level unmapped values and unknown
151    constructor kwargs can be inspected with :meth:`get_unmapped_payload`.
152    """
153
154    __json_encoder__: ClassVar[msgspec.json.Encoder] = get_json_encoder()
155    __json_decoder__: ClassVar[msgspec.json.Decoder | None] = None
156    __schema__: ClassVar[dict[str, Any] | None] = None
157    __datasource_defs__: ClassVar[tuple["DataSource", ...] | None] = None
158
159    @classmethod
160    def _split_payload_for_convert(
161        cls,
162        payload: Mapping[str, Any],
163    ) -> tuple[dict[str, Any], dict[str, Any]]:
164        """Split payload keys into model-mapped and unmapped dictionaries.
165
166        Args:
167            payload: Raw mapping payload.
168
169        Returns:
170            Tuple ``(mapped, unmapped)`` where ``mapped`` contains only
171            model-recognized keys normalized to encoded field names, and
172            ``unmapped`` contains keys not recognized by ``cls``.
173        """
174        return split_mapping_by_model_fields(payload, cls)
175
176    @classmethod
177    def _split_constructor_kwargs(
178        cls,
179        payload: Mapping[str, Any],
180    ) -> tuple[dict[str, Any], dict[str, Any]]:
181        """Split constructor kwargs into mapped and unmapped top-level keys.
182
183        Args:
184            payload: Constructor kwargs mapping.
185
186        Returns:
187            Tuple ``(mapped, unmapped)`` where ``mapped`` uses encoded
188            top-level field names.
189        """
190        return split_top_level_mapping_by_model_fields(payload, cls)
191
192    @classmethod
193    def _prepare_payload_for_convert(
194        cls,
195        payload: Mapping[str, Any],
196    ) -> dict[str, Any]:
197        """Normalize payload keys to model-accepted field names.
198
199        Args:
200            payload: Raw mapping payload.
201
202        Returns:
203            Mapping filtered to recognized model keys and normalized to encoded
204            field names.
205        """
206        mapped, _ = cls._split_payload_for_convert(payload)
207        return mapped
208
209    @staticmethod
210    def _setup_instance(
211        instance: "DataModel",
212        datasource_instances: tuple["DataSource", ...] | None = None,
213        unmapped_cache: dict[str, Any] | None = None,
214        constructor_unmapped: Mapping[str, Any] | None = None,
215    ) -> "DataModel":
216        """Attach runtime datasource and unmapped state to one instance.
217
218        Args:
219            instance: Model instance to mutate.
220            datasource_instances: Runtime source instances used to build
221                ``instance``.
222            unmapped_cache: Precomputed unmapped cache, or ``None`` to defer
223                computation until unmapped payload access.
224            constructor_unmapped: Unmapped key/value pairs provided directly as
225                constructor kwargs.
226
227        Returns:
228            Same ``instance`` with runtime attributes initialized.
229        """
230        if datasource_instances is None:
231            datasource_instances = ()
232        instance.__datasource_instances__ = datasource_instances
233        instance.__unmapped_cache__ = unmapped_cache
234        instance.__constructor_unmapped__ = (
235            constructor_unmapped if constructor_unmapped is not None else {}
236        )
237        return instance
238
239    @classmethod
240    def from_data(cls, data: dict[str, Any]) -> Self:
241        """Build an instance from a plain mapping payload.
242
243        Args:
244            data: Input mapping to validate against ``cls``.
245
246        Returns:
247            Validated model instance with empty runtime datasource state.
248            Keys not recognized by ``cls`` are ignored.
249        """
250        prepared = cls._prepare_payload_for_convert(data)
251        instance = msgspec.convert(prepared, type=cls)
252        return cls._setup_instance(instance)
253
254    @classmethod
255    def from_json(cls, json_str: str | bytes) -> Self:
256        """Build an instance from a JSON payload.
257
258        Args:
259            json_str: JSON string or bytes.
260
261        Returns:
262            Decoded model instance with empty runtime datasource state.
263            Keys not recognized by ``cls`` are ignored.
264
265        Raises:
266            TypeError: If decoded JSON is not an object mapping.
267        """
268        raw = msgspec.json.decode(json_str)
269        if not isinstance(raw, Mapping):
270            raise TypeError("JSON payload must decode to an object mapping")
271        prepared = cls._prepare_payload_for_convert(raw)
272        instance = msgspec.convert(prepared, type=cls)
273        return cls._setup_instance(instance)
274
275    @classmethod
276    def model_json_schema(cls, indent: int = 0) -> str:
277        """Serialize the model JSON schema.
278
279        Args:
280            indent: Pretty-print indentation level. ``0`` disables formatting.
281
282        Returns:
283            JSON schema string representation.
284        """
285        if cls.__schema__ is None:
286            cls.__schema__ = msgspec.json.schema(cls)
287        json_str = cls.__json_encoder__.encode(cls.__schema__).decode()
288        if indent > 0:
289            return msgspec.json.format(json_str, indent=indent)
290        return json_str
291
292    @classmethod
293    def get_datasources_payload(
294        cls,
295        *datasource_args: "DataSource",
296        **kwargs: Any,
297    ) -> Mapping[str, Any]:
298        """Merge payloads from datasource instances and explicit keyword values.
299
300        Datasources are evaluated left-to-right. Explicit ``kwargs`` are merged
301        last and therefore have highest precedence.
302
303        Args:
304            *datasource_args: Datasource instances to execute.
305            **kwargs: Explicit field overrides. Unknown keys are ignored.
306
307        Returns:
308            Merged mapping ready for ``msgspec.convert``.
309
310        Raises:
311            TypeError: If any datasource returns a non-mapping value.
312        """
313        prepared_kwargs = cls._prepare_payload_for_convert(kwargs)
314        merged_data, _ = cls._collect_datasources_payload(
315            *datasource_args, **prepared_kwargs
316        )
317        return merged_data
318
319    @classmethod
320    def _collect_datasources_payload(
321        cls,
322        *datasource_defs: "DataSource",
323        **kwargs: Any,
324    ) -> tuple[dict[str, Any], tuple["DataSource", ...]]:
325        """Clone datasource definitions, load them, and merge payloads.
326
327        Args:
328            *datasource_defs: Datasource template instances bound on the class.
329            **kwargs: Explicit mapped keyword overrides merged after source
330                payloads.
331
332        Returns:
333            Tuple ``(merged_payload, datasource_instances)`` where
334            ``datasource_instances`` are the cloned runtime instances used for
335            this build.
336
337        Raises:
338            TypeError: If any source returns a non-mapping value.
339        """
340        merged_data: dict[str, Any] = {}
341        datasource_instances: list["DataSource"] = []
342
343        for datasource_def in datasource_defs:
344            datasource_instance = datasource_def.clone()
345            datasource_data = datasource_instance.resolve(model=cls)
346            if not isinstance(datasource_data, Mapping):
347                raise TypeError(
348                    f"DataSource {datasource_def.__class__.__name__} returned a "
349                    "non-mapping value"
350                )
351
352            deep_merge_into(merged_data, datasource_data)
353            datasource_instances.append(datasource_instance)
354
355        if kwargs:
356            deep_merge_into(merged_data, kwargs)
357
358        return merged_data, tuple(datasource_instances)
359
360    def model_dump(self) -> dict[str, Any]:
361        """Serialize this instance into Python builtins.
362
363        Returns:
364            ``dict`` representation containing model field values.
365        """
366        return msgspec.to_builtins(self)
367
368    def model_dump_json(self, indent: int = 0) -> str:
369        """Serialize this instance to JSON text.
370
371        Args:
372            indent: Pretty-print indentation level. ``0`` disables formatting.
373
374        Returns:
375            JSON string representation of the model.
376        """
377        json_str = self.__json_encoder__.encode(self).decode()
378        if indent > 0:
379            return msgspec.json.format(json_str, indent=indent)
380        return json_str
381
382    def get_unmapped_payload(self) -> dict[str, Any]:
383        """Return lazily merged unmapped payload from sources and kwargs.
384
385        Returns:
386            Deep-merged mapping of source-level unmapped keys (in source order)
387            plus constructor kwargs that were not recognized by the model.
388            Constructor unmapped keys are merged last. A shallow copy is
389            returned so callers cannot mutate the cache directly.
390        """
391        cached = getattr(self, "__unmapped_cache__", None)
392        if isinstance(cached, dict):
393            return dict(cached)
394
395        merged: dict[str, Any] = {}
396        datasource_instances = getattr(self, "__datasource_instances__", ())
397        for datasource in datasource_instances:
398            source_unmapped = getattr(datasource, "__unmapped_kwargs__", None)
399            if isinstance(source_unmapped, Mapping) and source_unmapped:
400                deep_merge_into(merged, source_unmapped)
401
402        constructor_unmapped = getattr(self, "__constructor_unmapped__", None)
403        if isinstance(constructor_unmapped, Mapping) and constructor_unmapped:
404            deep_merge_into(merged, constructor_unmapped)
405
406        self.__unmapped_cache__ = merged
407        return dict(merged)

410class DataSource(DataModel):
411    """Base class for configuration sources that emit mapping patches.
412
413    Subclasses implement :meth:`load`. Use :meth:`resolve` to run the full
414    source lifecycle (reset, normalize return shape, mapped/unmapped split).
415    """
416
417    def _normalize_load_result(
418        self,
419        result: Mapping[str, Any] | tuple[Mapping[str, Any], Mapping[str, Any]],
420    ) -> tuple[Mapping[str, Any], Mapping[str, Any] | None]:
421        """Normalize ``load`` output into ``(payload, unmapped_override)``.
422
423        Args:
424            result: Return value from ``load``.
425
426        Returns:
427            Tuple containing payload mapping and optional unmapped override.
428
429        Raises:
430            TypeError: If the return value does not follow the expected shape.
431        """
432        if isinstance(result, tuple):
433            if len(result) != 2:
434                raise TypeError(
435                    f"DataSource {self.__class__.__name__} load() must return "
436                    "Mapping or tuple[Mapping, Mapping]"
437                )
438            payload, unmapped = result
439            if not isinstance(payload, Mapping) or not isinstance(unmapped, Mapping):
440                raise TypeError(
441                    f"DataSource {self.__class__.__name__} load() tuple must be "
442                    "tuple[Mapping, Mapping]"
443                )
444            return payload, unmapped
445
446        if not isinstance(result, Mapping):
447            raise TypeError(
448                f"DataSource {self.__class__.__name__} load() must return "
449                "Mapping or tuple[Mapping, Mapping]"
450            )
451        return result, None
452
453    def _reset_instance(self) -> None:
454        """Reset per-load runtime state on this source instance.
455
456        This clears any previously collected unmapped keys.
457
458        Returns:
459            ``None``.
460        """
461        self.__unmapped_kwargs__ = {}
462
463    def _set_unmapped(self, unmapped: Mapping[str, Any]) -> None:
464        """Store unmapped payload values on this source instance.
465
466        Args:
467            unmapped: Mapping of keys that could not be mapped to model fields.
468
469        Returns:
470            ``None``.
471        """
472        self.__unmapped_kwargs__ = dict(unmapped)
473
474    def _split_payload_against_model(
475        self,
476        payload: Mapping[str, Any],
477        model: type[msgspec.Struct] | None,
478    ) -> tuple[dict[str, Any], dict[str, Any]]:
479        """Split one payload into mapped and unmapped sections.
480
481        Args:
482            payload: Raw payload produced by a source.
483            model: Target model type used for key recognition.
484
485        Returns:
486            Tuple ``(mapped, unmapped)``. ``mapped`` uses encoded field names
487            when ``model`` provides aliases.
488        """
489        if model is None:
490            return dict(payload), {}
491        return split_mapping_by_model_fields(payload, model)
492
493    def _finalize_payload(
494        self,
495        payload: Mapping[str, Any],
496        model: type[msgspec.Struct] | None,
497        unmapped_override: Mapping[str, Any] | None = None,
498    ) -> dict[str, Any]:
499        """Finalize a source payload and persist source unmapped state.
500
501        Args:
502            payload: Raw source payload.
503            model: Target model type used for split logic.
504            unmapped_override: Optional unmapped mapping to merge on top of
505                unmapped values inferred from ``payload``.
506
507        Returns:
508            Model-mapped payload fragment that should be merged into model input.
509        """
510        mapped, unmapped = self._split_payload_against_model(payload, model)
511
512        if unmapped_override is not None:
513            merged_unmapped = dict(unmapped)
514            deep_merge_into(merged_unmapped, unmapped_override)
515            self._set_unmapped(merged_unmapped)
516        else:
517            self._set_unmapped(unmapped)
518
519        return mapped
520
521    def load(
522        self,
523        model: type[DataModel] | None = None,
524    ) -> Mapping[str, Any] | tuple[Mapping[str, Any], Mapping[str, Any]]:
525        """Return raw source data before mapping finalization.
526
527        Subclasses should override this method to implement source-specific
528        loading behavior. :meth:`resolve` performs instance reset,
529        return-shape normalization, and finalize/split logic.
530
531        Args:
532            model: Optional target model requesting data.
533
534        Returns:
535            Either:
536            - mapping payload to be finalized against ``model``, or
537            - tuple ``(payload, unmapped_override)``.
538        """
539        raise NotImplementedError
540
541    def resolve(self, model: type[DataModel] | None = None) -> dict[str, Any]:
542        """Load and finalize source data for safe model merging.
543
544        Args:
545            model: Optional target model requesting data.
546
547        Returns:
548            Model-mapped payload fragment ready for merge. When ``model`` is
549            provided, output keys are normalized to encoded field names.
550        """
551        return self._load(model)
552
553    def _load(self, model: type[DataModel] | None = None) -> dict[str, Any]:
554        """Execute source loading lifecycle for one model resolution.
555
556        This internal wrapper clears per-instance runtime state, calls
557        :meth:`load`, validates the return shape, and finalizes the mapped
558        payload while persisting unmapped state.
559
560        Args:
561            model: Optional target model requesting data.
562
563        Returns:
564            Model-mapped payload fragment ready for merge.
565
566        Raises:
567            TypeError: If ``load`` returns an invalid payload shape.
568        """
569        self._reset_instance()
570        payload, unmapped_override = self._normalize_load_result(self.load(model))
571        return self._finalize_payload(
572            payload, model, unmapped_override=unmapped_override
573        )
574
575    def clone(self) -> Self:
576        """Clone this datasource configuration.
577
578        Returns:
579            Deep-copied datasource instance suitable for per-model execution.
580        """
581        return copy.deepcopy(self)

584def datasources(*datasource_args: DataSource):
585    """Bind datasource template definitions to a ``DataModel`` class.
586
587    Args:
588        *datasource_args: Datasource templates evaluated during model
589            instantiation.
590
591    Returns:
592        Decorator that writes cloned datasource templates to
593        ``cls.__datasource_defs__``.
594    """
595
596    def decorator(cls: type[DataModel]) -> type[DataModel]:
597        """Attach datasource templates to one model class.
598
599        Args:
600            cls: Model class being decorated.
601
602        Returns:
603            Same class with datasource definitions attached.
604        """
605        if datasource_args:
606            cls.__datasource_defs__ = tuple(
607                datasource.clone() for datasource in datasource_args
608            )
609        else:
610            cls.__datasource_defs__ = None
611        return cls
612
613    return decorator

189def entry(value: Any = NODEFAULT, *, name: str | None = None, **kwargs: Any) -> Any:
190    """Declare one model field with optional ``Meta`` kwargs.
191
192    Behavior:
193    - Without extra kwargs, returns ``msgspec.field(...)`` directly.
194    - With extra kwargs, returns an internal sentinel so the metaclass can
195      inject ``Annotated[..., Meta(...)]`` metadata.
196
197    Mutable defaults (``list``, ``dict``, ``set``) are converted to default
198    factories to avoid shared state across instances.
199
200    Args:
201        value: Field default value.
202        name: Optional encoded field name (``msgspec.field(name=...)``).
203        **kwargs: ``msgspec.Meta`` arguments plus extra schema keys
204            (``hidden_if``, ``disabled_if``, ``parent_group``,
205            ``ui_component``), CLI include/exclude key (``cli``), and CLI
206            override keys (``cli_flag``, ``cli_short_flag``). Extra keys are
207            stored under ``Meta.extra_json_schema``.
208
209    Returns:
210        ``msgspec.field`` output or an ``EntryInfo`` sentinel.
211    """
212    field_kwargs: dict[str, Any] = {}
213    if value is not NODEFAULT:
214        default, default_factory = _to_field_default(value)
215        field_kwargs["default"] = default
216        field_kwargs["default_factory"] = default_factory
217    if name is not None:
218        field_kwargs["name"] = name
219
220    if not kwargs:
221        return field(**field_kwargs)
222
223    return EntryInfo(
224        default=field_kwargs.get("default", NODEFAULT),
225        default_factory=field_kwargs.get("default_factory", NODEFAULT),
226        name=field_kwargs.get("name"),
227        meta_kwargs=kwargs,
228    )

350def group(*, collapsed: bool = False, mutable: bool = False) -> GroupInfo:
351    """Declare a grouped field inferred from its type annotation.
352
353    Supported annotation shapes:
354    - object types with zero-argument constructor
355    - ``list[...]``
356    - ``dict[...]``
357
358    Args:
359        collapsed: Whether UI consumers should render this group collapsed.
360        mutable: Whether UI consumers should treat this group as mutable.
361
362    Returns:
363        ``GroupInfo`` sentinel consumed during metaclass rewriting.
364
365    Raises:
366        TypeError: If ``collapsed`` or ``mutable`` are not bool.
367    """
368    if type(collapsed) is not bool:
369        raise TypeError(
370            f"group() 'collapsed' must be bool, got {type(collapsed).__name__}"
371        )
372    if type(mutable) is not bool:
373        raise TypeError(f"group() 'mutable' must be bool, got {type(mutable).__name__}")
374    return GroupInfo(collapsed=collapsed, mutable=mutable)

14class APISource(DataSource):
15    """Load configuration data by calling an HTTP JSON endpoint.
16
17    Attributes:
18        api_url: Endpoint URL used for the GET request.
19        header_name: Optional request header name (for auth, etc.).
20        header_value: Optional request header value.
21        root_node: Optional top-level JSON key to unwrap from response payload.
22        timeout_seconds: Request timeout in seconds.
23    """
24
25    api_url: str | None = None
26    header_name: str | None = None
27    header_value: str | None = None
28    root_node: str | None = None
29    timeout_seconds: float = 10.0
30
31    def load(self, model: type[msgspec.Struct] | None = None) -> Mapping[str, Any]:
32        """Fetch and decode endpoint JSON payload.
33
34        Args:
35            model: Optional target model requesting data. Accepted for interface
36                compatibility.
37
38        Returns:
39            Parsed mapping payload, or an empty mapping when URL is unset or
40            selected payload is missing/non-mapping.
41
42        Raises:
43            ValueError: If only one of ``header_name``/``header_value`` is set.
44            RuntimeError: If request fails or response JSON is invalid.
45        """
46        if not self.api_url:
47            return {}
48
49        if (self.header_name is None) != (self.header_value is None):
50            raise ValueError("header_name and header_value must be set together")
51
52        headers: dict[str, str] = {}
53        if self.header_name is not None and self.header_value is not None:
54            headers[self.header_name] = self.header_value
55
56        request = Request(self.api_url, headers=headers, method="GET")
57
58        try:
59            with urlopen(request, timeout=self.timeout_seconds) as response:
60                raw_data = response.read()
61        except (HTTPError, URLError, TimeoutError, OSError) as exc:
62            raise RuntimeError(f"Failed to fetch API endpoint: {self.api_url}") from exc
63
64        try:
65            data: Any = msgspec.json.decode(raw_data)
66        except Exception as exc:
67            raise RuntimeError(
68                f"Failed to parse API response JSON: {self.api_url}"
69            ) from exc
70
71        if self.root_node:
72            if not isinstance(data, Mapping):
73                return {}
74            data = data.get(self.root_node)
75
76        if not isinstance(data, Mapping):
77            return {}
78        return data

363class CliSource(DataSource):
364    """Load configuration values from command-line arguments.
365
366    Attributes:
367        cli_args: Optional argument list. Uses ``sys.argv[1:]`` when unset.
368        autogenerate: Emit generated options for fields without explicit CLI
369            metadata when true.
370        kebab_case: Use kebab-case long flags when true.
371        theme: Rich-click theme name passed via context settings.
372
373    Notes:
374        CLI declarations are generated from model fields and aliases.
375        Per-field behavior can be controlled via
376        ``entry(..., cli=..., cli_flag=..., cli_short_flag=...)`` metadata.
377        Precedence is:
378        - ``cli=False``: exclude field.
379        - ``cli=True``: include field.
380        - explicit ``cli_flag``/``cli_short_flag``: include field.
381        - otherwise include based on ``autogenerate``.
382    """
383
384    cli_args: list[str] | None = None
385    autogenerate: bool = True
386    kebab_case: bool = True
387    theme: str = "cargo-slim"
388
389    def load(
390        self,
391        model: type[msgspec.Struct] | None = None,
392    ) -> dict[str, Any] | tuple[dict[str, Any], dict[str, Any]]:
393        """Parse command-line options into model-shaped nested data.
394
395        Args:
396            model: Target model used to generate options and coerce values.
397
398        Returns:
399            Either:
400            - nested mapping of explicitly provided values, or
401            - tuple ``(mapped, unmapped)`` when unknown CLI args are present.
402            Mapped keys use encoded field names so payloads can be passed
403            directly to ``msgspec.convert`` for aliased models.
404            Field inclusion is controlled by ``autogenerate`` and per-field
405            ``cli`` metadata. ``cli_flag`` and ``cli_short_flag`` override
406            emitted declarations for included fields. Automatic short-flag
407            assignment is applied only when ``autogenerate=True``.
408
409        Raises:
410            TypeError: On generated option-name collisions.
411            click.BadParameter: On invalid literal coercion.
412            SystemExit: For help/exit pathways raised by click.
413        """
414        if model is None:
415            return {}
416
417        flat_with_alias = flatten_model_fields_with_alias(model)
418        if not flat_with_alias:
419            return {}
420
421        path_to_type: dict[str, Any] = {}
422        for canonical_path, field_meta in flat_with_alias.items():
423            alias_path, field_type = field_meta
424            path_to_type[canonical_path] = field_type
425            path_to_type[alias_path] = field_type
426
427        params: list[click.Parameter] = []
428        param_to_path: dict[str, str] = {}
429        decl_to_path: dict[str, str] = {}
430        primary_long_flag_by_path: dict[str, str] = {}
431        bool_neg_map: dict[str, str] = {}
432        json_struct_params: dict[str, str] = {}
433        use_kebab = self.kebab_case
434
435        def _register_decl(decl: str, dotted_path: str) -> None:
436            """Track one emitted option declaration and detect collisions.
437
438            Args:
439                decl: Option declaration token such as ``"--host"``.
440                dotted_path: Canonical model field path for ``decl``.
441
442            Returns:
443                ``None``.
444
445            Raises:
446                TypeError: If ``decl`` is already associated with another path.
447            """
448            existing_path = decl_to_path.get(decl)
449            if existing_path is not None and existing_path != dotted_path:
450                raise TypeError(
451                    f"CLI option declaration collision: '{existing_path}' and "
452                    f"'{dotted_path}' both map to '{decl}'"
453                )
454            decl_to_path[decl] = dotted_path
455
456        # Collect top-level Struct fields to expose JSON options and policy.
457        struct_field_names: dict[
458            str,
459            tuple[str, str, Any, bool | None, str | None, str | None],
460        ] = {}
461        for field_info in struct_fields(model):
462            if get_struct_subtype(field_info.type) is None:
463                continue
464            encode_name = getattr(field_info, "encode_name", field_info.name)
465            if not isinstance(encode_name, str) or not encode_name:
466                encode_name = field_info.name
467            custom_enabled, custom_long, custom_short = _extract_custom_cli_metadata(
468                field_info.type, field_info.name
469            )
470            struct_field_names[field_info.name] = (
471                field_info.name,
472                encode_name,
473                field_info.type,
474                custom_enabled,
475                custom_long,
476                custom_short,
477            )
478
479        emitted_json_opts: set[str] = set()
480
481        ctx_settings: dict[str, Any] = {
482            "help_option_names": ["--help", "-?"],
483            "ignore_unknown_options": True,
484            "allow_extra_args": True,
485            "rich_help_config": {
486                "theme": self.theme,
487                "enable_theme_env_var": False,
488            },
489        }
490
491        reserved_short: set[str] = set(
492            item.replace("-", "") for item in ctx_settings["help_option_names"]
493        )
494        assigned_short: set[str] = set()
495
496        for dotted_path, field_meta in flat_with_alias.items():
497            alias_path, field_type = field_meta
498            top_field = dotted_path.split(".")[0]
499
500            if top_field in struct_field_names and top_field not in emitted_json_opts:
501                emitted_json_opts.add(top_field)
502                (
503                    _,
504                    top_alias,
505                    _,
506                    top_custom_enabled,
507                    top_custom_long,
508                    top_custom_short,
509                ) = struct_field_names[top_field]
510                include_top_json = _should_include_field(
511                    self.autogenerate,
512                    top_custom_enabled,
513                    top_custom_long,
514                    top_custom_short,
515                )
516                if include_top_json:
517                    if top_custom_long is not None:
518                        top_long_flags = [top_custom_long]
519                    else:
520                        top_long_flags = [
521                            _make_flag_name(top_field, kebab_case=use_kebab)
522                        ]
523                        alias_top_flag = _make_flag_name(
524                            top_alias, kebab_case=use_kebab
525                        )
526                        if alias_top_flag not in top_long_flags:
527                            top_long_flags.append(alias_top_flag)
528                        top_long_flags = dedupe_keep_order(top_long_flags)
529
530                    json_param_name = top_field.replace(".", "_").replace("-", "_")
531                    json_struct_params[json_param_name] = top_alias
532                    json_decls = list(top_long_flags)
533                    json_decls.append(json_param_name)
534                    json_decls = dedupe_keep_order(json_decls)
535
536                    if top_custom_short is not None:
537                        short = _reserve_short(
538                            top_custom_short,
539                            reserved_short,
540                            assigned_short,
541                            top_field,
542                        )
543                    elif self.autogenerate:
544                        short = _assign_short(
545                            top_long_flags[0], reserved_short, assigned_short
546                        )
547                    else:
548                        short = None
549                    if short is not None:
550                        json_decls.insert(0, "-" + short)
551
552                    for decl in json_decls:
553                        if decl.startswith("-"):
554                            _register_decl(decl, top_field)
555
556                    params.append(
557                        click.Option(
558                            param_decls=json_decls,
559                            type=click.STRING,
560                            help="JSON string",
561                        )
562                    )
563
564            top_struct = struct_field_names.get(top_field)
565            if top_struct is not None:
566                _, _, _, top_custom_enabled, _, _ = top_struct
567                if top_custom_enabled is False:
568                    continue
569
570            custom_enabled, custom_long, custom_short = _extract_custom_cli_metadata(
571                field_type, dotted_path
572            )
573            include_leaf = _should_include_field(
574                self.autogenerate,
575                custom_enabled,
576                custom_long,
577                custom_short,
578            )
579            if not include_leaf:
580                continue
581
582            if custom_long is not None:
583                long_flags = [custom_long]
584            else:
585                long_flags = [_make_flag_name(dotted_path, kebab_case=use_kebab)]
586                alias_flag = _make_flag_name(alias_path, kebab_case=use_kebab)
587                if alias_flag not in long_flags:
588                    long_flags.append(alias_flag)
589                long_flags = dedupe_keep_order(long_flags)
590
591            param_name = dotted_path.replace(".", "_").replace("-", "_")
592            existing_path = param_to_path.get(param_name)
593            if existing_path is not None and existing_path != alias_path:
594                raise TypeError(
595                    f"CLI option name collision: '{existing_path}' and "
596                    f"'{alias_path}' both map to '{param_name}'"
597                )
598            param_to_path[param_name] = alias_path
599            primary_long_flag_by_path[dotted_path] = long_flags[0]
600            primary_long_flag_by_path[alias_path] = long_flags[0]
601
602            for decl in long_flags:
603                _register_decl(decl, dotted_path)
604
605            click_kwargs = _python_type_to_click(field_type)
606            is_bool_flag = bool(click_kwargs.pop("is_bool_flag", False))
607
608            if is_bool_flag:
609                pos_decls = list(long_flags)
610                pos_decls.append(param_name)
611                if custom_short is not None:
612                    short = _reserve_short(
613                        custom_short,
614                        reserved_short,
615                        assigned_short,
616                        dotted_path,
617                    )
618                    pos_decls.insert(0, "-" + short)
619                elif self.autogenerate and "." not in dotted_path:
620                    short = _assign_short(long_flags[0], reserved_short, assigned_short)
621                    if short is not None:
622                        pos_decls.insert(0, "-" + short)
623                pos_decls = dedupe_keep_order(pos_decls)
624                params.append(
625                    click.Option(
626                        param_decls=pos_decls,
627                        is_flag=True,
628                        flag_value=True,
629                    )
630                )
631
632                neg_param_name = "no_" + param_name
633                existing_neg = bool_neg_map.get(neg_param_name)
634                if existing_neg is not None and existing_neg != alias_path:
635                    raise TypeError(
636                        f"CLI bool negation collision: '{existing_neg}' and "
637                        f"'{alias_path}' both map to '{neg_param_name}'"
638                    )
639                bool_neg_map[neg_param_name] = alias_path
640
641                neg_decls = [f"--no-{flag[2:]}" for flag in long_flags]
642                neg_decls.append(neg_param_name)
643                neg_decls = dedupe_keep_order(neg_decls)
644                for decl in neg_decls:
645                    _register_decl(decl, dotted_path)
646                params.append(
647                    click.Option(
648                        param_decls=neg_decls,
649                        is_flag=True,
650                        flag_value=True,
651                        hidden=True,
652                    )
653                )
654                continue
655
656            click_kwargs.setdefault("default", None)
657            click_kwargs["required"] = False
658
659            decls = list(long_flags)
660            decls.append(param_name)
661            if custom_short is not None:
662                short = _reserve_short(
663                    custom_short,
664                    reserved_short,
665                    assigned_short,
666                    dotted_path,
667                )
668                decls.insert(0, "-" + short)
669            elif self.autogenerate and "." not in dotted_path:
670                short = _assign_short(long_flags[0], reserved_short, assigned_short)
671                if short is not None:
672                    decls.insert(0, "-" + short)
673            decls = dedupe_keep_order(decls)
674
675            help_text = click_kwargs.pop("help", None) or ""
676            params.append(
677                click.Option(
678                    param_decls=decls,
679                    help=help_text or None,
680                    **click_kwargs,
681                )
682            )
683
684        command_name = _resolve_command_name()
685
686        epilog_parts: list[str] = []
687        if bool_neg_map:
688            epilog_parts.append(
689                "Untyped flags (toggles) can be negated with the --no- prefix "
690                "(e.g. --no-debug)."
691            )
692        if json_struct_params:
693            epilog_parts.append(
694                "Nested options accept a JSON string "
695                '(e.g. --log \'{"level": "DEBUG"}\').'
696            )
697
698        command = click.RichCommand(
699            name=command_name,
700            params=params,
701            context_settings=ctx_settings,
702            epilog="\n\n".join(epilog_parts) if epilog_parts else None,
703        )
704
705        args = list(self.cli_args) if self.cli_args is not None else sys.argv[1:]
706
707        try:
708            ctx: click.Context = command.make_context(command_name, list(args))
709        except click.exceptions.Exit as exc:
710            raise SystemExit(getattr(exc, "code", 0)) from None
711
712        extra_args = list(ctx.args)
713
714        raw_values: dict[str, Any] = ctx.params
715        result: dict[str, Any] = {}
716
717        # Pass 1: struct JSON options.
718        for param_name, value in raw_values.items():
719            if value is None or param_name not in json_struct_params:
720                continue
721            source = ctx.get_parameter_source(param_name)
722            if source is not None and source != click.core.ParameterSource.COMMANDLINE:
723                continue
724            decoded = try_json_decode(value)
725            if decoded is not _COERCE_FAILED and isinstance(decoded, dict):
726                set_nested(result, json_struct_params[param_name], decoded)
727
728        # Pass 2: scalar fields + bool negations (override JSON subkeys).
729        for param_name, value in raw_values.items():
730            if value is None or param_name in json_struct_params:
731                continue
732            source = ctx.get_parameter_source(param_name)
733            if source is not None and source != click.core.ParameterSource.COMMANDLINE:
734                continue
735
736            neg_path = bool_neg_map.get(param_name)
737            if neg_path is not None:
738                set_nested(result, neg_path, False)
739                continue
740
741            dotted_path = param_to_path.get(param_name)
742            if dotted_path is None:
743                continue
744
745            field_type = path_to_type.get(dotted_path)
746            if field_type is not None and isinstance(value, str):
747                unwrapped = unwrap_annotated(field_type)
748                coerced = coerce_env_value(value, field_type)
749                if coerced is _COERCE_FAILED and get_origin(unwrapped) is Literal:
750                    allowed = ", ".join(repr(item) for item in get_args(unwrapped))
751                    param_hint = primary_long_flag_by_path.get(
752                        dotted_path,
753                        _make_flag_name(dotted_path, kebab_case=use_kebab),
754                    )
755                    raise click.BadParameter(
756                        f"invalid literal value {value!r}; allowed values: {allowed}",
757                        param_hint=param_hint,
758                    )
759                if coerced is not _COERCE_FAILED:
760                    value = coerced
761
762            set_nested(result, dotted_path, value)
763
764        if extra_args:
765            return result, _parse_unmapped_cli_args(extra_args)
766        return result