I’ve used decorators before but have never built my own. I tried reading some docs and online help on this topic but I’m kind of confused at this point.
I have a server I want to make API calls to and it requires retrieving a token to make your api calls. If your token ever expires you will get back an http 401 and you just simply need to get another token.
Right now in all of my functions I just have an If statement checking for the 401 and refreshing the token if that is the case.
But this seems like the appropriate case to use a decorator but I’m kind of unsure how I would go about it.
Also, please let me know if this is not a good question for this site.
Here is my code now:
def get_token(self) -> str:
r = self.session.post(
BASE_URL + "authentication/signin",
data=dumps({"name": self.username, "password": self.password}),
verify=False,
)
if not r.ok:
raise Exception()
elif "token" not in (response := r.json()):
raise Exception("Token not found in response")
return response["token"]
def get_device(self, ip: str) -> dict[str, Any]:
data = dumps({"ipAddress": ip.strip()})
if (
r := self.session.post(BASE_URL + "devices/filter", data=data)
).status_code == 401:
self.session.headers.update({"X-AUTH-TOKEN": self.get_token()})
r = self.session.post(BASE_URL + "devices/filter", data=data)
if not r.ok:
raise Exception()
elif "content" not in (response := r.json()):
raise Exception()
return response["content"][0]
>Solution :
Instead of calling self.session.post directly, use a function that refactors the post/repost process. There’s no need for a decorator here.
class Foo:
def get_token(self) -> str:
# same as before
...
def _post(self, *args, **kwargs):
r = self.session.post(*args, **kwargs)
if r.status_code == 401:
self.session.headers.update({"X-AUTH-TOKEN": self.get_token()})
r = self.session.post(*args, **kwargs)
return r
def get_device(self, ip: str) -> dict[str, Any]:
data = dumps({"ipAddress": ip.strip()})
r = self._post(BASE_URL + "devices/filter", data=data)
if not r.ok:
raise Exception()
elif "content" not in (response := r.json()):
raise Exception()
return response["content"][0]