_options.py 8.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234
  1. # Copyright (c) 2025-2026 Buf Technologies, Inc.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. """Dataclass option parser for protoc plugin parameters."""
  15. from __future__ import annotations
  16. import dataclasses
  17. import enum
  18. import types
  19. import typing
  20. from typing import TYPE_CHECKING, Any, TypeVar, get_type_hints
  21. if TYPE_CHECKING:
  22. from _typeshed import DataclassInstance
  23. _Options = TypeVar("_Options", bound="DataclassInstance")
  24. def parse_options(cls: type[_Options], parameter: str) -> tuple[_Options, str]:
  25. """Parse a comma-separated `key=value` parameter string into a dataclass.
  26. Args:
  27. cls: A dataclass type whose fields define the schema.
  28. parameter: The `CodeGeneratorRequest.parameter` string.
  29. Returns:
  30. A tuple of (instance of `cls`, unparsed parameter string). The
  31. unparsed string contains comma-separated tokens for keys that do
  32. not correspond to any field in `cls`.
  33. Raises:
  34. ValueError: On missing required fields, type coercion failures,
  35. or bare keys for non-bool fields.
  36. """
  37. # Resolve string annotations (from `from __future__ import annotations`)
  38. # into actual types. field.type is unreliable for this.
  39. hints = get_type_hints(cls)
  40. field_by_key = {f.metadata.get("name", f.name): f for f in dataclasses.fields(cls)}
  41. # Parse the raw parameter string into (key, value | None) pairs.
  42. # An empty string produces no pairs.
  43. raw_pairs: list[tuple[str, str | None]] = []
  44. if parameter:
  45. for token in parameter.split(","):
  46. if "=" in token:
  47. key, _, value = token.partition("=")
  48. raw_pairs.append((key.strip(), value))
  49. else:
  50. raw_pairs.append((token.strip(), None))
  51. # Separate known and unknown keys, preserving original tokens for
  52. # unknown keys.
  53. unknown_tokens: list[str] = []
  54. grouped: dict[str, list[str | None]] = {}
  55. for key, value in raw_pairs:
  56. if key not in field_by_key:
  57. unknown_tokens.append(key if value is None else f"{key}={value}")
  58. else:
  59. grouped.setdefault(key, []).append(value)
  60. # Build keyword arguments for the dataclass constructor.
  61. kwargs: dict[str, Any] = {}
  62. for key, field in field_by_key.items():
  63. hint = hints[field.name]
  64. entries = grouped.get(key, [])
  65. kwargs[field.name] = _parse_field(key, hint, entries, field)
  66. return cls(**kwargs), ",".join(unknown_tokens)
  67. def _parse_field(
  68. name: str, hint: Any, entries: list[str | None], field: dataclasses.Field[Any]
  69. ) -> Any:
  70. origin = typing.get_origin(hint)
  71. if origin is list:
  72. return _parse_list(name, hint, entries)
  73. if origin is dict:
  74. return _parse_dict(name, hint, entries)
  75. # Scalar field — at most one entry expected.
  76. if not entries:
  77. if field.default is not dataclasses.MISSING:
  78. return field.default
  79. if field.default_factory is not dataclasses.MISSING:
  80. return field.default_factory()
  81. msg = f"missing required option '{name}'"
  82. raise ValueError(msg)
  83. if len(entries) > 1:
  84. msg = (
  85. f"option '{name}' specified {len(entries)} times but expects a single value"
  86. )
  87. raise ValueError(msg)
  88. return _parse_scalar(name, hint, entries[0])
  89. def _parse_list(name: str, hint: Any, entries: list[str | None]) -> list[Any]:
  90. args = typing.get_args(hint)
  91. if not args:
  92. msg = f"list field '{name}' has no element type annotation"
  93. raise TypeError(msg)
  94. elem_type = args[0]
  95. _assert_primitive_element(name, elem_type)
  96. return [_parse_scalar(name, elem_type, raw_value) for raw_value in entries]
  97. def _parse_dict(name: str, hint: Any, entries: list[str | None]) -> dict[str, Any]:
  98. args = typing.get_args(hint)
  99. if len(args) != 2:
  100. msg = f"dict field '{name}' has incomplete type annotation"
  101. raise TypeError(msg)
  102. key_type, val_type = args[0], args[1]
  103. if key_type is not str:
  104. msg = f"dict field '{name}' key type must be str, not {key_type}"
  105. raise TypeError(msg)
  106. _assert_primitive_element(name, val_type)
  107. result: dict[str, Any] = {}
  108. for raw_value in entries:
  109. if raw_value is None:
  110. msg = f"option '{name}' requires a value in the form '{name}=key:value'"
  111. raise ValueError(msg)
  112. if ":" not in raw_value:
  113. msg = f"option '{name}' value '{raw_value}' must contain ':' to separate key and value"
  114. raise ValueError(msg)
  115. key, _, val = raw_value.partition(":")
  116. if key in result:
  117. msg = f"option '{name}' has duplicate key '{key}'"
  118. raise ValueError(msg)
  119. result[key] = _parse_scalar(name, val_type, val)
  120. return result
  121. def _parse_scalar(name: str, hint: Any, raw_value: str | None) -> Any:
  122. origin = typing.get_origin(hint)
  123. # Unwrap `X | None` to its inner type. Combined with a
  124. # `None` default this yields a tri-state option: unset, true, or false.
  125. if origin is typing.Union or origin is types.UnionType:
  126. inner = [arg for arg in typing.get_args(hint) if arg is not type(None)]
  127. if len(inner) == 1:
  128. return _parse_scalar(name, inner[0], raw_value)
  129. msg = f"option '{name}': unsupported union type {hint}"
  130. raise TypeError(msg)
  131. if hint is bool:
  132. if raw_value is None:
  133. # Bare key — treated as True.
  134. return True
  135. lower = raw_value.lower()
  136. if lower == "true":
  137. return True
  138. if lower == "false":
  139. return False
  140. msg = f"option '{name}': cannot parse '{raw_value}' as bool; use 'true' or 'false'"
  141. raise ValueError(msg)
  142. # All other types require an explicit value; a bare key is an error.
  143. if raw_value is None:
  144. msg = f"option '{name}' requires a value (bare key only valid for bool fields)"
  145. raise ValueError(msg)
  146. if hint is str:
  147. return raw_value
  148. if hint is int:
  149. try:
  150. return int(raw_value)
  151. except ValueError:
  152. msg = f"option '{name}': cannot parse '{raw_value}' as int"
  153. raise ValueError(msg) from None
  154. if hint is float:
  155. try:
  156. return float(raw_value)
  157. except ValueError:
  158. msg = f"option '{name}': cannot parse '{raw_value}' as float"
  159. raise ValueError(msg) from None
  160. if origin is typing.Literal:
  161. allowed = typing.get_args(hint)
  162. if raw_value not in allowed:
  163. msg = f"option '{name}': '{raw_value}' is not a valid value; allowed: {', '.join(repr(a) for a in allowed)}"
  164. raise ValueError(msg)
  165. return raw_value
  166. if isinstance(hint, type) and issubclass(hint, str) and issubclass(hint, enum.Enum):
  167. lower = raw_value.lower()
  168. for member in hint:
  169. if member.name.lower() == lower or str(member.value).lower() == lower:
  170. return member
  171. allowed = ", ".join(m.name for m in hint)
  172. msg = f"option '{name}': '{raw_value}' is not a valid {hint.__name__}; allowed: {allowed}"
  173. raise ValueError(msg)
  174. if isinstance(hint, type) and issubclass(hint, enum.IntEnum):
  175. # Try by name first (case-insensitive).
  176. lower = raw_value.lower()
  177. for member in hint:
  178. if member.name.lower() == lower:
  179. return member
  180. # Try by integer value.
  181. try:
  182. int_val = int(raw_value)
  183. return hint(int_val)
  184. except (ValueError, KeyError):
  185. pass
  186. allowed = ", ".join(m.name for m in hint)
  187. msg = f"option '{name}': '{raw_value}' is not a valid {hint.__name__}; allowed names: {allowed}"
  188. raise ValueError(msg)
  189. msg = f"option '{name}': unsupported field type {hint}"
  190. raise TypeError(msg)
  191. def _assert_primitive_element(name: str, elem_type: Any) -> None:
  192. if elem_type not in (bool, str, int, float):
  193. msg = f"field '{name}' element type must be one of bool, str, int, or float; got {elem_type}"
  194. raise TypeError(msg)