1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import asyncio
import functools
from bdb import BdbQuit
from typing import Callable, ParamSpec, TypeVar
import click
P = ParamSpec("P")
T = TypeVar("T")
def handle_exceptions(
application_main: Callable[P, T],
with_debugger: bool,
) -> Callable[P, T]:
"""Wraps a function so that it drops a user into a debugger if it raises an error.
This is intended to be used as a wrapper for the main function of a CLI application.
It will catch all errors and drop a user into a debugger if the error is not a
KeyboardInterrupt. If the error is a KeyboardInterrupt, it will raise the error.
If the error is not a KeyboardInterrupt, it will log the error and drop a user into a
debugger if with_debugger is True. If with_debugger is False, it will raise the error.
Parameters
----------
application_main
The function to wrap.
with_debugger
Whether to drop a user into a debugger if an error is raised.
Returns
-------
Callable
The wrapped function.
"""
@functools.wraps(application_main)
async def wrapped(*args: P.args, **kwargs: P.kwargs) -> T:
try:
return await application_main(*args, **kwargs)
except (BdbQuit, KeyboardInterrupt, click.Abort):
raise
except Exception as e:
if with_debugger:
print(f"Uncaught exception {e}")
import pdb
pdb.post_mortem()
else:
raise
return wrapped
def coroutine(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
return asyncio.run(f(*args, **kwargs))
return wrapper
|
