将预定函数导入另一个文件 python
Import a scheduled functions to another file python
我正在制作一个工具,使用 python 将 ost 注册信息发送到 API。我遇到了令牌过期的问题。 token的过期时间是300秒,所以我要不断刷新token。有什么办法可以在 300 秒后安排令牌刷新吗?
我在一个名为 logAPI.py
的文件中有一个 python 脚本,其中包含多个任务、进程和函数(并且它有效)。
我有另一个名为 authorization.py
的文件,其中包含调用授权 API 和 returns 令牌的函数,因此我可以调用其他 APIs。
由于令牌过期时间为 300 秒,我打算在 300 秒后 return 一个新令牌。
我尝试了 while 函数、调度模块、cron 模块和导入 os 但我都失败了。
我认为这是最直接的解决方案,无需单独的线程来刷新令牌。
AuthToken
包含活动令牌以及刷新它所需的逻辑。当你需要一个令牌时,你调用AuthToken.get()
。如果它是第一次被调用,或者如果令牌已过期,这将获取一个新令牌。获取新令牌后,它会存储在 _auth_token
单例中。这会缓存它以备后用,并在它过期时通知我们。
听起来你在使用多线程,所以我添加了一个 Lock
来防止在调用 AuthToken.get()
时出现竞争条件
请务必在提出请求之前先致电 AuthToken.get()
。即使只是连续发出两个请求,令牌也有可能在发送第二个请求之前过期。
您可能还想将 TTL 调整得稍微低一些,以防发出请求时出现延迟。
import requests
import time
from threading import Lock
from typing import Optional
from typing_extensions import Self
class AuthToken:
"""Represents an auth token used by the API."""
# A singleton AuthToken instance
_auth_token: Optional[Self] = None
_lock = Lock()
# The token's TTL (in seconds)
ttl = 300
expires: int
token: str
def __init__(self, token):
self.token = token
self.expires = time.time() + self.ttl
def is_expired(self):
"""Returns True if the token is expired."""
return time.time() >= self.expires
@classmethod
def refresh(cls, lock=True):
if lock:
cls._lock.acquire()
# do API request...
token = "foo"
cls._auth_token = AuthToken(token)
if lock:
cls._lock.release()
@classmethod
def get(cls) -> Self:
"""Returns a valid auth token, refreshing the current one if it's expired."""
cls._lock.acquire()
if not cls._auth_token or cls._auth_token.is_expired():
cls.refresh(lock=False)
cls._lock.release()
return cls._auth_token
def request_that_needs_token():
auth = AuthToken.get()
# use auth.token in your request
requests.post(
"http://example.com", headers={"Authorization": f"Bearer {auth.token}"}
)
我正在制作一个工具,使用 python 将 ost 注册信息发送到 API。我遇到了令牌过期的问题。 token的过期时间是300秒,所以我要不断刷新token。有什么办法可以在 300 秒后安排令牌刷新吗?
我在一个名为 logAPI.py
的文件中有一个 python 脚本,其中包含多个任务、进程和函数(并且它有效)。
我有另一个名为 authorization.py
的文件,其中包含调用授权 API 和 returns 令牌的函数,因此我可以调用其他 APIs。
由于令牌过期时间为 300 秒,我打算在 300 秒后 return 一个新令牌。
我尝试了 while 函数、调度模块、cron 模块和导入 os 但我都失败了。
我认为这是最直接的解决方案,无需单独的线程来刷新令牌。
AuthToken
包含活动令牌以及刷新它所需的逻辑。当你需要一个令牌时,你调用AuthToken.get()
。如果它是第一次被调用,或者如果令牌已过期,这将获取一个新令牌。获取新令牌后,它会存储在 _auth_token
单例中。这会缓存它以备后用,并在它过期时通知我们。
听起来你在使用多线程,所以我添加了一个 Lock
来防止在调用 AuthToken.get()
请务必在提出请求之前先致电 AuthToken.get()
。即使只是连续发出两个请求,令牌也有可能在发送第二个请求之前过期。
您可能还想将 TTL 调整得稍微低一些,以防发出请求时出现延迟。
import requests
import time
from threading import Lock
from typing import Optional
from typing_extensions import Self
class AuthToken:
"""Represents an auth token used by the API."""
# A singleton AuthToken instance
_auth_token: Optional[Self] = None
_lock = Lock()
# The token's TTL (in seconds)
ttl = 300
expires: int
token: str
def __init__(self, token):
self.token = token
self.expires = time.time() + self.ttl
def is_expired(self):
"""Returns True if the token is expired."""
return time.time() >= self.expires
@classmethod
def refresh(cls, lock=True):
if lock:
cls._lock.acquire()
# do API request...
token = "foo"
cls._auth_token = AuthToken(token)
if lock:
cls._lock.release()
@classmethod
def get(cls) -> Self:
"""Returns a valid auth token, refreshing the current one if it's expired."""
cls._lock.acquire()
if not cls._auth_token or cls._auth_token.is_expired():
cls.refresh(lock=False)
cls._lock.release()
return cls._auth_token
def request_that_needs_token():
auth = AuthToken.get()
# use auth.token in your request
requests.post(
"http://example.com", headers={"Authorization": f"Bearer {auth.token}"}
)