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

174def entry(value: Any = NODEFAULT, *, name: str | None = None, **kwargs: Any) -> Any:
175    """Declare one model field with optional ``Meta`` kwargs.
176
177    Behavior:
178    - Without extra kwargs, returns ``msgspec.field(...)`` directly.
179    - With extra kwargs, returns an internal sentinel so the metaclass can
180      inject ``Annotated[..., Meta(...)]`` metadata.
181
182    Mutable defaults (``list``, ``dict``, ``set``) are converted to default
183    factories to avoid shared state across instances.
184
185    Args:
186        value: Field default value.
187        name: Optional encoded field name (``msgspec.field(name=...)``).
188        **kwargs: ``msgspec.Meta`` arguments plus supported UI schema keys.
189
190    Returns:
191        ``msgspec.field`` output or an ``EntryInfo`` sentinel.
192    """
193    field_kwargs: dict[str, Any] = {}
194    if value is not NODEFAULT:
195        default, default_factory = _to_field_default(value)
196        field_kwargs["default"] = default
197        field_kwargs["default_factory"] = default_factory
198    if name is not None:
199        field_kwargs["name"] = name
200
201    if not kwargs:
202        return field(**field_kwargs)
203
204    return EntryInfo(
205        default=field_kwargs.get("default", NODEFAULT),
206        default_factory=field_kwargs.get("default_factory", NODEFAULT),
207        name=field_kwargs.get("name"),
208        meta_kwargs=kwargs,
209    )

331def group(*, collapsed: bool = False, mutable: bool = False) -> GroupInfo:
332    """Declare a grouped field inferred from its type annotation.
333
334    Supported annotation shapes:
335    - object types with zero-argument constructor
336    - ``list[...]``
337    - ``dict[...]``
338
339    Args:
340        collapsed: Whether UI consumers should render this group collapsed.
341        mutable: Whether UI consumers should treat this group as mutable.
342
343    Returns:
344        ``GroupInfo`` sentinel consumed during metaclass rewriting.
345
346    Raises:
347        TypeError: If ``collapsed`` or ``mutable`` are not bool.
348    """
349    if type(collapsed) is not bool:
350        raise TypeError(
351            f"group() 'collapsed' must be bool, got {type(collapsed).__name__}"
352        )
353    if type(mutable) is not bool:
354        raise TypeError(f"group() 'mutable' must be bool, got {type(mutable).__name__}")
355    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

192class CliSource(DataSource):
193    """Load configuration values from command-line arguments.
194
195    Attributes:
196        cli_args: Optional argument list. Uses ``sys.argv[1:]`` when unset.
197        kebab_case: Use kebab-case long flags when true.
198        theme: Rich-click theme name passed via context settings.
199    """
200
201    cli_args: list[str] | None = None
202    kebab_case: bool = True
203    theme: str = "cargo-slim"
204
205    def load(
206        self,
207        model: type[msgspec.Struct] | None = None,
208    ) -> dict[str, Any] | tuple[dict[str, Any], dict[str, Any]]:
209        """Parse command-line options into model-shaped nested data.
210
211        Args:
212            model: Target model used to generate options and coerce values.
213
214        Returns:
215            Either:
216            - nested mapping of explicitly provided values, or
217            - tuple ``(mapped, unmapped)`` when unknown CLI args are present.
218            Mapped keys use encoded field names so payloads can be passed
219            directly to ``msgspec.convert`` for aliased models.
220
221        Raises:
222            TypeError: On generated option-name collisions.
223            click.BadParameter: On invalid literal coercion.
224            SystemExit: For help/exit pathways raised by click.
225        """
226        if model is None:
227            return {}
228
229        flat_with_alias = flatten_model_fields_with_alias(model)
230        if not flat_with_alias:
231            return {}
232
233        path_to_type: dict[str, Any] = {}
234        for canonical_path, field_meta in flat_with_alias.items():
235            alias_path, field_type = field_meta
236            path_to_type[canonical_path] = field_type
237            path_to_type[alias_path] = field_type
238
239        params: list[click.Parameter] = []
240        param_to_path: dict[str, str] = {}
241        decl_to_path: dict[str, str] = {}
242        bool_neg_map: dict[str, str] = {}
243        json_struct_params: dict[str, str] = {}
244        use_kebab = self.kebab_case
245
246        def _register_decl(decl: str, dotted_path: str) -> None:
247            """Track one emitted option declaration and detect collisions.
248
249            Args:
250                decl: Option declaration token such as ``"--host"``.
251                dotted_path: Canonical model field path for ``decl``.
252
253            Returns:
254                ``None``.
255
256            Raises:
257                TypeError: If ``decl`` is already associated with another path.
258            """
259            existing_path = decl_to_path.get(decl)
260            if existing_path is not None and existing_path != dotted_path:
261                raise TypeError(
262                    f"CLI option declaration collision: '{existing_path}' and "
263                    f"'{dotted_path}' both map to '{decl}'"
264                )
265            decl_to_path[decl] = dotted_path
266
267        # Collect top-level Struct fields to expose JSON options.
268        struct_field_names: dict[str, tuple[str, str]] = {}
269        for field_info in struct_fields(model):
270            if get_struct_subtype(field_info.type) is None:
271                continue
272            encode_name = getattr(field_info, "encode_name", field_info.name)
273            if not isinstance(encode_name, str) or not encode_name:
274                encode_name = field_info.name
275            struct_field_names[field_info.name] = (field_info.name, encode_name)
276
277        emitted_json_opts: set[str] = set()
278
279        ctx_settings: dict[str, Any] = {
280            "help_option_names": ["--help", "-?"],
281            "ignore_unknown_options": True,
282            "allow_extra_args": True,
283            "rich_help_config": {
284                "theme": self.theme,
285                "enable_theme_env_var": False,
286            },
287        }
288
289        reserved_short: set[str] = set(
290            item.replace("-", "") for item in ctx_settings["help_option_names"]
291        )
292        assigned_short: set[str] = set()
293
294        for dotted_path, field_meta in flat_with_alias.items():
295            alias_path, field_type = field_meta
296            top_field = dotted_path.split(".")[0]
297
298            if top_field in struct_field_names and top_field not in emitted_json_opts:
299                emitted_json_opts.add(top_field)
300                _, top_alias = struct_field_names[top_field]
301
302                top_long_flags = [_make_flag_name(top_field, kebab_case=use_kebab)]
303                alias_top_flag = _make_flag_name(top_alias, kebab_case=use_kebab)
304                if alias_top_flag not in top_long_flags:
305                    top_long_flags.append(alias_top_flag)
306                top_long_flags = dedupe_keep_order(top_long_flags)
307
308                json_param_name = top_field.replace(".", "_").replace("-", "_")
309                json_struct_params[json_param_name] = top_alias
310                json_decls = list(top_long_flags)
311                if not use_kebab:
312                    json_decls.append(json_param_name)
313                json_decls = dedupe_keep_order(json_decls)
314
315                short = _assign_short(top_long_flags[0], reserved_short, assigned_short)
316                if short is not None:
317                    json_decls.insert(0, "-" + short)
318
319                for decl in json_decls:
320                    if decl.startswith("-"):
321                        _register_decl(decl, top_field)
322
323                params.append(
324                    click.Option(
325                        param_decls=json_decls,
326                        type=click.STRING,
327                        help="JSON string",
328                    )
329                )
330
331            long_flags = [_make_flag_name(dotted_path, kebab_case=use_kebab)]
332            alias_flag = _make_flag_name(alias_path, kebab_case=use_kebab)
333            if alias_flag not in long_flags:
334                long_flags.append(alias_flag)
335            long_flags = dedupe_keep_order(long_flags)
336
337            param_name = dotted_path.replace(".", "_").replace("-", "_")
338            existing_path = param_to_path.get(param_name)
339            if existing_path is not None and existing_path != alias_path:
340                raise TypeError(
341                    f"CLI option name collision: '{existing_path}' and "
342                    f"'{alias_path}' both map to '{param_name}'"
343                )
344            param_to_path[param_name] = alias_path
345
346            for decl in long_flags:
347                _register_decl(decl, dotted_path)
348
349            click_kwargs = _python_type_to_click(field_type)
350            is_bool_flag = bool(click_kwargs.pop("is_bool_flag", False))
351
352            if is_bool_flag:
353                pos_decls = list(long_flags)
354                if not use_kebab:
355                    pos_decls.append(param_name)
356                if "." not in dotted_path:
357                    short = _assign_short(long_flags[0], reserved_short, assigned_short)
358                    if short is not None:
359                        pos_decls.insert(0, "-" + short)
360                pos_decls = dedupe_keep_order(pos_decls)
361                params.append(
362                    click.Option(
363                        param_decls=pos_decls,
364                        is_flag=True,
365                        flag_value=True,
366                    )
367                )
368
369                neg_param_name = "no_" + param_name
370                existing_neg = bool_neg_map.get(neg_param_name)
371                if existing_neg is not None and existing_neg != alias_path:
372                    raise TypeError(
373                        f"CLI bool negation collision: '{existing_neg}' and "
374                        f"'{alias_path}' both map to '{neg_param_name}'"
375                    )
376                bool_neg_map[neg_param_name] = alias_path
377
378                neg_decls = [f"--no-{flag[2:]}" for flag in long_flags]
379                if not use_kebab:
380                    neg_decls.append(neg_param_name)
381                neg_decls = dedupe_keep_order(neg_decls)
382                for decl in neg_decls:
383                    _register_decl(decl, dotted_path)
384                params.append(
385                    click.Option(
386                        param_decls=neg_decls,
387                        is_flag=True,
388                        flag_value=True,
389                        hidden=True,
390                    )
391                )
392                continue
393
394            click_kwargs.setdefault("default", None)
395            click_kwargs["required"] = False
396
397            decls = list(long_flags)
398            if not use_kebab:
399                decls.append(param_name)
400            if "." not in dotted_path:
401                short = _assign_short(long_flags[0], reserved_short, assigned_short)
402                if short is not None:
403                    decls.insert(0, "-" + short)
404            decls = dedupe_keep_order(decls)
405
406            help_text = click_kwargs.pop("help", None) or ""
407            params.append(
408                click.Option(
409                    param_decls=decls,
410                    help=help_text or None,
411                    **click_kwargs,
412                )
413            )
414
415        command_name = _resolve_command_name()
416
417        epilog_parts: list[str] = []
418        if bool_neg_map:
419            epilog_parts.append(
420                "Untyped flags (toggles) can be negated with the --no- prefix "
421                "(e.g. --no-debug)."
422            )
423        if json_struct_params:
424            epilog_parts.append(
425                "Nested options accept a JSON string "
426                '(e.g. --log \'{"level": "DEBUG"}\').'
427            )
428
429        command = click.RichCommand(
430            name=command_name,
431            params=params,
432            context_settings=ctx_settings,
433            epilog="\n\n".join(epilog_parts) if epilog_parts else None,
434        )
435
436        args = list(self.cli_args) if self.cli_args is not None else sys.argv[1:]
437
438        try:
439            ctx: click.Context = command.make_context(command_name, list(args))
440        except click.exceptions.Exit as exc:
441            raise SystemExit(getattr(exc, "code", 0)) from None
442
443        extra_args = list(ctx.args)
444
445        raw_values: dict[str, Any] = ctx.params
446        result: dict[str, Any] = {}
447
448        # Pass 1: struct JSON options.
449        for param_name, value in raw_values.items():
450            if value is None or param_name not in json_struct_params:
451                continue
452            source = ctx.get_parameter_source(param_name)
453            if source is not None and source != click.core.ParameterSource.COMMANDLINE:
454                continue
455            decoded = try_json_decode(value)
456            if decoded is not _COERCE_FAILED and isinstance(decoded, dict):
457                set_nested(result, json_struct_params[param_name], decoded)
458
459        # Pass 2: scalar fields + bool negations (override JSON subkeys).
460        for param_name, value in raw_values.items():
461            if value is None or param_name in json_struct_params:
462                continue
463            source = ctx.get_parameter_source(param_name)
464            if source is not None and source != click.core.ParameterSource.COMMANDLINE:
465                continue
466
467            neg_path = bool_neg_map.get(param_name)
468            if neg_path is not None:
469                set_nested(result, neg_path, False)
470                continue
471
472            dotted_path = param_to_path.get(param_name)
473            if dotted_path is None:
474                continue
475
476            field_type = path_to_type.get(dotted_path)
477            if field_type is not None and isinstance(value, str):
478                unwrapped = unwrap_annotated(field_type)
479                coerced = coerce_env_value(value, field_type)
480                if coerced is _COERCE_FAILED and get_origin(unwrapped) is Literal:
481                    allowed = ", ".join(repr(item) for item in get_args(unwrapped))
482                    raise click.BadParameter(
483                        f"invalid literal value {value!r}; allowed values: {allowed}",
484                        param_hint=_make_flag_name(dotted_path, kebab_case=use_kebab),
485                    )
486                if coerced is not _COERCE_FAILED:
487                    value = coerced
488
489            set_nested(result, dotted_path, value)
490
491        if extra_args:
492            return result, _parse_unmapped_cli_args(extra_args)
493        return result