Merge pull request #31 from cterrazas2/token-response-object

Token scheduler added.

docs/explanation/token_scheduler.md

@@ -0,0 +1,27 @@
+# Token Scheduler
+
+## Least Privilege
+
+`pyvet` uses the principle of
+[least privilege](https://en.wikipedia.org/wiki/Principle_of_least_privilege)
+when interacting with various VA apis. A veteran must initially request and
+authorize access for each api and then `pyvet` will place the bearer token in a
+cache.
+
+Below is an example of the token cache for a va api
+name and its token, or key value pair respectively.
+
+```python3
+{
+    'veteran': 'somerandomtoken',
+    'claims': 'somerandomtoken',
+    ...
+}
+```
+
+### Eviction Policy
+
+The token cache will be updated on initial entry of a bearer token and then
+once it becomes invalid (expired or revoked). The refresh token will be used
+to retrieve another token. If the refresh token is expired, then the entire
+authentication process is initiated for the veteran's approval.

pyvet/benefits/claims/api.py

@@ -9,13 +9,11 @@
 from pyvet.client import (
     current_session as session,
 )
-from pyvet.client import (
-    get_bearer_token,
-)
+from pyvet.client import token_scheduler
 from pyvet.creds import API_URL
 from pyvet.json_alias import Json
 
-BENEFITS_INTAKE_URL = API_URL + "claims/v1/"
+BENEFITS_CLAIMS_URL = API_URL + "claims/v1/"
 CLAIM_SCOPE = "openid profile offline_access claim.read claim.write"
 
 
@@ -44,10 +42,10 @@ def get_claims(
     r : json
         Response in json format.
     """
-    claims_url = BENEFITS_INTAKE_URL + "claims"
+    claims_url = BENEFITS_CLAIMS_URL + "claims"
     session.headers[
         "Authorization"
-    ] = f"""Bearer {get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
+    ] = f"""Bearer {token_scheduler.get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
     if session.headers.get("Authorization") is None:
         return None
     if is_representative:
@@ -91,10 +89,10 @@ def get_claim(
     r : json
         Response in json format.
     """
-    claim_url = BENEFITS_INTAKE_URL + f"claims/{claim_id}"
+    claim_url = BENEFITS_CLAIMS_URL + f"claims/{claim_id}"
     session.headers[
         "Authorization"
-    ] = f"""Bearer {get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
+    ] = f"""Bearer {token_scheduler.get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
     if session.headers.get("Authorization") is None:
         return None
     if is_representative:
@@ -138,8 +136,8 @@ def submit_526(
     r : json
         Response in json format.
     """
-    submission_url = BENEFITS_INTAKE_URL + "forms/526"
-    token = get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)
+    submission_url = BENEFITS_CLAIMS_URL + "forms/526"
+    token = token_scheduler.get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)
     if session.headers.get("Authorization") is None:
         session.headers["Authorization"] = f"Bearer {token}"
 
@@ -275,10 +273,10 @@ def get_last_active_intent_to_file(
     r : json
         Response in json format.
     """
-    active_intent_url = BENEFITS_INTAKE_URL + "forms/0966/active"
+    active_intent_url = BENEFITS_CLAIMS_URL + "forms/0966/active"
     session.headers[
         "Authorization"
-    ] = f"""Bearer {get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
+    ] = f"""Bearer {token_scheduler.get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
     if session.headers.get("Authorization") is None:
         return None
     if is_representative:
@@ -383,10 +381,10 @@ def get_poa_status_by_id(
     r : json
         Response in json format.
     """
-    poa_url = BENEFITS_INTAKE_URL + f"forms/2122/{poa_id}"
+    poa_url = BENEFITS_CLAIMS_URL + f"forms/2122/{poa_id}"
     session.headers[
         "Authorization"
-    ] = f"""Bearer {get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
+    ] = f"""Bearer {token_scheduler.get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
     if session.headers.get("Authorization") is None:
         return None
     if is_representative:
@@ -427,10 +425,10 @@ def get_status_poa_last_active(
     r : json
         Response in json format.
     """
-    poa_url = BENEFITS_INTAKE_URL + "forms/2122/active"
+    poa_url = BENEFITS_CLAIMS_URL + "forms/2122/active"
     session.headers[
         "Authorization"
-    ] = f"""Bearer {get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
+    ] = f"""Bearer {token_scheduler.get_bearer_token(va_api="claims", scope=CLAIM_SCOPE)}"""
     if session.headers.get("Authorization") is None:
         return None
     if is_representative:

pyvet/client.py

@@ -3,7 +3,6 @@
 """
 import logging
 
-import oidc_client as oidc
 import requests
 from requests.adapters import HTTPAdapter
 from requests.packages.urllib3.util.retry import Retry
@@ -13,86 +12,8 @@
     API_FORCE_LIST,
     API_KEY_HEADER,
     API_RETRIES,
-    AUTH_SERVER,
-    CLIENT_ID,
-    DEFAULT_SCOPE,
-    ISSUER,
-    REDIRECT,
 )
-
-
-class TokenCache:
-    """A simple token cache."""
-
-    def __init__(self):
-        self.tokens: dict[str, str] = {}
-
-    def has_token(self, va_api: str) -> bool:
-        """Check if a token exists for a VA API.
-        Parameters
-        ----------
-        va_api : str
-            The VA API to check for a token.
-        Returns
-        -------
-        has_token : bool
-            Whether or not a token exists for the VA API.
-        """
-        has_token = va_api in self.tokens
-        return has_token
-
-    def get_token(self, va_api: str) -> str | None:
-        """Get a token for a VA API.
-        Parameters
-        ----------
-        va_api : str
-            The VA API to get a token for.
-        Returns
-        -------
-        token : str
-            A bearer token.
-        """
-        token = self.tokens.get(va_api)
-        return token if token else None
-
-
-def get_bearer_token(va_api: str, scope: str = DEFAULT_SCOPE) -> str | None:
-    """Get a bearer token from the VA OIDC server.
-    Parameters
-    ----------
-    va_api : str
-        The VA API to request a token for.
-    scope : str
-        A scope to request from the VA OIDC server (different per VA API).
-    Returns
-    -------
-    token : str
-        A bearer token.
-    """
-    try:
-        if token_cache.has_token(va_api):
-            return token_cache.get_token(va_api)
-        logging.error(
-            "No token found, requesting a new one for VA %s api.", va_api.capitalize()
-        )
-        token = oidc.login(
-            provider_config=oidc.config.ProviderConfig(
-                issuer=ISSUER,
-                authorization_endpoint=f"{AUTH_SERVER}/authorization",
-                token_endpoint=f"{AUTH_SERVER}/token",
-            ),
-            client_id=CLIENT_ID,
-            redirect_uri=REDIRECT,  # update this later
-            scope=scope,
-            interactive=True,
-        )
-        if token is None:
-            logging.error("Fetching token failed.")
-            return None
-        token_cache.tokens[va_api] = token.access_token
-        return token.access_token
-    except Exception as e:
-        logging.error(e)
+from pyvet.token_scheduler import TokenScheduler
 
 
 def create_session() -> requests.Session:
@@ -118,7 +39,7 @@ def create_session() -> requests.Session:
 
 
 current_session = create_session()
-token_cache = TokenCache()
+token_scheduler = TokenScheduler(session=current_session)
 
 if current_session.headers.get("apiKey") == "REPLACE ME":
     logging.error(

pyvet/health/community_care/api.py

@@ -8,9 +8,7 @@
 from pyvet.client import (
     current_session as session,
 )
-from pyvet.client import (
-    get_bearer_token,
-)
+from pyvet.client import token_scheduler
 from pyvet.creds import API_URL
 from pyvet.json_alias import Json
 
@@ -39,7 +37,7 @@ def get_eligibility(
     """
     session.headers[
         "Authorization"
-    ] = f"""Bearer {get_bearer_token(va_api="community_care", scope=HEALTH_SCOPE)}"""
+    ] = f"""Bearer {token_scheduler.get_bearer_token(va_api="community_care", scope=HEALTH_SCOPE)}"""
     if session.headers.get("Authorization") is None:
         return None
     status_url = HEALTH_URL + "search"