Merge pull request #91 from GunnerUjjwol/Integrate-session-storage-apis-from-v27

Integrate session-storage apis from v27

Author: cuongssh

privx_api/auth.py

@@ -1,8 +1,10 @@
 from http import HTTPStatus
+from typing import Optional
 
 from privx_api.base import BasePrivXAPI
 from privx_api.enums import UrlEnum
 from privx_api.response import PrivXAPIResponse
+from privx_api.utils import get_value
 
 
 class AuthAPI(BasePrivXAPI):
@@ -91,3 +93,105 @@ def regenerate_idp_client_config(self, idp_id: str) -> PrivXAPIResponse:
             UrlEnum.AUTH.REGENERATE_IDP_CLIENT, path_params={"idp_id": idp_id}
         )
         return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
+
+    def get_user_sessions(
+        self,
+        user_id: str,
+        offset: Optional[int] = None,
+        limit: Optional[int] = None,
+        sort_key: Optional[str] = None,
+        sort_dir: Optional[str] = None,
+    ) -> PrivXAPIResponse:
+        """
+        Fetch valid sessions for the user.
+
+        Returns:
+            PrivXAPIResponse
+        """
+        response_status, data = self._http_get(
+            UrlEnum.AUTH.USER_SESSIONS,
+            path_params={"user_id": user_id},
+        )
+        print(response_status, data)
+        return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
+
+    def get_source_sessions(
+        self,
+        source_id: str,
+        offset: Optional[int] = None,
+        limit: Optional[int] = None,
+        sort_key: Optional[str] = None,
+        sort_dir: Optional[str] = None,
+    ) -> PrivXAPIResponse:
+        """
+        Fetch valid source sessions.
+
+        Returns:
+            PrivXAPIResponse
+        """
+        response_status, data = self._http_get(
+            UrlEnum.AUTH.SOURCE_SESSIONS,
+            path_params={"source_id": source_id},
+        )
+        return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
+
+    def search_sessions(
+        self,
+        offset: Optional[int] = None,
+        limit: Optional[int] = None,
+        sort_key: Optional[str] = None,
+        sort_dir: Optional[str] = None,
+        search_payload: Optional[dict] = None,
+    ) -> PrivXAPIResponse:
+        """
+        Search for sessions.
+
+        Returns:
+            PrivXAPIResponse
+        """
+        search_params = self._get_search_params(
+            offset=offset, limit=limit, sortkey=sort_key, sortdir=sort_dir
+        )
+
+        response_status, data = self._http_post(
+            UrlEnum.AUTH.SEARCH_SESSIONS,
+            query_params=search_params,
+            body=get_value(search_payload, dict()),
+        )
+        return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
+
+    def terminate_session(self, session_id: str) -> PrivXAPIResponse:
+        """
+        Terminate single session by ID.
+
+        Returns:
+            PrivXAPIResponse
+        """
+        response_status, data = self._http_post(
+            UrlEnum.AUTH.TERMINATE_SESSION,
+            path_params={"session_id": session_id},
+        )
+        return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
+
+    def terminate_user_sessions(self, user_id: str) -> PrivXAPIResponse:
+        """
+        Terminate all sessions for a user.
+
+        Returns:
+            PrivXAPIResponse
+        """
+        response_status, data = self._http_post(
+            UrlEnum.AUTH.TERMINATE_USER_SESSIONS,
+            path_params={"user_id": user_id},
+        )
+        return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
+
+    def logout(self) -> PrivXAPIResponse:
+        """
+        User logout.
+
+        Returns:
+            PrivXAPIResponse
+        """
+        response_status, data = self._http_post(UrlEnum.AUTH.LOGOUT)
+        return PrivXAPIResponse(response_status, HTTPStatus.OK, data)

privx_api/enums.py

@@ -12,6 +12,12 @@ class AuthEnum:
     IDP_CLIENT = "AUTH.IDP_CLIENT"
     IDP_CLIENTS = "AUTH.IDP_CLIENTS"
     REGENERATE_IDP_CLIENT = "AUTH.REGENERATE_IDP_CLIENT"
+    USER_SESSIONS = "AUTH.USER_SESSIONS"
+    SOURCE_SESSIONS = "AUTH.SOURCE_SESSIONS"
+    SEARCH_SESSIONS = "AUTH.SEARCH_SESSIONS"
+    TERMINATE_SESSION = "AUTH.TERMINATE_SESSION"
+    TERMINATE_USER_SESSIONS = "AUTH.TERMINATE_USER_SESSIONS"
+    LOGOUT = "AUTH.LOGOUT"
 
     urls = {
         AUTHORIZE: "/auth/api/v1/oauth/authorize",
@@ -20,6 +26,14 @@ class AuthEnum:
         IDP_CLIENT: "/auth/api/v1/idp/clients/{idp_id}",
         IDP_CLIENTS: "/auth/api/v1/idp/clients",
         REGENERATE_IDP_CLIENT: "/auth/api/v1/idp/clients/{idp_id}/regenerate",
+        USER_SESSIONS: "/auth/api/v1/sessionstorage/users/{user_id}/sessions",
+        SOURCE_SESSIONS: "/auth/api/v1/sessionstorage/sources/{source_id}/sessions",
+        SEARCH_SESSIONS: "/auth/api/v1/sessionstorage/sessions/search",
+        TERMINATE_SESSION: "/auth/api/v1/sessionstorage/sessions/{session_id}"
+        "/terminate",
+        TERMINATE_USER_SESSIONS: "/auth/api/v1/sessionstorage/users/{user_id}"
+        "/sessions/terminate",
+        LOGOUT: "/auth/api/v1/logout",
     }