msgspec_config

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

450class DataSource(DataModel):
451    """Base class for configuration sources that emit mapping patches.
452
453    Subclasses implement :meth:`load`. Use :meth:`resolve` to run the full
454    source lifecycle (reset, normalize return shape, mapped/unmapped split,
455    and runtime state persistence).
456    """
457
458    def _normalize_load_result(
459        self,
460        result: Mapping[str, Any] | tuple[Mapping[str, Any], Mapping[str, Any]],
461    ) -> tuple[Mapping[str, Any], Mapping[str, Any] | None]:
462        """Normalize ``load`` output into ``(payload, unmapped_override)``.
463
464        Args:
465            result: Return value from ``load``.
466
467        Returns:
468            Tuple containing payload mapping and optional unmapped override.
469
470        Raises:
471            TypeError: If the return value does not follow the expected shape.
472        """
473        if isinstance(result, tuple):
474            if len(result) != 2:
475                raise TypeError(
476                    f"DataSource {self.__class__.__name__} load() must return "
477                    "Mapping or tuple[Mapping, Mapping]"
478                )
479            payload, unmapped = result
480            if not isinstance(payload, Mapping) or not isinstance(unmapped, Mapping):
481                raise TypeError(
482                    f"DataSource {self.__class__.__name__} load() tuple must be "
483                    "tuple[Mapping, Mapping]"
484                )
485            return payload, unmapped
486
487        if not isinstance(result, Mapping):
488            raise TypeError(
489                f"DataSource {self.__class__.__name__} load() must return "
490                "Mapping or tuple[Mapping, Mapping]"
491            )
492        return result, None
493
494    def _reset_instance(self) -> None:
495        """Reset per-load runtime state on this source instance.
496
497        This clears any previously collected unmapped keys and raw argv
498        leftovers.
499
500        Returns:
501            ``None``.
502        """
503        msgspec.structs.force_setattr(self, "__unmapped_kwargs__", {})
504        msgspec.structs.force_setattr(self, "__raw_argv__", [])
505
506    def _set_unmapped(self, unmapped: Mapping[str, Any]) -> None:
507        """Store unmapped payload values on this source instance.
508
509        Args:
510            unmapped: Mapping of keys that could not be mapped to model fields.
511
512        Returns:
513            ``None``.
514        """
515        msgspec.structs.force_setattr(self, "__unmapped_kwargs__", dict(unmapped))
516
517    def _set_raw_argv(self, raw_argv: list[str]) -> None:
518        """Store raw argv leftovers on this source instance.
519
520        Args:
521            raw_argv: Ordered CLI tokens that were not consumed as mapped
522                options for the target model.
523        """
524        msgspec.structs.force_setattr(self, "__raw_argv__", list(raw_argv))
525
526    def _split_payload_against_model(
527        self,
528        payload: Mapping[str, Any],
529        model: type[msgspec.Struct] | None,
530    ) -> tuple[dict[str, Any], dict[str, Any]]:
531        """Split one payload into mapped and unmapped sections.
532
533        Args:
534            payload: Raw payload produced by a source.
535            model: Target model type used for key recognition.
536
537        Returns:
538            Tuple ``(mapped, unmapped)``. ``mapped`` uses encoded field names
539            when ``model`` provides aliases.
540        """
541        if model is None:
542            return dict(payload), {}
543        return split_mapping_by_model_fields(payload, model)
544
545    def _finalize_payload(
546        self,
547        payload: Mapping[str, Any],
548        model: type[msgspec.Struct] | None,
549        unmapped_override: Mapping[str, Any] | None = None,
550    ) -> dict[str, Any]:
551        """Finalize a source payload and persist source unmapped state.
552
553        Args:
554            payload: Raw source payload.
555            model: Target model type used for split logic.
556            unmapped_override: Optional unmapped mapping to merge on top of
557                unmapped values inferred from ``payload``.
558
559        Returns:
560            Model-mapped payload fragment that should be merged into model input.
561        """
562        mapped, unmapped = self._split_payload_against_model(payload, model)
563
564        if unmapped_override is not None:
565            merged_unmapped = dict(unmapped)
566            deep_merge_into(merged_unmapped, unmapped_override)
567            self._set_unmapped(merged_unmapped)
568        else:
569            self._set_unmapped(unmapped)
570
571        return mapped
572
573    def load(
574        self,
575        model: type[DataModel] | None = None,
576    ) -> Mapping[str, Any] | tuple[Mapping[str, Any], Mapping[str, Any]]:
577        """Return raw source data before mapping finalization.
578
579        Subclasses should override this method to implement source-specific
580        loading behavior. :meth:`resolve` performs instance reset,
581        return-shape normalization, and finalize/split logic.
582
583        Args:
584            model: Optional target model requesting data.
585
586        Returns:
587            Either:
588            - mapping payload to be finalized against ``model``, or
589            - tuple ``(payload, unmapped_override)``.
590        """
591        raise NotImplementedError
592
593    def resolve(self, model: type[DataModel] | None = None) -> dict[str, Any]:
594        """Load and finalize source data for safe model merging.
595
596        Args:
597            model: Optional target model requesting data.
598
599        Returns:
600            Model-mapped payload fragment ready for merge. When ``model`` is
601            provided, output keys are normalized to encoded field names.
602        """
603        return self._load(model)
604
605    def _load(self, model: type[DataModel] | None = None) -> dict[str, Any]:
606        """Execute source loading lifecycle for one model resolution.
607
608        This internal wrapper clears per-instance runtime state, calls
609        :meth:`load`, validates the return shape, and finalizes the mapped
610        payload while persisting runtime state.
611
612        Args:
613            model: Optional target model requesting data.
614
615        Returns:
616            Model-mapped payload fragment ready for merge.
617
618        Raises:
619            TypeError: If ``load`` returns an invalid payload shape.
620        """
621        self._reset_instance()
622        payload, unmapped_override = self._normalize_load_result(self.load(model))
623        return self._finalize_payload(
624            payload, model, unmapped_override=unmapped_override
625        )
626
627    def clone(self) -> Self:
628        """Clone this datasource configuration.
629
630        Returns:
631            Deep-copied datasource instance suitable for per-model execution.
632        """
633        return copy.deepcopy(self)

636def datasources(*datasource_args: DataSource):
637    """Bind datasource template definitions to a ``DataModel`` class.
638
639    Args:
640        *datasource_args: Datasource templates evaluated during model
641            instantiation.
642
643    Returns:
644        Decorator that writes cloned datasource templates to
645        ``cls.__datasource_defs__``.
646    """
647
648    def decorator(cls: type[DataModel]) -> type[DataModel]:
649        """Attach datasource templates to one model class.
650
651        Args:
652            cls: Model class being decorated.
653
654        Returns:
655            Same class with datasource definitions attached.
656        """
657        if datasource_args:
658            cls.__datasource_defs__ = tuple(
659                datasource.clone() for datasource in datasource_args
660            )
661        else:
662            cls.__datasource_defs__ = None
663        return cls
664
665    return decorator

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

394def group(
395    *, collapsed: bool = False, mutable: bool = False, **kwargs: Any
396) -> GroupInfo:
397    """Declare a grouped field inferred from its type annotation.
398
399    Supported annotation shapes:
400    - object types with zero-argument constructor
401    - ``list[...]``
402    - ``dict[...]``
403
404    Args:
405        collapsed: Whether UI consumers should render this group collapsed.
406        mutable: Whether UI consumers should treat this group as mutable.
407        **kwargs: ``msgspec.Meta`` parameters plus arbitrary schema keys.
408            ``Meta`` parameters are forwarded directly. Any non-``Meta`` key
409            is stored under ``Meta.extra_json_schema``.
410
411    Returns:
412        ``GroupInfo`` sentinel consumed during metaclass rewriting.
413
414    Raises:
415        TypeError: If ``collapsed`` or ``mutable`` are not bool.
416    """
417    if type(collapsed) is not bool:
418        raise TypeError(
419            f"group() 'collapsed' must be bool, got {type(collapsed).__name__}"
420        )
421    if type(mutable) is not bool:
422        raise TypeError(f"group() 'mutable' must be bool, got {type(mutable).__name__}")
423    return GroupInfo(collapsed=collapsed, mutable=mutable, meta_kwargs=dict(kwargs))

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