Coverage for .tox/coverage/lib/python3.13/site-packages/wuttjamaican/util.py: 100%

1# -*- coding: utf-8; -*-

2################################################################################

4# WuttJamaican -- Base package for Wutta Framework

7# This file is part of Wutta Framework.

9# Wutta Framework is free software: you can redistribute it and/or modify it

10# under the terms of the GNU General Public License as published by the Free

11# Software Foundation, either version 3 of the License, or (at your option) any

12# later version.

13#

14# Wutta Framework is distributed in the hope that it will be useful, but

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for

17# more details.

18#

19# You should have received a copy of the GNU General Public License along with

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.

21#

22################################################################################

23"""

24WuttJamaican - utilities

25"""

27import datetime

28import importlib

29import logging

30import os

31import shlex

32import warnings

34from uuid_extensions import uuid7

36log = logging.getLogger(__name__)

39# nb. this is used as default kwarg value in some places, to

40# distinguish passing a ``None`` value, vs. *no* value at all

41UNSPECIFIED = object()

44def get_class_hierarchy(klass, topfirst=True):

45 """

46 Returns a list of all classes in the inheritance chain for the

47 given class.

49 For instance::

51 class A:

52 pass

54 class B(A):

55 pass

57 class C(B):

58 pass

60 get_class_hierarchy(C)

61 # -> [A, B, C]

63 :param klass: The reference class. The list of classes returned

64 will include this class and all its parents.

66 :param topfirst: Whether the returned list should be sorted in a

67 "top first" way, e.g. A) grandparent, B) parent, C) child.

68 This is the default but pass ``False`` to get the reverse.

69 """

70 hierarchy = []

72 def traverse(cls):

73 if cls is not object:

74 hierarchy.append(cls)

75 for parent in cls.__bases__:

76 traverse(parent)

78 traverse(klass)

79 if topfirst:

80 hierarchy.reverse()

81 return hierarchy

84def get_value(obj, key):

85 """

86 Convenience function to retrive a value by name from the given

87 object. This will first try to assume the object is a dict but

88 will fallback to using ``getattr()`` on it.

90 :param obj: Arbitrary dict or object of any kind which would have

91 named attributes.

93 :param key: Key/name of the field to get.

95 :returns: Whatever value is found. Or maybe an ``AttributeError``

96 is raised if the object does not have the key/attr set.

97 """

98 # nb. we try dict access first, since wutta data model objects

99 # should all support that anyway, so it's 2 birds 1 stone.

100 try:

101 return obj[key]

102

103 except (KeyError, TypeError):

104 # nb. key error means the object supports key lookup (i.e. is

105 # dict-like) but did not have that key set. which is actually

106 # an expected scenario for association proxy fields, but for

107 # those a getattr() should still work; see also

108 # wuttjamaican.db.util.ModelBase

109 return getattr(obj, key)

110

111

112def load_entry_points(group, lists=False, ignore_errors=False):

113 """

114 Load a set of ``setuptools``-style :term:`entry points <entry

115 point>`.

116

117 This is used to locate "plugins" and similar things, e.g. discover

118 which batch handlers are installed.

119

120 Logic will inspect the registered entry points and return a dict

121 whose keys are the entry point names. By default the dict values

122 will be the loaded objects as referenced by each entry point.

123

124 In some cases (notably, import handlers for wuttasync) the keys

125 may not always be unique. This allows multiple projects to define

126 entry points for the same key. If you specify ``lists=True`` then

127 the dict values will each be *lists* of loaded objects instead.

128 (Otherwise some entry points would be discarded when duplicate

129 keys are found.)

130

131 :param group: The group (string name) of entry points to be

132 loaded, e.g. ``'wutta.commands'``.

133

134 :param lists: Whether to return lists instead of single object

135 values.

136

137 :param ignore_errors: If false (the default), any errors will be

138 raised normally. If true, errors will be logged but not

139 raised.

140

141 :returns: A dict of entry points, as described above.

142 """

143 entry_points = {}

144

145 try:

146 # nb. this package was added in python 3.8

147 import importlib.metadata as importlib_metadata # pylint: disable=import-outside-toplevel

148 except ImportError:

149 import importlib_metadata # pylint: disable=import-outside-toplevel

150

151 eps = importlib_metadata.entry_points()

152 if not hasattr(eps, "select"):

153 # python < 3.10

154 eps = eps.get(group, [])

155 else:

156 # python >= 3.10

157 eps = eps.select(group=group)

158 for entry_point in eps:

159 try:

160 ep = entry_point.load()

161 except Exception: # pylint: disable=broad-exception-caught

162 if not ignore_errors:

163 raise

164 log.warning("failed to load entry point: %s", entry_point, exc_info=True)

165 else:

166 if lists:

167 entry_points.setdefault(entry_point.name, []).append(ep)

168 else:

169 if entry_point.name in entry_points:

170 # TODO: not sure why the "same" entry point can be

171 # discovered multiple times, but in practice i did

172 # see this. however it was on a python 3.8 system

173 # so i'm guessing that has something to do with

174 # it. we'll avoid the warning if that's the case.

175 if entry_points[entry_point.name] is not ep:

176 log.warning(

177 "overwriting existing key '%s' with entry point: %s",

178 entry_point.name,

179 ep,

180 )

181 entry_points[entry_point.name] = ep

182

183 return entry_points

184

185

186def load_object(spec):

187 """

188 Load an arbitrary object from a module, according to the spec.

189

190 The spec string should contain a dotted path to an importable module,

191 followed by a colon (``':'``), followed by the name of the object to be

192 loaded. For example:

193

194 .. code-block:: none

195

196 wuttjamaican.util:parse_bool

197

198 You'll notice from this example that "object" in this context refers to any

199 valid Python object, i.e. not necessarily a class instance. The name may

200 refer to a class, function, variable etc. Once the module is imported, the

201 ``getattr()`` function is used to obtain a reference to the named object;

202 therefore anything supported by that approach should work.

203

204 :param spec: Spec string.

205

206 :returns: The specified object.

207 """

208 if not spec:

209 raise ValueError("no object spec provided")

210

211 module_path, name = spec.split(":")

212 module = importlib.import_module(module_path)

213 return getattr(module, name)

214

215

216def make_title(text):

217 """

218 Return a human-friendly "title" for the given text.

219

220 This is mostly useful for converting a Python variable name (or

221 similar) to a human-friendly string, e.g.::

222

223 make_title('foo_bar') # => 'Foo Bar'

224 """

225 text = text.replace("_", " ")

226 text = text.replace("-", " ")

227 words = text.split()

228 return " ".join([x.capitalize() for x in words])

229

230

231def make_full_name(*parts):

232 """

233 Make a "full name" from the given parts.

234

235 :param \\*parts: Distinct name values which should be joined

236 together to make the full name.

237

238 :returns: The full name.

239

240 For instance::

241

242 make_full_name('First', '', 'Last', 'Suffix')

243 # => "First Last Suffix"

244 """

245 parts = [(part or "").strip() for part in parts]

246 parts = [part for part in parts if part]

247 return " ".join(parts)

248

249

250def get_timezone_by_name(tzname):

251 """

252 Retrieve a timezone object by name.

253

254 This is mostly a compatibility wrapper, since older Python is

255 missing the :mod:`python:zoneinfo` module.

256

257 For Python 3.9 and newer, this instantiates

258 :class:`python:zoneinfo.ZoneInfo`.

259

260 For Python 3.8, this calls :func:`dateutil:dateutil.tz.gettz()`.

261

262 See also :meth:`~wuttjamaican.app.AppHandler.get_timezone()` on

263 the app handler.

264

265 :param tzname: String name for timezone.

266

267 :returns: :class:`python:datetime.tzinfo` instance

268 """

269 try:

270 from zoneinfo import ZoneInfo # pylint: disable=import-outside-toplevel

271

272 return ZoneInfo(tzname)

273

274 except ImportError: # python 3.8

275 from dateutil.tz import gettz # pylint: disable=import-outside-toplevel

276

277 return gettz(tzname)

278

279

280def localtime(dt=None, from_utc=True, want_tzinfo=True, local_zone=None):

281 """

282 This produces a datetime in the "local" timezone. By default it

283 will be *zone-aware*.

284

285 See also the shortcut

286 :meth:`~wuttjamaican.app.AppHandler.localtime()` method on the app

287 handler. For usage examples see :ref:`convert-to-localtime`.

288

289 See also :func:`make_utc()` which is sort of the inverse.

290

291 :param dt: Optional :class:`python:datetime.datetime` instance.

292 If not specified, the current time will be used.

293

294 :param from_utc: Boolean indicating whether a naive ``dt`` is

295 already (effectively) in UTC timezone. Set this to false when

296 providing a naive ``dt`` which is already in "local" timezone

297 instead of UTC. This flag is ignored if ``dt`` is zone-aware.

298

299 :param want_tzinfo: Boolean indicating whether the resulting

300 datetime should have its

301 :attr:`~python:datetime.datetime.tzinfo` attribute set. Set

302 this to false if you want a naive value; it's true by default,

303 for zone-aware.

304

305 :param local_zone: Optional :class:`python:datetime.tzinfo`

306 instance to use as "local" timezone, instead of relying on

307 Python to determine the system local timezone.

308

309 :returns: :class:`python:datetime.datetime` instance in local

310 timezone.

311 """

312 # use current time if none provided

313 if dt is None:

314 dt = datetime.datetime.now(datetime.timezone.utc)

315

316 # set dt's timezone if needed

317 if not dt.tzinfo:

318 # UTC is default assumption unless caller says otherwise

319 if from_utc:

320 dt = dt.replace(tzinfo=datetime.timezone.utc)

321 elif local_zone:

322 dt = dt.replace(tzinfo=local_zone)

323 else: # default system local timezone

324 tz = dt.astimezone().tzinfo

325 dt = dt.replace(tzinfo=tz)

326

327 # convert to local timezone

328 if local_zone:

329 dt = dt.astimezone(local_zone)

330 else:

331 dt = dt.astimezone()

332

333 # maybe strip tzinfo

334 if want_tzinfo:

335 return dt

336 return dt.replace(tzinfo=None)

337

338

339def make_utc(dt=None, tzinfo=False):

340 """

341 This returns a datetime local to the UTC timezone. By default it

342 will be a *naive* datetime; the common use case is to convert as

343 needed for sake of writing to the database.

344

345 See also the shortcut

346 :meth:`~wuttjamaican.app.AppHandler.make_utc()` method on the app

347 handler. For usage examples see :ref:`convert-to-utc`.

348

349 See also :func:`localtime()` which is sort of the inverse.

350

351 :param dt: Optional :class:`python:datetime.datetime` instance.

352 If not specified, the current time will be used.

353

354 :param tzinfo: Boolean indicating whether the return value should

355 have its :attr:`~python:datetime.datetime.tzinfo` attribute

356 set. This is false by default in which case the return value

357 will be naive.

358

359 :returns: :class:`python:datetime.datetime` instance local to UTC.

360 """

361 # use current time if none provided

362 if dt is None:

363 now = datetime.datetime.now(datetime.timezone.utc)

364 if tzinfo:

365 return now

366 return now.replace(tzinfo=None)

367

368 # otherwise may need to convert timezone

369 if dt.tzinfo:

370 if dt.tzinfo is not datetime.timezone.utc:

371 dt = dt.astimezone(datetime.timezone.utc)

372 if tzinfo:

373 return dt

374 return dt.replace(tzinfo=None)

375

376 # naive value returned as-is..

377 if not tzinfo:

378 return dt

379

380 # ..unless tzinfo is wanted, in which case this assumes naive

381 # value is in the UTC timezone

382 return dt.replace(tzinfo=datetime.timezone.utc)

383

384

385# TODO: deprecate / remove this eventually

386def make_true_uuid():

387 """

388 Generate a new v7 UUID.

389

390 See also :func:`make_uuid()`.

391

392 :returns: :class:`python:uuid.UUID` instance

393 """

394 return uuid7()

395

396

397# TODO: deprecate / remove this eventually

398def make_str_uuid():

399 """

400 Generate a new v7 UUID value as string.

401

402 See also :func:`make_uuid()`.

403

404 :returns: UUID as 32-character hex string

405 """

406 return make_true_uuid().hex

407

408

409# TODO: eventually refactor, to return true uuid

410def make_uuid():

411 """

412 Generate a new v7 UUID value.

413

414 See also the app handler shortcut,

415 :meth:`~wuttjamaican.app.AppHandler.make_uuid()`.

416

417 :returns: UUID as 32-character hex string

418

419 .. warning::

420

421 **TEMPORARY BEHAVIOR**

422

423 For the moment, use of this function is discouraged. Instead you

424 should use :func:`make_true_uuid()` or :func:`make_str_uuid()` to

425 be explicit about the return type you expect.

426

427 *Eventually* (once it's clear most/all callers are using the

428 explicit functions) this will be refactored to return a UUID

429 instance. But for now this function returns a string.

430 """

431 warnings.warn(

432 "util.make_uuid() is temporarily deprecated, in favor of "

433 "explicit functions, util.make_true_uuid() and util.make_str_uuid()",

434 DeprecationWarning,

435 stacklevel=2,

436 )

437 return make_str_uuid()

438

439

440def parse_bool(value):

441 """

442 Derive a boolean from the given string value.

443 """

444 if value is None:

445 return None

446 if isinstance(value, bool):

447 return value

448 if str(value).lower() in ("true", "yes", "y", "on", "1"):

449 return True

450 return False

451

452

453def parse_list(value):

454 """

455 Parse a configuration value, splitting by whitespace and/or commas

456 and taking quoting into account etc., yielding a list of strings.

457 """

458 if value is None:

459 return []

460 if isinstance(value, list):

461 return value

462 parser = shlex.shlex(value)

463 parser.whitespace += ","

464 parser.whitespace_split = True

465 values = list(parser)

466 for i, val in enumerate(values):

467 if val.startswith('"') and val.endswith('"'):

468 values[i] = val[1:-1]

469 elif val.startswith("'") and val.endswith("'"):

470 values[i] = val[1:-1]

471 return values

472

473

474def progress_loop(func, items, factory, message=None):

475 """

476 Convenience function to iterate over a set of items, invoking

477 logic for each, and updating a progress indicator along the way.

478

479 This function may also be called via the :term:`app handler`; see

480 :meth:`~wuttjamaican.app.AppHandler.progress_loop()`.

481

482 The ``factory`` will be called to create the progress indicator,

483 which should be an instance of

484 :class:`~wuttjamaican.progress.ProgressBase`.

485

486 The ``factory`` may also be ``None`` in which case there is no

487 progress, and this is really just a simple "for loop".

488

489 :param func: Callable to be invoked for each item in the sequence.

490 See below for more details.

491

492 :param items: Sequence of items over which to iterate.

493

494 :param factory: Callable which creates/returns a progress

495 indicator, or can be ``None`` for no progress.

496

497 :param message: Message to display along with the progress

498 indicator. If no message is specified, whether a default is

499 shown will be up to the progress indicator.

500

501 The ``func`` param should be a callable which accepts 2 positional

502 args ``(obj, i)`` - meaning for which is as follows:

503

504 :param obj: This will be an item within the sequence.

505

506 :param i: This will be the *one-based* sequence number for the

507 item.

508

509 See also :class:`~wuttjamaican.progress.ConsoleProgress` for a

510 usage example.

511 """

512 progress = None

513 if factory:

514 count = len(items)

515 progress = factory(message, count)

516

517 for i, item in enumerate(items, 1):

518 func(item, i)

519 if progress:

520 progress.update(i)

521

522 if progress:

523 progress.finish()

524

525

526def resource_path(path):

527 """

528 Returns the absolute file path for the given resource path.

529

530 A "resource path" is one which designates a python package name,

531 plus some path under that. For instance:

532

533 .. code-block:: none

534

535 wuttjamaican.email:templates

536

537 Assuming such a path should exist, the question is "where?"

538

539 So this function uses :mod:`python:importlib.resources` to locate

540 the path, possibly extracting the file(s) from a zipped package,

541 and returning the final path on disk.

542

543 It only does this if it detects it is needed, based on the given

544 ``path`` argument. If that is already an absolute path then it

545 will be returned as-is.

546

547 :param path: Either a package resource specifier as shown above,

548 or regular file path.

549

550 :returns: Absolute file path to the resource.

551 """

552 if not os.path.isabs(path) and ":" in path:

553 try:

554 # nb. these were added in python 3.9

555 from importlib.resources import ( # pylint: disable=import-outside-toplevel

556 files,

557 as_file,

558 )

559 except ImportError: # python < 3.9

560 from importlib_resources import ( # pylint: disable=import-outside-toplevel

561 files,

562 as_file,

563 )

564

565 package, filename = path.split(":")

566 ref = files(package) / filename

567 with as_file(ref) as p:

568 return str(p)

569

570 return path

571

572

573def simple_error(error):

574 """

575 Return a "simple" string for the given error. Result will look

576 like::

577

578 "ErrorClass: Description for the error"

579

580 However the logic checks to ensure the error has a descriptive

581 message first; if it doesn't the result will just be::

582

583 "ErrorClass"

584 """

585 cls = type(error).__name__

586 msg = str(error)

587 if msg:

588 return f"{cls}: {msg}"

589 return cls