Coverage for pyTooling/Warning/__init_

1# ==================================================================================================================== #

2# _____ _ _ __ __ _ #

3# _ __ _ |_ _|__ ___ | (_)_ __ __ \ \ / /_ _ _ __ _ __ (_)_ __ __ _ #

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` \ \ /\ / / _` | '__| '_ \| | '_ \ / _` | #

5# | |_) | |_| || | (_) | (_) | | | | | | (_| |\ V V / (_| | | | | | | | | | | (_| | #

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)_/\_/ \__,_|_| |_| |_|_|_| |_|\__, | #

7# |_| |___/ |___/ |___/ #

8# ==================================================================================================================== #

9# Authors: #

10# Patrick Lehmann #

11# #

12# License: #

13# ==================================================================================================================== #

15# #

16# Licensed under the Apache License, Version 2.0 (the "License"); #

17# you may not use this file except in compliance with the License. #

18# You may obtain a copy of the License at #

19# #

20# http://www.apache.org/licenses/LICENSE-2.0 #

21# #

22# Unless required by applicable law or agreed to in writing, software #

23# distributed under the License is distributed on an "AS IS" BASIS, #

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #

25# See the License for the specific language governing permissions and #

26# limitations under the License. #

27# #

28# SPDX-License-Identifier: Apache-2.0 #

29# ==================================================================================================================== #

30#

31"""

32A solution to send warnings like exceptions to a handler in the upper part of the call-stack.

34.. hint::

36 See :ref:`high-level help <WARNING>` for explanations and usage examples.

38.. seealso::

40 :mod:`pyTooling.Exceptions`

41 |rarr| Exceptions, which are raised instead of collected.

42 :mod:`pyTooling.TerminalUI`

43 |rarr| Writing the collected warnings to the terminal.

44"""

45from __future__ import annotations

47from threading import local, Lock

48from types import TracebackType

49from typing import Callable, Optional as Nullable, Iterator, Self, Iterable, Union

51from pyTooling.Decorators import export, readonly

52from pyTooling.Common import getFullyQualifiedName

53from pyTooling.Exceptions import ExceptionBase

56__all__ = ["_threadLocalData", "AnyWarning"]

59_threadLocalData = local()

60"""A reference to the thread local data needed by the pyTooling.Warning classes."""

63@export

64class CriticalWarning(BaseException):

65 """

66 Base-exception of all critical warnings handled by :class:`WarningCollector`.

68 .. tip::

70 Critical warnings must be unhandled within a call hierarchy, otherwise a :exc:`UnhandledCriticalWarningException`

71 will be raised.

72 """

74 @readonly

75 def HasNotes(self) -> bool:

76 """

77 Read-only property to return if the warning has attached notes.

79 :returns: True, if the warning has attached notes.

80 """

81 return hasattr(self, "__notes__") and self.__notes__ is not None and len(self.__notes__) > 0

83 @readonly

84 def Notes(self) -> tuple[str, ...]:

85 """

86 Read-only property to return warning's attached notes.

88 :returns: Attached notes.

89 """

90 return tuple(self.__notes__) if hasattr(self, "__notes__") else tuple()

93@export

94class Warning(BaseException):

95 """

96 Base-exception of all warnings handled by :class:`WarningCollector`.

98 .. tip::

100 Warnings can be unhandled within a call hierarchy.

101 """

102

103 @readonly

104 def HasNotes(self) -> bool:

105 """

106 Read-only property to return if the warning has attached notes.

107

108 :returns: True, if the warning has attached notes.

109 """

110 return hasattr(self, "__notes__") and self.__notes__ is not None and len(self.__notes__) > 0

111

112 @readonly

113 def Notes(self) -> tuple[str, ...]:

114 """

115 Read-only property to return warning's attached notes.

116

117 :returns: Attached notes.

118 """

119 return tuple(self.__notes__) if hasattr(self, "__notes__") else tuple()

120

121

122AnyWarning = Union[CriticalWarning, Warning]

123

124

125@export

126class UnhandledCriticalWarningException(ExceptionBase):

127 """

128 This exception is raised when a critical warning isn't handled by a :class:`WarningCollector` within the

129 call-hierarchy.

130 """

131

132

133@export

134class UnhandledExceptionException(ExceptionBase):

135 """

136 This exception is raised when an exception isn't handled by a :class:`WarningCollector` within the call-hierarchy.

137 """

138

139

140@export

141class EscalatedWarningException(ExceptionBase):

142 """

143 This exception is raised when a :class:`WarningCollector` decides a collected warning should not be collected but

144 raised.

145

146 A collector's handler returning ``True`` asks for the warning to be escalated; the warning itself becomes the

147 exception's cause.

148 """

149

150

151@export

152class WarningCollector:

153 """

154 A context manager to collect warnings within the call hierarchy.

155 """

156 _parent: Nullable[WarningCollector] #: Parent WarningCollector

157 _warnings: list[BaseException] #: List of collected warnings (and exceptions).

158 _handler: Nullable[Callable[[BaseException], bool]] #: Optional handler function, which is called per collected warning.

159

160 __slots__ = ("_parent", "_warnings", "_handler")

161

162 def __init__(

163 self,

164 warnings: Nullable[list[BaseException]] = None,

165 handler: Nullable[Callable[[BaseException], bool]] = None

166 ) -> None:

167 """

168 Initializes a warning collector.

169

170 :param warnings: Optional, reference to a list of warnings, which can be modified (appended) by this warning

171 collector. If ``None``, an internal list is created and can be referenced by the collector's

172 instance.

173 :param handler: Optional, handler function, which processes the current warning and decides if a warning should

174 be reraised as an exception.

175 :raises TypeError: If optional parameter 'warnings' is not of type list.

176 :raises TypeError: If optional parameter 'handler' is not a callable.

177 """

178 if warnings is None:

179 warnings = []

180 elif not isinstance(warnings, list): 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true

181 ex = TypeError(f"Parameter 'warnings' is not a list.")

182 ex.add_note(f"Got type '{getFullyQualifiedName(warnings)}'.")

183 raise ex

184

185 if handler is not None and not isinstance(handler, Callable): 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true

186 ex = TypeError(f"Parameter 'handler' is not callable.")

187 ex.add_note(f"Got type '{getFullyQualifiedName(handler)}'.")

188 raise ex

189

190 self._parent = None

191 self._warnings = warnings

192 self._handler = handler

193

194 def __len__(self) -> int:

195 """

196 Returns the number of collected warnings.

197

198 :returns: Number of collected warnings.

199 """

200 return len(self._warnings)

201

202 def __iter__(self) -> Iterator[Warning | CriticalWarning | Exception]:

203 """

204 Return an iterator over all collected warnings.

205

206 :returns: Iterator over the collected warnings.

207 """

208 return iter(self._warnings)

209

210 def __getitem__(self, index: int) -> Warning | CriticalWarning | Exception:

211 """

212 Access a collected warning by index.

213

214 :param index: Index of the warning.

215 :returns: Collected warning.

216 """

217 return self._warnings[index]

218

219 def __enter__(self) -> Self:

220 """

221 Enter the warning collector context.

222

223 :returns: The warning collector instance.

224 """

225 global _threadLocalData

226

227 try:

228 self._parent = _threadLocalData.warningCollector

229 except AttributeError:

230 pass

231

232 _threadLocalData.warningCollector = self

233

234 return self

235

236 def __exit__(

237 self,

238 exc_type: Nullable[type[BaseException]] = None,

239 exc_val: Nullable[BaseException] = None,

240 exc_tb: Nullable[TracebackType] = None

241 ) -> Nullable[bool]:

242 """

243 Exit the warning collector context.

244

245 :param exc_type: Exception type

246 :param exc_val: Exception instance

247 :param exc_tb: Exception's traceback.

248 :returns: ``None``

249 """

250 global _threadLocalData

251

252 _threadLocalData.warningCollector = self._parent

253

254 return False

255

256 @property

257 def Parent(self) -> Nullable[Self]:

258 """

259 Property to access the parent warning collector.

260

261 :returns: The parent warning collector or ``None``.

262 """

263 return self._parent

264

265 @Parent.setter

266 def Parent(self, value: Self) -> None:

267 self._parent = value

268

269 @readonly

270 def Warnings(self) -> list[Warning | CriticalWarning | Exception]:

271 """

272 Read-only property to access the list of collected warnings.

273

274 :returns: A list of collected warnings.

275 """

276 return self._warnings

277

278 def AddWarning(self, warning: Warning | CriticalWarning | Exception) -> bool:

279 """

280 Add a warning to the list of warnings managed by this warning collector.

281

282 :param warning: The warning to add to the collectors internal warning list.

283 :returns: Return ``True`` if the warning collector has a local handler callback and this handler returned

284 ``True``; otherwise ``False``.

285 :raises ValueError: If parameter ``warning`` is None.

286 :raises TypeError: If parameter ``warning`` is not of type :class:`Warning`.

287 """

288 if warning is None: 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true

289 raise ValueError("Parameter 'warning' is None.")

290 elif not isinstance(warning, (Warning, CriticalWarning, Exception)): 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true

291 ex = TypeError(f"Parameter 'warning' is not of type 'Warning', 'CriticalWarning' or 'Exception'.")

292 ex.add_note(f"Got type '{getFullyQualifiedName(warning)}'.")

293 raise ex

294

295 self._warnings.append(warning)

296

297 return False if self._handler is None else self._handler(warning)

298

299 @classmethod

300 def Raise(

301 cls,

302 warning: Warning | CriticalWarning | Exception,

303 cause: Nullable[Exception] = None,

304 *,

305 notes: Nullable[str | Iterable[str]] = None

306 ) -> None:

307 """

308 Walk the callstack frame by frame upwards and search for the first warning collector.

309

310 :param warning: Warning to send upwards in the call stack.

311 :param cause: Optional, root cause to be added to the warning.

312 :param notes: Optional, a single note or a list of notes to be added to the warning.

313 :raises EscalatedWarningException: If the warning collector asks for the warning to be raised.

314 :raises UnhandledExceptionException: If no warning collector was found along the call-hierarchy to collect and

315 handle an exception.

316 :raises UnhandledCriticalWarningException: If no warning collector was found along the call-hierarchy to collect

317 and handle a critical warning. |br|

318 Add a with-statement using :class:`WarningCollector` somewhere up the

319 call-hierarchy to receive and collect warnings.

320 """

321 global _threadLocalData

322

323 if cause is not None: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true

324 warning.__cause__ = cause

325

326 if notes is not None:

327 if isinstance(notes, str): 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true

328 warning.add_note(notes)

329 else:

330 for note in notes:

331 warning.add_note(note)

332

333 try:

334 warningCollector = _threadLocalData.warningCollector

335 if warningCollector.AddWarning(warning):

336 raise EscalatedWarningException(f"Warning: {warning}") from warning

337 except AttributeError:

338 ex = None

339 if isinstance(warning, Exception):

340 ex = UnhandledExceptionException(f"Unhandled Exception: {warning}")

341 elif isinstance(warning, CriticalWarning):

342 ex = UnhandledCriticalWarningException(f"Unhandled Critical Warning: {warning}")

343

344 if ex is not None:

345 ex.add_note(f"Add a 'with'-statement using '{cls.__name__}' somewhere up the call-hierarchy to receive and collect warnings.")

346 raise ex from warning

347

348

349@export

350class SupervisedWarningCollectorException(ExceptionBase):

351 """

352 This exception is raised when a supervised warning collector is not the top-most collector in its thread.

353

354 A supervised collector hands its warnings to the thread supervisor, which only works if nothing else collects them

355 first.

356 """

357

358

359@export

360class SupervisedWarningCollector(WarningCollector):

361 """

362 A context manager to collect warnings within the call hierarchy.

363 """

364 _supervisor: Nullable[ThreadSupervisor] #: Supervisor collecting warnings and exceptions of all threads.

365 _exceptionHandler: Nullable[Callable[[BaseException], bool]] #: Handler called for an exception escaping the thread.

366 _finallyHandler: Nullable[Callable[[], None]] #: Handler called when the thread ends, in either case.

367

368 __slots__ = ("_supervisor", "_exceptionHandler", "_finallyHandler")

369

370 def __init__(

371 self,

372 warnings: Nullable[list[BaseException]] = None,

373 handler: Nullable[Callable[[BaseException], bool]] = None,

374 /,

375 supervisor: Nullable[ThreadSupervisor] = None,

376 exceptionHandler: Nullable[Callable[[BaseException], bool]] = None,

377 finallyHandler: Nullable[Callable[[], None]] = None

378 ) -> None:

379 """

380 Initializes a warning collector.

381

382 :param warnings: Optional, reference to a list of warnings, which can be modified (appended) by this warning

383 collector. If ``None``, an internal list is created and can be referenced by the

384 collector's instance.

385 :param handler: Optional, handler function, which processes the current warning and decides if a warning

386 should be reraised as an exception.

387 :param supervisor: Optional, thread supervisor. On leaving the context, the collected warnings and an

388 exception leaving the block are handed to it, so the thread that started this one can

389 reraise them. Without a supervisor, an exception leaves the block unchanged.

390 :param exceptionHandler: Optional, handler function, called with an exception leaving the block when a supervisor is

391 set. Its result decides whether the exception is suppressed.

392 :param finallyHandler: Optional, function called when the context is left, whether or not an exception left it.

393 :raises TypeError: If optional parameter 'warnings' is not of type list.

394 :raises TypeError: If optional parameter 'handler' is not a callable.

395 """

396 super().__init__(warnings, handler)

397

398 self._supervisor = supervisor

399 self._exceptionHandler = exceptionHandler

400 self._finallyHandler = finallyHandler

401

402 def __enter__(self) -> Self:

403 """

404 Enter the warning collector context.

405

406 :returns: The warning collector instance.

407 :raises SupervisedWarningCollectorException: If this collector is not the top-most warning collector of its thread.

408 """

409 global _threadLocalData

410

411 if hasattr(_threadLocalData, "warningCollector") and _threadLocalData.warningCollector is not None:

412 raise SupervisedWarningCollectorException("This warning collector is not the top-most warning collector within the current thread.")

413

414 _threadLocalData.warningCollector = self

415

416 return self

417

418 def __exit__(

419 self,

420 exc_type: Nullable[type[BaseException]] = None,

421 exc_val: Nullable[BaseException] = None,

422 exc_tb: Nullable[TracebackType] = None

423 ) -> Nullable[bool]:

424 """

425 Exit the warning collector context.

426

427 :param exc_type: Exception type

428 :param exc_val: Exception instance

429 :param exc_tb: Exception's traceback.

430 :returns: ``None``

431 """

432 global _threadLocalData

433

434 _threadLocalData.warningCollector = None

435

436 if self._supervisor is not None:

437 result = True

438 if len(self._warnings) > 0:

439 self._supervisor.AddWarnings(self._warnings)

440

441 if exc_val is not None:

442 self._supervisor.AddException("", exc_val)

443

444 if self._exceptionHandler is not None:

445 result = self._exceptionHandler(exc_val)

446 else:

447 result = None

448

449 if self._finallyHandler is not None:

450 self._finallyHandler()

451

452 return result

453

454

455@export

456class SupervisedThreadException(ExceptionBase):

457 """

458 The exception is raise if a supervised thread received an unhandled exception which got collected by

459 :class:`ExceptionCollector`.

460 """

461 _threadName: Nullable[str] #: Name of the thread the exception was raised in.

462

463 def __init__(

464 self,

465 message: str,

466 /,

467 *,

468 threadName: Nullable[str] = None,

469 cause: Nullable[BaseException] = None

470 ) -> None:

471 """

472 Initializes the exception with the name of the thread that failed.

473

474 :param message: The exception's message.

475 :param threadName: Optional, name of the thread that raised the collected exception.

476 :param cause: Optional, the exception collected from that thread.

477 """

478 super().__init__(message)

479 self._threadName = threadName

480 self.__cause__ = cause

481

482 @readonly

483 def ThreadName(self) -> Nullable[str]:

484 """

485 Read-only property to access the name of the thread that raised the exception (:attr:`_threadName`).

486

487 :returns: Name of the thread, or ``None`` if it wasn't recorded.

488 """

489 return self._threadName

490

491

492@export

493class ThreadSupervisor:

494 """

495 Thread-safe collector of exceptions and warnings raised in worker threads for surfacing on another thread.

496

497 This thread supervisor should be used in combination with :class:`WarningCollector` to accumulate exceptions

498 (:class:`BaseException`) and warnings (:class:`CriticalWarning` or :class:`Warning`).

499

500 .. code-block:: python

501

502 @export

503 class MyThread(Thread):

504 def __init__(

505 self,

506 threadSupervisor: ThreadSupervisor,

507 stopEvent: Event

508 ) -> None:

509 super().__init__(name="MyThread", daemon=True)

510

511 self._threadSupervisor = threadSupervisor

512 self._stopEvent = stopEvent

513

514 def run(self) -> None:

515 def exceptionHandler(ex: BaseException) -> None:

516 self._stopEvent.set()

517

518 def finallyHandler() -> None:

519 # some finally code

520

521 with SupervisedWarningCollector(

522 supervisor=self._threadSupervisor,

523 exceptionHandler=exceptionHandler,

524 finallyHandler=finallyHandler

525 ) as warnings:

526 # Thread body

527

528 .. code-block:: python

529

530 def RunVivadoPipeline(

531 self,

532 ) -> list[AnyWarning]:

533 stopEvent = Event()

534 threadSupervisor = ThreadSupervisor()

535

536 myThread = MyThread(threadSupervisor, stopEvent)

537 myThread.start()

538

539 try:

540 myThread.join()

541 except KeyboardInterrupt:

542 stopEvent.set()

543 myThread.join(timeout=2.0)

544 raise

545

546 threadSupervisor.ReRaise()

547

548 return threadSupervisor.Warnings

549 """

550

551 _lock: Lock #: Lock serializing the collection from multiple threads.

552 _exceptions: list[tuple[str, BaseException]] #: Exceptions of all supervised threads, as (thread name, exception).

553 _warnings: list[tuple[str, AnyWarning]] #: Warnings of all supervised threads, as (thread name, warning).

554

555 __slots__ = ("_lock", "_exceptions", "_warnings")

556

557 def __init__(self) -> None:

558 """

559 Initialize a thread supervisor with empty lists of exceptions and warnings.

560 """

561 self._lock = Lock()

562 self._exceptions = []

563 self._warnings = []

564

565 @readonly

566 def HasWarning(self) -> bool:

567 """

568 Check if at least one warning was collected from a supervised thread.

569

570 :returns: ``True``, if at least one warning was collected.

571 """

572 with self._lock:

573 return len(self._warnings) > 0

574

575 @readonly

576 def HasExceptions(self) -> bool:

577 """

578 Check if at least one exception was collected from a supervised thread.

579

580 :returns: ``True``, if at least one exception was collected.

581 """

582 with self._lock:

583 return len(self._exceptions) > 0

584

585 @readonly

586 def Warnings(self) -> list[AnyWarning]:

587 """

588 Read-only property to return all warnings collected from supervised threads (:attr:`_warnings`).

589

590 :returns: List of collected warnings, without their thread names.

591 """

592 with self._lock:

593 return [warning for _, warning in self._warnings]

594

595 def AddWarning(self, threadName: str, warning: AnyWarning) -> None:

596 """

597 Collect a warning raised in a supervised thread.

598

599 :param threadName: Optional, name of the thread the warning was raised in.

600 :param warning: The warning to collect.

601 """

602 with self._lock:

603 self._warnings.append((threadName, warning))

604

605 def AddWarnings(self, threadName: str, warnings: list[AnyWarning]) -> None:

606 """

607 Collect several warnings raised in a supervised thread.

608

609 :param threadName: Optional, name of the thread the warnings were raised in.

610 :param warnings: Optional, the warnings to collect.

611 """

612 with self._lock:

613 self._warnings.extend((threadName, warning) for warning in warnings)

614

615 def AddException(self, threadName: str, ex: BaseException) -> None:

616 """

617 Collect an exception that escaped a supervised thread.

618

619 :param threadName: Optional, name of the thread the exception was raised in.

620 :param ex: The exception to collect.

621 """

622 with self._lock:

623 self._exceptions.append((threadName, ex))

624

625 def ReRaise(self, unwrapped: bool = False) -> None:

626 """

627 Re-raise the exceptions collected from the supervised threads in the supervising thread.

628

629 A single exception is re-raised as itself - wrapped in a :exc:`SupervisedThreadException` naming its thread,

630 unless ``unwrapped`` is set. Several exceptions are raised as an :exc:`ExceptionGroup`, so none of them is lost.

631

632 :param unwrapped: Optional, if ``True``, a single exception is raised as it was, without naming its

633 thread.

634 :raises SupervisedThreadException: If exactly one thread failed.

635 :raises ExceptionGroup: If more than one thread failed.

636 """

637 with self._lock:

638 if len(self._exceptions) == 0:

639 return

640

641 exceptions = list(self._exceptions)

642

643 if len(exceptions) == 1:

644 threadName, ex = exceptions[0]

645 if unwrapped:

646 raise ex

647 else:

648 raise SupervisedThreadException(f"Thread '{threadName}' failed.", threadName=threadName) from ex

649

650 elif unwrapped:

651 raise ExceptionGroup(

652 "Multiple threads failed.",

653 [ex for _, ex in exceptions]

654 )

655 else:

656 raise ExceptionGroup(

657 "Multiple threads failed.",

658 [

659 SupervisedThreadException(f"Thread '{threadName}' failed.", threadName=threadName, cause=ex)

660 for threadName, ex in exceptions

661 ]

662 )

Coverage for pyTooling/Warning/init.py: 72%

195 statements