-
Notifications
You must be signed in to change notification settings - Fork 13
/
_utils.py
70 lines (49 loc) · 1.83 KB
/
_utils.py
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
62
63
64
65
66
67
68
69
70
import asyncio
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any, Awaitable, Iterable, List, Optional
from httpx import Response
if TYPE_CHECKING:
from ._types import BaseConfigProtocol
__all__ = 'get_config_attr', 'to_unix_s', 'utcnow', 'ManyTasks', 'pretty_response'
EPOCH = datetime(1970, 1, 1)
EPOCH_TZ = EPOCH.replace(tzinfo=timezone.utc)
def get_config_attr(config: 'BaseConfigProtocol', name: str) -> str:
try:
s = getattr(config, name)
except AttributeError:
raise TypeError(f'config has not attribute {name}')
if isinstance(s, str):
return s
else:
raise TypeError(f'config.{name} must be a string not {s.__class__.__name__}')
def to_unix_s(dt: datetime) -> int:
if dt.utcoffset() is None:
diff = dt - EPOCH
else:
diff = dt - EPOCH_TZ
return int(round(diff.total_seconds()))
def utcnow() -> datetime:
return datetime.utcnow().replace(tzinfo=timezone.utc)
class ManyTasks:
"""
Simply utility to start many tasks without awaiting them, then wait for them all to finish
"""
__slots__ = '_tasks'
def __init__(self) -> None:
self._tasks: List[asyncio.Task[Any]] = []
def add(self, coroutine: Awaitable[Any], *, name: Optional[str] = None) -> None:
task = asyncio.create_task(coroutine, name=name)
self._tasks.append(task)
async def finish(self) -> Iterable[Any]:
return await asyncio.gather(*self._tasks)
def pretty_response(r: Response) -> None: # pragma: no cover
from xml.etree import ElementTree
from devtools import debug
xml_root = ElementTree.fromstring(r.content)
debug(
status=r.status_code,
url=str(r.url),
headers=dict(r.request.headers),
history=r.history,
xml={el.tag: el.text for el in xml_root},
)