fastapi wrapper with auth example

Author: katzy687

quali_api/fast-api-proxy/config.json

@@ -0,0 +1,5 @@
+{
+  "CS_SERVER": "localhost",
+  "CS_DOMAIN": "Global",
+  "QUALI_API_PORT": 9000
+}

quali_api/fast-api-proxy/main.py

@@ -0,0 +1,54 @@
+import json
+import uvicorn as uvicorn
+
+from fastapi import Depends, FastAPI, HTTPException
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from dataclasses import dataclass
+from quali_api import QualiAPIHandler
+
+app = FastAPI()
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+@dataclass
+class CloudshellConfig:
+    server: str
+    domain: str
+    port: str
+
+
+def _get_config():
+    with open("config.json") as f:
+        data = json.load(f)
+    return CloudshellConfig(server=data["CS_SERVER"],
+                            domain=data["CS_DOMAIN"],
+                            port=data["QUALI_API_PORT"])
+
+
+config = _get_config()
+quali_api = QualiAPIHandler(host=config.server, port=config.port)
+
+
+@app.post("/token")
+async def login(form_data: OAuth2PasswordRequestForm = Depends()):
+    user_name = form_data.username
+    password = form_data.password
+    if not user_name and password:
+        raise HTTPException(status_code=400,
+                            detail="Must provide user and password")
+    response = quali_api.login(user_name=user_name, password=password, domain=config.domain)
+    if not response.ok:
+        raise HTTPException(status_code=response.status_code,
+                            detail=response.reason)
+    token = quali_api.get_token_from_login(response)
+    return {"access_token": token, "token_type": "bearer"}
+
+
+@app.get("/suites")
+async def read_users_me(token: str = Depends(oauth2_scheme)):
+    return quali_api.get_available_suites(token)
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7000)

quali_api/fast-api-proxy/quali_api.py

@@ -0,0 +1,37 @@
+import json
+
+import requests
+from requests import Response
+
+
+class QualiAPIHandler:
+    def __init__(self, host, port=9000):
+        self._base_url = f"http://{host}:{port}/Api"
+
+    def login(self, user_name: str = None, password: str = None, domain: str = "Global") -> Response:
+        headers = {"username": user_name, "password": password, "domain": domain}
+        return requests.put(f"{self._base_url}/Auth/Login", headers)
+
+    @staticmethod
+    def get_token_from_login(response: Response):
+        return response.text[1:-1]
+
+    @staticmethod
+    def _prep_auth_header(token):
+        return f"Basic {token}"
+
+    def get_available_suites(self, token) -> dict:
+        response = requests.get(self._base_url + "/Scheduling/SuiteTemplates",
+                                headers={"Authorization": self._prep_auth_header(token)})
+        if not response.ok:
+            response.raise_for_status()
+        return response.json()
+
+
+if __name__ == "__main__":
+    # sanity test
+    api = QualiAPIHandler("localhost")
+    token = api.login(user_name="admin", password="admin", domain="Global")
+    suites = api.get_available_suites(token)
+    print(json.dumps(suites, indent=4))
+    pass

quali_api/fast-api-proxy/requirements.txt

@@ -0,0 +1,4 @@
+python-multipart
+fastapi
+uvicorn
+requests