Python 当令牌过期时,使用新令牌重新运行代码
Python rerun code with new token, when token has expired
我想为 api 调用编写某种类型的通用包装器,允许执行请求而不必担心令牌过期,在后台刷新令牌。
类似于在后台处理令牌刷新的上下文管理器,对用户不可见。为此,如果新令牌出现 TokenExpiredException,“包装器”必须能够重新 运行 代码。
例如,此代码使用 2 级 try/except 块,重复相同的调用,您必须将 api_call
作为字符串传递并重复调用 api 的代码:
def call_api_with_login(api_call: str, *args, **kwargs)
"""Call ``api_call`` method on the MyApi client with token error handling."""
def get_method(client: ApiClient, call: str):
"""Get method from a dot-separated string"""
return functools.reduce(getattr, call.split("."), client)
api = MyApi()
api_method = get_method(api.client, api_call)
try:
result = api_method(token, *args, **kwargs)
report_api_call()
except exceptions.TokenExpiredException as exc:
token = api.login().token
try:
result = api_method(token, *args, **kwargs)
except Exception as exc:
logging.exception(exc)
result = []
除了上面的代码重复和这个“模式”非常有限的事实之外,它会像这样使用:
call_api_with_login("books.list", author="Carl")
... 这有点糟糕,因为我们将方法名称作为字符串传递,无法访问代码助手,容易出错等
我最初的想法是我想使用上下文管理器之类的东西来处理这个问题,比如:
with authenticated_client as api_client, token:
api_client.books.list(token, author="xyz")
上下文管理器会产生客户端和令牌? ...但是,我无法想到在出现异常和刷新令牌的情况下重播内部代码(除非我在上下文管理器中进行某种循环,更像是一个生成器,也许吧?)
def authenticated_client():
api = MyApi()
token = cache_session.cache_get("API_TOKEN")
try:
yield api, token
except exceptions.TokenExpiredException as exc:
token = api.login().token
# ... how to rerun code?
希望这个例子在没有完全描述 api 客户和所有...
的情况下有一定意义
有人可以推荐一种 better/cleaner 方法来执行此操作,或者可以推荐其他方法来处理令牌刷新吗?
我尝试了上面解释的想法,第一个可行,但从长远来看并不是真正的好做法。
所以我所说的装饰的意思是想象你有
def api_call(token, *args, **kwargs):
... # some logic here
你的装饰器看起来像这样
def authorize_on_expire(func):
def wrapper(token, *args, **kwargs):
try:
result = func(token, *args, **kwargs)
except exceptions.TokenExpiredException as e:
token = ... # token refreshing logic
result = func(token, *args, **kwargs)
finally:
return result
return wrapper
你只需像这样装饰你的 api_call(...)
:
@authorize_on_expire
def api_call(token, *args, **kwargs):
... # some logic here
创建上下文管理器主要是为了在出错时安全关闭 streams/connections/etc。我有一个很好的例子是在出现任何错误时回滚数据库事务并在之后引发异常
我想为 api 调用编写某种类型的通用包装器,允许执行请求而不必担心令牌过期,在后台刷新令牌。
类似于在后台处理令牌刷新的上下文管理器,对用户不可见。为此,如果新令牌出现 TokenExpiredException,“包装器”必须能够重新 运行 代码。
例如,此代码使用 2 级 try/except 块,重复相同的调用,您必须将 api_call
作为字符串传递并重复调用 api 的代码:
def call_api_with_login(api_call: str, *args, **kwargs)
"""Call ``api_call`` method on the MyApi client with token error handling."""
def get_method(client: ApiClient, call: str):
"""Get method from a dot-separated string"""
return functools.reduce(getattr, call.split("."), client)
api = MyApi()
api_method = get_method(api.client, api_call)
try:
result = api_method(token, *args, **kwargs)
report_api_call()
except exceptions.TokenExpiredException as exc:
token = api.login().token
try:
result = api_method(token, *args, **kwargs)
except Exception as exc:
logging.exception(exc)
result = []
除了上面的代码重复和这个“模式”非常有限的事实之外,它会像这样使用:
call_api_with_login("books.list", author="Carl")
... 这有点糟糕,因为我们将方法名称作为字符串传递,无法访问代码助手,容易出错等
我最初的想法是我想使用上下文管理器之类的东西来处理这个问题,比如:
with authenticated_client as api_client, token:
api_client.books.list(token, author="xyz")
上下文管理器会产生客户端和令牌? ...但是,我无法想到在出现异常和刷新令牌的情况下重播内部代码(除非我在上下文管理器中进行某种循环,更像是一个生成器,也许吧?)
def authenticated_client():
api = MyApi()
token = cache_session.cache_get("API_TOKEN")
try:
yield api, token
except exceptions.TokenExpiredException as exc:
token = api.login().token
# ... how to rerun code?
希望这个例子在没有完全描述 api 客户和所有...
的情况下有一定意义有人可以推荐一种 better/cleaner 方法来执行此操作,或者可以推荐其他方法来处理令牌刷新吗?
我尝试了上面解释的想法,第一个可行,但从长远来看并不是真正的好做法。
所以我所说的装饰的意思是想象你有
def api_call(token, *args, **kwargs):
... # some logic here
你的装饰器看起来像这样
def authorize_on_expire(func):
def wrapper(token, *args, **kwargs):
try:
result = func(token, *args, **kwargs)
except exceptions.TokenExpiredException as e:
token = ... # token refreshing logic
result = func(token, *args, **kwargs)
finally:
return result
return wrapper
你只需像这样装饰你的 api_call(...)
:
@authorize_on_expire
def api_call(token, *args, **kwargs):
... # some logic here
创建上下文管理器主要是为了在出错时安全关闭 streams/connections/etc。我有一个很好的例子是在出现任何错误时回滚数据库事务并在之后引发异常