add timeout to FastAPIKeycloak init (#44)

fastapi_keycloak/api.py

@@ -135,6 +135,7 @@ def __init__(
             admin_client_secret: str,
             callback_uri: str,
             admin_client_id: str = "admin-cli",
+            timeout: int = 10,
     ):
         """FastAPIKeycloak constructor
 
@@ -147,6 +148,7 @@ def __init__(
             admin_client_secret (str): Secret for the `admin-cli` client
             callback_uri (str): Callback URL of the instance, used for auth flows. Must match at least one
             `Valid Redirect URIs` of Keycloak and should point to an endpoint that utilizes the authorization_code flow.
+            timeout (int): Timeout in seconds to wait for the server
         """
         self.server_url = server_url
         self.realm = realm
@@ -155,6 +157,7 @@ def __init__(
         self.admin_client_id = admin_client_id
         self.admin_client_secret = admin_client_secret
         self.callback_uri = callback_uri
+        self.timeout = timeout
         self._get_admin_token()  # Requests an admin access token on startup
 
     @property
@@ -273,7 +276,8 @@ def open_id_configuration(self) -> dict:
             dict: Open ID Configuration
         """
         response = requests.get(
-            url=f"{self.realm_uri}/.well-known/openid-configuration"
+            url=f"{self.realm_uri}/.well-known/openid-configuration",
+            timeout=self.timeout,
         )
         return response.json()
 
@@ -310,6 +314,7 @@ def proxy(
             url=f"{self.server_url}{relative_path}",
             data=json.dumps(payload),
             headers=headers,
+            timeout=self.timeout,
         )
 
     def _get_admin_token(self) -> None:
@@ -331,7 +336,7 @@ def _get_admin_token(self) -> None:
             "client_secret": self.admin_client_secret,
             "grant_type": "client_credentials",
         }
-        response = requests.post(url=self.token_uri, headers=headers, data=data)
+        response = requests.post(url=self.token_uri, headers=headers, data=data, timeout=self.timeout)
         try:
             self.admin_token = response.json()["access_token"]
         except JSONDecodeError as e:
@@ -353,7 +358,7 @@ def public_key(self) -> str:
         Returns:
             str: Public key for JWT decoding
         """
-        response = requests.get(url=self.realm_uri)
+        response = requests.get(url=self.realm_uri, timeout=self.timeout)
         public_key = response.json()["public_key"]
         return f"-----BEGIN PUBLIC KEY-----\n{public_key}\n-----END PUBLIC KEY-----"
 
@@ -936,7 +941,7 @@ def user_login(self, username: str, password: str) -> KeycloakToken:
             "password": password,
             "grant_type": "password",
         }
-        response = requests.post(url=self.token_uri, headers=headers, data=data)
+        response = requests.post(url=self.token_uri, headers=headers, data=data, timeout=self.timeout)
         if response.status_code == 401:
             raise HTTPException(status_code=401, detail="Invalid user credentials")
         if response.status_code == 400:
@@ -987,7 +992,7 @@ def exchange_authorization_code(
             "grant_type": "authorization_code",
             "redirect_uri": self.callback_uri,
         }
-        return requests.post(url=self.token_uri, headers=headers, data=data)
+        return requests.post(url=self.token_uri, headers=headers, data=data, timeout=self.timeout)
 
     def _admin_request(
             self,
@@ -1013,7 +1018,7 @@ def _admin_request(
             "Authorization": f"Bearer {self.admin_token}",
         }
         return requests.request(
-            method=method.name, url=url, data=json.dumps(data), headers=headers
+            method=method.name, url=url, data=json.dumps(data), headers=headers, timeout=self.timeout,
         )
 
     @functools.cached_property

tests/test_integration.py

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from fastapi.security import OAuth2PasswordBearer
 from jose import JWTError
+from requests import ReadTimeout
 
 from fastapi_keycloak import HTTPMethod
 from fastapi_keycloak.model import KeycloakRole
@@ -56,6 +57,14 @@ def test_proxy(self, idp):
         response = idp.proxy(relative_path="/realms/Test", method=HTTPMethod.GET)
         assert type(response.json()) == dict
 
+    def test_timeout(self, idp):
+        idp.timeout = 0.001
+        try:
+            idp.proxy(relative_path="/realms/Test", method=HTTPMethod.GET)
+            assert False
+        except ReadTimeout:
+            assert True
+
     def test_get_all_roles_and_get_roles(self, idp):
         roles: List[KeycloakRole] = idp.get_all_roles()
         assert roles