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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
import aiohttp
from auth0.authentication import Users
from auth0.authentication.base import AuthenticationBase
from auth0.rest import RestClientOptions
from auth0.rest_async import AsyncRestClient
def _gen_async(client, method):
m = getattr(client, method)
async def closure(*args, **kwargs):
return await m(*args, **kwargs)
return closure
def asyncify(cls):
methods = [
func
for func in dir(cls)
if callable(getattr(cls, func)) and not func.startswith("_")
]
class UsersAsyncClient(cls):
def __init__(
self,
domain,
telemetry=True,
timeout=5.0,
protocol="https",
):
super().__init__(domain, telemetry, timeout, protocol)
self.client = AsyncRestClient(None, telemetry=telemetry, timeout=timeout)
class AsyncManagementClient(cls):
def __init__(
self,
domain,
token,
telemetry=True,
timeout=5.0,
protocol="https",
rest_options=None,
):
super().__init__(domain, token, telemetry, timeout, protocol, rest_options)
self.client = AsyncRestClient(
jwt=token, telemetry=telemetry, timeout=timeout, options=rest_options
)
class AsyncAuthenticationClient(cls):
def __init__(
self,
domain,
client_id,
client_secret=None,
client_assertion_signing_key=None,
client_assertion_signing_alg=None,
telemetry=True,
timeout=5.0,
protocol="https",
):
super().__init__(
domain,
client_id,
client_secret,
client_assertion_signing_key,
client_assertion_signing_alg,
telemetry,
timeout,
protocol,
)
self.client = AsyncRestClient(
None,
options=RestClientOptions(
telemetry=telemetry, timeout=timeout, retries=0
),
)
class Wrapper(cls):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if cls == Users:
self._async_client = UsersAsyncClient(*args, **kwargs)
elif AuthenticationBase in cls.__bases__:
self._async_client = AsyncAuthenticationClient(*args, **kwargs)
else:
self._async_client = AsyncManagementClient(*args, **kwargs)
for method in methods:
setattr(
self,
f"{method}_async",
_gen_async(self._async_client, method),
)
def set_session(self, session):
"""Set Client Session to improve performance by reusing session.
Args:
session (aiohttp.ClientSession): The client session which should be closed
manually or within context manager.
"""
self._session = session
self._async_client.client.set_session(self._session)
async def __aenter__(self):
"""Automatically create and set session within context manager."""
self.set_session(aiohttp.ClientSession())
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Automatically close session within context manager."""
await self._session.close()
return Wrapper
|
