Coverage for yuio / util.py: 94%

1# Yuio project, MIT license. 

2# 

3# https://github.com/taminomara/yuio/ 

4# 

5# You're free to copy this file to your project and edit it for your needs, 

6# just keep this copyright line please :3 

7 

8""" 

9.. autofunction:: to_dash_case 

10 

11.. autofunction:: dedent 

12 

13.. autofunction:: find_docs 

14 

15.. autofunction:: commonprefix 

16 

17.. autofunction:: merge_dicts 

18 

19.. autofunction:: merge_dicts_opt 

20 

21.. autoclass:: UserString 

22 

23 .. automethod:: _wrap 

24 

25.. autoclass:: ClosedIO 

26 

27""" 

28 

29from __future__ import annotations 

30 

31import io as _io 

32import re as _re 

33import textwrap as _textwrap 

34import weakref 

35 

36from typing import TYPE_CHECKING 

37 

38if TYPE_CHECKING: 

39 import typing_extensions as _t 

40else: 

41 from yuio import _typing as _t 

42 

43__all__ = [ 

44 "ClosedIO", 

45 "UserString", 

46 "commonprefix", 

47 "dedent", 

48 "find_docs", 

49 "merge_dicts", 

50 "merge_dicts_opt", 

51 "to_dash_case", 

52] 

53 

54_UNPRINTABLE = "".join([chr(i) for i in range(32)]) + "\x7f" 

55_UNPRINTABLE_TRANS = str.maketrans(_UNPRINTABLE, " " * len(_UNPRINTABLE)) 

56_UNPRINTABLE_RE = r"[" + _re.escape(_UNPRINTABLE) + "]" 

57_UNPRINTABLE_RE_WITHOUT_NL = r"[" + _re.escape(_UNPRINTABLE.replace("\n", "")) + "]" 

58 

59_TO_DASH_CASE_RE = _re.compile( 

60 r""" 

61 # We will add a dash (bear with me here): 

62 [_\s] # 1. instead of underscore or space, 

63 | ( # 2. OR in the following case: 

64 (?<!^) # - not at the beginning of the string, 

65 ( # - AND EITHER: 

66 (?<=[A-Z])(?=[A-Z][a-z]) # - before case gets lower (`XMLTag` -> `XML-Tag`), 

67 | (?<=[a-zA-Z])(?![a-zA-Z_]) # - between a letter and a non-letter (`HTTP20` -> `HTTP-20`), 

68 | (?<![A-Z_])(?=[A-Z]) # - between non-uppercase and uppercase letter (`TagXML` -> `Tag-XML`), 

69 ) # - AND ALSO: 

70 (?!$) # - not at the end of the string. 

71 ) 

72 """, 

73 _re.VERBOSE | _re.MULTILINE, 

74) 

75 

76 

77def to_dash_case(msg: str, /) -> str: 

78 """ 

79 Convert ``CamelCase`` or ``snake_case`` identifier to a ``dash-case`` one. 

80 

81 This function assumes ASCII input, and will not work correctly 

82 with non-ASCII characters. 

83 

84 :param msg: 

85 identifier to convert. 

86 :returns: 

87 identifier in ``dash-case``. 

88 :example: 

89 :: 

90 

91 >>> to_dash_case("SomeClass") 

92 'some-class' 

93 >>> to_dash_case("HTTP20XMLUberParser") 

94 'http-20-xml-uber-parser' 

95 

96 """ 

97 

98 return _TO_DASH_CASE_RE.sub("-", msg).lower() 

99 

100 

101def dedent(msg: str, /): 

102 """ 

103 Remove leading indentation from a message and normalize trailing newlines. 

104 

105 This function is intended to be used with triple-quote string literals, 

106 such as docstrings. It will remove common indentation from second 

107 and subsequent lines, then it will strip any leading and trailing whitespaces 

108 and add a new line at the end. 

109 

110 :param msg: 

111 message to dedent. 

112 :returns: 

113 normalized message. 

114 :example: 

115 :: 

116 

117 >>> def foo(): 

118 ... \"""Documentation for function ``foo``. 

119 ... 

120 ... Leading indent is stripped. 

121 ... \""" 

122 ... 

123 ... ... 

124 

125 >>> dedent(foo.__doc__) 

126 'Documentation for function ``foo``.\\n\\nLeading indent is stripped.\\n' 

127 

128 """ 

129 

130 if not msg: 

131 return msg 

132 

133 first, *rest = msg.splitlines(keepends=True) 

134 return (first.rstrip() + "\n" + _textwrap.dedent("".join(rest))).strip() + "\n" 

135 

136 

137_COMMENT_RE = _re.compile(r"^\s*#:(.*)\r?\n?$") 

138_DOCS_CACHE: weakref.WeakKeyDictionary[_t.Any, dict[str, str]] = ( 

139 weakref.WeakKeyDictionary() 

140) 

141 

142 

143def find_docs(obj: _t.Any, /) -> dict[str, str]: 

144 """ 

145 Find documentation for fields of a class. 

146 

147 Inspects source code of a class and finds docstrings and doc comments (``#:``) 

148 for variables in its body. Doesn't inspect ``__init__``, doesn't return documentation 

149 for class methods. 

150 

151 """ 

152 

153 # Based on code from Sphinx, two clause BSD license. 

154 # See https://github.com/sphinx-doc/sphinx/blob/master/LICENSE.rst. 

155 

156 try: 

157 return _DOCS_CACHE[obj] 

158 except KeyError: 

159 pass 

160 except TypeError: 

161 return {} 

162 

163 import ast 

164 import inspect 

165 import itertools 

166 

167 if (qualname := getattr(obj, "__qualname__", None)) is None: 

168 # Not a known object. 

169 _DOCS_CACHE[obj] = {} 

170 return {} 

171 

172 if "<locals>" in qualname: 

173 # This will not work as expected! 

174 _DOCS_CACHE[obj] = {} 

175 return {} 

176 

177 try: 

178 sourcelines, _ = inspect.getsourcelines(obj) 

179 except TypeError: 

180 _DOCS_CACHE[obj] = {} 

181 return {} 

182 

183 docs: dict[str, str] = {} 

184 

185 node = ast.parse(_textwrap.dedent("".join(sourcelines))) 

186 assert isinstance(node, ast.Module) 

187 assert len(node.body) == 1 

188 cdef = node.body[0] 

189 

190 if isinstance(cdef, ast.ClassDef): 

191 fields: list[tuple[int, str]] = [] 

192 last_field: str | None = None 

193 for stmt in cdef.body: 

194 if ( 

195 last_field 

196 and isinstance(stmt, ast.Expr) 

197 and isinstance(stmt.value, ast.Constant) 

198 and isinstance(stmt.value.value, str) 

199 ): 

200 docs[last_field] = dedent(stmt.value.value).removesuffix("\n") 

201 last_field = None 

202 if isinstance(stmt, ast.AnnAssign): 

203 target = stmt.target 

204 elif isinstance(stmt, ast.Assign) and len(stmt.targets) == 1: 

205 target = stmt.targets[0] 

206 else: 

207 continue 

208 if isinstance(target, ast.Name) and not target.id.startswith("_"): 

209 fields.append((stmt.lineno, target.id)) 

210 last_field = target.id 

211 elif isinstance(cdef, ast.FunctionDef): 

212 fields = [ 

213 (field.lineno, field.arg) 

214 for field in itertools.chain( 

215 cdef.args.posonlyargs, 

216 cdef.args.args, 

217 [cdef.args.vararg] if cdef.args.vararg else [], 

218 cdef.args.kwonlyargs, 

219 ) 

220 ] 

221 else: # pragma: no cover 

222 _DOCS_CACHE[obj] = {} 

223 return {} 

224 

225 for pos, name in fields: 

226 if name in docs: 

227 continue 

228 

229 comment_lines: list[str] = [] 

230 for before_line in sourcelines[pos - 2 :: -1]: 

231 if match := _COMMENT_RE.match(before_line): 

232 comment_lines.append(match.group(1)) 

233 else: 

234 break 

235 

236 if comment_lines: 

237 docs[name] = dedent("\n".join(reversed(comment_lines))).removesuffix("\n") 

238 

239 _DOCS_CACHE[obj] = docs 

240 return docs 

241 

242 

243def commonprefix(m: _t.Collection[str]) -> str: 

244 if not m: 

245 return "" 

246 s1 = min(m) 

247 s2 = max(m) 

248 for i, c in enumerate(s1): 

249 if c != s2[i]: 

250 return s1[:i] 

251 return s1 

252 

253 

254K = _t.TypeVar("K") 

255V = _t.TypeVar("V") 

256 

257 

258def merge_dicts( 

259 merge_values: _t.Callable[[V, V], V], / 

260) -> _t.Callable[[dict[K, V], dict[K, V]], dict[K, V]]: 

261 """ 

262 Create a function that merges two :class:`dict`\\ s, using the given 

263 callable to merge values that appear in both dicts. 

264 

265 :param merge_values: 

266 function that will be used to merge values that appear in both dicts. 

267 :returns: 

268 function that merges dicts. 

269 :example: 

270 This is useful for merging dicts of configs, other dicts, and so on: 

271 

272 .. code-block:: python 

273 

274 merge_dicts_of_configs = merge_dicts(lambda l, r: l | r) 

275 

276 """ 

277 

278 def merge(lhs: dict[K, V], rhs: dict[K, V]) -> dict[K, V]: 

279 # Handle `None` in case we're joining `dict[K, V] | None`. 

280 if lhs is None and rhs is None: # pyright: ignore[reportUnnecessaryComparison] 

281 return None 

282 elif lhs is None: # pyright: ignore[reportUnnecessaryComparison] 

283 return rhs.copy() 

284 elif rhs is None: # pyright: ignore[reportUnnecessaryComparison] 

285 return lhs.copy() 

286 

287 res = lhs.copy() 

288 for k, v in rhs.items(): 

289 if k in res: 

290 res[k] = merge_values(res[k], v) 

291 else: 

292 res[k] = v 

293 return res 

294 

295 return merge 

296 

297 

298def merge_dicts_opt( 

299 merge_values: _t.Callable[[V, V], V], / 

300) -> _t.Callable[[dict[K, V] | None, dict[K, V] | None], dict[K, V] | None]: 

301 """ 

302 Like :func:`merge_dicts`, but also handles :data:`None`\\ s. 

303 

304 """ 

305 

306 # `merge_dicts` actually handles `None`, we need `merge_dicts_opt` 

307 # only for correct type hints. 

308 return merge_dicts(merge_values) # pyright: ignore[reportReturnType] 

309 

310 

311if TYPE_CHECKING: 

312 

313 class _FormatMapMapping(_t.Protocol): 

314 def __getitem__(self, key: str, /) -> _t.Any: ... 

315 

316 class _TranslateTable(_t.Protocol): 

317 def __getitem__(self, key: int, /) -> str | int | None: ... 

318 

319 

320class UserString(str): 

321 """ 

322 Base class for user string. 

323 

324 This class is similar to :class:`collections.UserString`, but actually derived 

325 from string, with customizable wrapping semantics, and returns custom string 

326 instances from all string methods (:class:`collections.UserString` doesn't 

327 wrap strings returned from :meth:`str.split` and similar). 

328 

329 .. tip:: 

330 

331 When deriving from this class, add ``__slots__`` to avoid making a string 

332 with a ``__dict__`` property. 

333 

334 """ 

335 

336 __slots__ = () 

337 

338 def _wrap(self, data: str) -> _t.Self: 

339 """ 

340 Wrap raw string that resulted from an operation on this instance into another 

341 instance of :class:`UserString`. 

342 

343 Override this method if you need to preserve some internal state during 

344 operations. 

345 

346 By default, this simply creates an instance of ``self.__class__`` with the 

347 given string. 

348 

349 """ 

350 

351 return self.__class__(data) 

352 

353 def __add__(self, value: str, /) -> _t.Self: 

354 return self._wrap(super().__add__(value)) 

355 

356 def __format__(self, format_spec: str, /) -> _t.Self: 

357 return self._wrap(super().__format__(format_spec)) 

358 

359 def __getitem__(self, key: _t.SupportsIndex | slice, /) -> _t.Self: 

360 return self._wrap(super().__getitem__(key)) 

361 

362 def __mod__(self, value: _t.Any, /) -> _t.Self: 

363 return self._wrap(super().__mod__(value)) 

364 

365 def __mul__(self, value: _t.SupportsIndex, /) -> _t.Self: 

366 return self._wrap(super().__mul__(value)) 

367 

368 def __rmul__(self, value: _t.SupportsIndex, /) -> _t.Self: 

369 return self._wrap(super().__rmul__(value)) 

370 

371 def capitalize(self) -> _t.Self: 

372 return self._wrap(super().capitalize()) 

373 

374 def casefold(self) -> _t.Self: 

375 return self._wrap(super().casefold()) 

376 

377 def center(self, width: _t.SupportsIndex, fillchar: str = " ", /) -> _t.Self: 

378 return self._wrap(super().center(width)) 

379 

380 def expandtabs(self, tabsize: _t.SupportsIndex = 8) -> _t.Self: 

381 return self._wrap(super().expandtabs(tabsize)) 

382 

383 def format_map(self, mapping: _FormatMapMapping, /) -> _t.Self: 

384 return self._wrap(super().format_map(mapping)) 

385 

386 def format(self, *args: object, **kwargs: object) -> _t.Self: 

387 return self._wrap(super().format(*args, **kwargs)) 

388 

389 def join(self, iterable: _t.Iterable[str], /) -> _t.Self: 

390 return self._wrap(super().join(iterable)) 

391 

392 def ljust(self, width: _t.SupportsIndex, fillchar: str = " ", /) -> _t.Self: 

393 return self._wrap(super().ljust(width, fillchar)) 

394 

395 def lower(self) -> _t.Self: 

396 return self._wrap(super().lower()) 

397 

398 def lstrip(self, chars: str | None = None, /) -> _t.Self: 

399 return self._wrap(super().lstrip(chars)) 

400 

401 def partition(self, sep: str, /) -> tuple[_t.Self, _t.Self, _t.Self]: 

402 l, c, r = super().partition(sep) 

403 return self._wrap(l), self._wrap(c), self._wrap(r) 

404 

405 def removeprefix(self, prefix: str, /) -> _t.Self: 

406 return self._wrap(super().removeprefix(prefix)) 

407 

408 def removesuffix(self, suffix: str, /) -> _t.Self: 

409 return self._wrap(super().removesuffix(suffix)) 

410 

411 def replace(self, old: str, new: str, count: _t.SupportsIndex = -1, /) -> _t.Self: 

412 return self._wrap(super().replace(old, new, count)) 

413 

414 def rjust(self, width: _t.SupportsIndex, fillchar: str = " ", /) -> _t.Self: 

415 return self._wrap(super().rjust(width, fillchar)) 

416 

417 def rpartition(self, sep: str, /) -> tuple[_t.Self, _t.Self, _t.Self]: 

418 l, c, r = super().rpartition(sep) 

419 return self._wrap(l), self._wrap(c), self._wrap(r) 

420 

421 def rsplit( # pyright: ignore[reportIncompatibleMethodOverride] 

422 self, sep: str | None = None, maxsplit: _t.SupportsIndex = -1 

423 ) -> list[_t.Self]: 

424 return [self._wrap(part) for part in super().rsplit(sep, maxsplit)] 

425 

426 def rstrip(self, chars: str | None = None, /) -> _t.Self: 

427 return self._wrap(super().rstrip(chars)) 

428 

429 def split( # pyright: ignore[reportIncompatibleMethodOverride] 

430 self, sep: str | None = None, maxsplit: _t.SupportsIndex = -1 

431 ) -> list[_t.Self]: 

432 return [self._wrap(part) for part in super().split(sep, maxsplit)] 

433 

434 def splitlines( # pyright: ignore[reportIncompatibleMethodOverride] 

435 self, keepends: bool = False 

436 ) -> list[_t.Self]: 

437 return [self._wrap(part) for part in super().splitlines(keepends)] 

438 

439 def strip(self, chars: str | None = None, /) -> _t.Self: 

440 return self._wrap(super().strip(chars)) 

441 

442 def swapcase(self) -> _t.Self: 

443 return self._wrap(super().swapcase()) 

444 

445 def title(self) -> _t.Self: 

446 return self._wrap(super().title()) 

447 

448 def translate(self, table: _TranslateTable, /) -> _t.Self: 

449 return self._wrap(super().translate(table)) 

450 

451 def upper(self) -> _t.Self: 

452 return self._wrap(super().upper()) 

453 

454 def zfill(self, width: _t.SupportsIndex, /) -> _t.Self: 

455 return self._wrap(super().zfill(width)) 

456 

457 

458class ClosedIO(_io.TextIOBase, _t.TextIO): # type: ignore 

459 """ 

460 Dummy stream that's always closed. 

461 

462 """ 

463 

464 def __init__(self) -> None: 

465 super().__init__() 

466 self.close()