KrustyPlanet/trustcafe-api-wrapper
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Handle token method for saving token token to a file

Browse source
This commit is contained in:
simonwt 2026-04-13 22:26:42 +01:00
parent dcc5cb1cb6
commit deb19d78ae
3 changed files with 29 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -26,4 +26,6 @@ dist/*
# Test Artifacts # Test Artifacts
token_data.json token_data.json
token_data_alpha.json
token_dataproduction.json
test_output.json test_output.json

25
src/trustcafeapiwrapper/apiclient.py
View file

@ -1,7 +1,6 @@
import requests import os,requests
import simplejson as json import simplejson as json
from urllib.parse import urlencode from urllib.parse import urlencode
from pydantic import ( from pydantic import (
PrivateAttr, BaseModel, SecretStr, HttpUrl PrivateAttr, BaseModel, SecretStr, HttpUrl
) )
@ -153,6 +152,28 @@ class APIClient(BaseModel):
current_time = int(time.time()) current_time = int(time.time())
return self._access_token and current_time < self._access_token_timeout return self._access_token and current_time < self._access_token_timeout
def handle_token(self, token_data_path=None):
"""
Handle access token retrieval and caching.
"""
if token_data_path is None:
token_data_path = f"token_data_{self.environment}.json"
if os.path.exists(token_data_path):
with open(token_data_path, "r") as f:
token_data = json.load(f)
self.set_token(token_data)
else:
if self.debug:
print(f"No token data file found at '{token_data_path}'. A new token will be obtained.")
raise FileNotFoundError(f"Token data file '{token_data_path}' not found. Please ensure the file exists or handle token retrieval appropriately.")
# Get a new one if we don't have one or if the existing one is expired
if not self.is_token_valid():
tokendata = self.sign_in()
with open(token_data_path, "w") as f:
json.dump(tokendata, f, indent=2)
def run_job(self, job_function, *args, **kwargs): def run_job(self, job_function, *args, **kwargs):
""" """
Utility method to run a job function with the API client as the first argument. Utility method to run a job function with the API client as the first argument.

17
testing.py
View file

@ -26,16 +26,7 @@ API = APIClient(
# Keep a token cache to avoid unnecessary sign-ins during development. # Keep a token cache to avoid unnecessary sign-ins during development.
# (In production, you'd handle this more robustly and securely) # (In production, you'd handle this more robustly and securely)
if os.path.exists("token_data.json"): API.handle_token() # This will load the token from file if it exists and is valid, or sign in to get a new one if not.
with open("token_data.json", "r") as f:
token_data = json.load(f)
API.set_token(token_data)
# Get a new one if we don't have one or if the existing one is expired
if not API.is_token_valid():
tokendata = API.sign_in()
with open("token_data.json", "w") as f:
json.dump(tokendata, f, indent=2)
''' '''
END IMPORTANT BIT END IMPORTANT BIT
@ -50,8 +41,8 @@ def save_response(response):
# print("-----------Get a user profile----------------") # print("-----------Get a user profile----------------")
# profile = API.run_job('userprofile.get', "simon-little") profile = API.run_job('userprofile.get', "simon-little")
# print(profile) print(profile)
# print("-------------- Get a branch -----------------") # print("-------------- Get a branch -----------------")
# branch = API.run_job('branch.get', "music") # branch = API.run_job('branch.get', "music")
# print(branch) # print(branch)
@ -138,7 +129,7 @@ def save_response(response):
# x += 1 # x += 1
# save_response(API.run_job('notification.listnotifications')) # save_response(API.run_job('notification.listnotifications'))
save_response(API.run_job('notification.markallasread')) # save_response(API.run_job('notification.markallasread'))
# save_response(API.run_job('feed.cafefeed')) # save_response(API.run_job('feed.cafefeed'))