add FastAPI

service/auth_service.py

@@ -0,0 +1,72 @@
+import logging as log
+from typing import Type
+import requests
+from sqlalchemy.orm import Session
+from sqlalchemy import select, desc
+
+import constants
+from db.helpers import get_engine
+from db.models import ApiToken
+from constants import *
+
+
+def generate_token(persist=False):
+    body = {
+        "consumer_key": api_consumer_key,
+        "consumer_secret": api_consumer_secret,
+        "code": api_code
+    }
+    response = requests.post(f"{constants.api_url}/auth", json=body)
+
+    if response.status_code in [201, 200]:
+        log.info('Generated tray access token: %s', response.json())
+
+    if persist:
+        log.info('Persisting token')
+        persist_auth_token(response.json())
+
+    return response
+
+
+def refresh_access_token():
+    generate_token(persist=True)
+    token = get_last_api_token()
+
+    log.info('Refreshing token')
+    response = requests.get(f"{api_url}/auth?refresh_token={token.refresh_token}").json()
+
+    try:
+        persist_auth_token(response)
+        return response['access_token']
+    except Exception as e:
+        log.error(e)
+
+    return None
+
+
+def persist_auth_token(api_token):
+    with Session(get_engine()) as session:
+        token = ApiToken(message=api_token["message"],
+                         code=api_token["code"],
+                         access_token=api_token["access_token"],
+                         refresh_token=api_token["refresh_token"],
+                         date_expiration_access_token=api_token["date_expiration_access_token"],
+                         date_expiration_refresh_token=api_token["date_expiration_refresh_token"],
+                         date_activated=api_token["date_activated"],
+                         api_host=api_token["api_host"],
+                         store_id=api_token["store_id"])
+
+        session.add(token)
+        session.commit()
+
+
+def delete_token(token):
+    with Session(get_engine()) as session:
+        log.info('Deleting token %s', token.id)
+        session.delete(token)
+        session.commit()
+
+
+def get_last_api_token() -> Type[ApiToken] | None:
+    with Session(get_engine()) as session:
+        return session.query(ApiToken).order_by(desc(ApiToken.id)).first()