clean code

Author: stremovsky

databunkerpro/__init__.py

@@ -6,4 +6,4 @@
 from .api import DatabunkerproAPI
 
 __version__ = "0.1.0"
-__all__ = ["DatabunkerproAPI"] 
+__all__ = ["DatabunkerproAPI"]

databunkerpro/api.py

@@ -8,10 +8,13 @@
 import json
 from datetime import datetime
 
+
 class DatabunkerproAPI:
     """Main client class for interacting with the DatabunkerPro API."""
 
-    def __init__(self, base_url: str, x_bunker_token: str = "", x_bunker_tenant: str = ""):
+    def __init__(
+        self, base_url: str, x_bunker_token: str = "", x_bunker_tenant: str = ""
+    ):
         """
         Initialize the DatabunkerPro API client.
 
@@ -20,12 +23,17 @@ def __init__(self, base_url: str, x_bunker_token: str = "", x_bunker_tenant: str
             x_bunker_token (str, optional): The X-Bunker-Token for authentication
             x_bunker_tenant (str, optional): The X-Bunker-Tenant identifier
         """
-        self.base_url = base_url.rstrip('/')
+        self.base_url = base_url.rstrip("/")
         self.x_bunker_token = x_bunker_token
         self.x_bunker_tenant = x_bunker_tenant
 
-    def _make_request(self, endpoint: str, method: str = "POST", data: Optional[Dict[str, Any]] = None,
-                     request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def _make_request(
+        self,
+        endpoint: str,
+        method: str = "POST",
+        data: Optional[Dict[str, Any]] = None,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         """
         Make a request to the DatabunkerPro API.
 
@@ -41,9 +49,7 @@ def _make_request(self, endpoint: str, method: str = "POST", data: Optional[Dict
         Raises:
             Exception: If the API request fails
         """
-        headers = {
-            "Content-Type": "application/json"
-        }
+        headers = {"Content-Type": "application/json"}
 
         if self.x_bunker_token:
             headers["X-Bunker-Token"] = self.x_bunker_token
@@ -75,8 +81,12 @@ def _make_request(self, endpoint: str, method: str = "POST", data: Optional[Dict
         except requests.exceptions.RequestException as e:
             raise Exception(f"Error making request: {str(e)}")
 
-    def create_user(self, profile: Dict[str, Any], options: Optional[Dict[str, Any]] = None,
-                   request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def create_user(
+        self,
+        profile: Dict[str, Any],
+        options: Optional[Dict[str, Any]] = None,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         """
         Create a new user in DatabunkerPro.
 
@@ -93,8 +103,12 @@ def create_user(self, profile: Dict[str, Any], options: Optional[Dict[str, Any]]
             data.update(options)
         return self._make_request("UserCreate", "POST", data, request_metadata)
 
-    def get_user(self, mode: str, identity: str,
-                request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def get_user(
+        self,
+        mode: str,
+        identity: str,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         """
         Get user information from DatabunkerPro.
 
@@ -106,10 +120,17 @@ def get_user(self, mode: str, identity: str,
         Returns:
             Dict[str, Any]: The API response
         """
-        return self._make_request("UserGet", "POST", {"mode": mode, "identity": identity}, request_metadata)
+        return self._make_request(
+            "UserGet", "POST", {"mode": mode, "identity": identity}, request_metadata
+        )
 
-    def update_user(self, mode: str, identity: str, profile: Dict[str, Any],
-                   request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def update_user(
+        self,
+        mode: str,
+        identity: str,
+        profile: Dict[str, Any],
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         """
         Update user information in DatabunkerPro.
 
@@ -122,12 +143,19 @@ def update_user(self, mode: str, identity: str, profile: Dict[str, Any],
         Returns:
             Dict[str, Any]: The API response
         """
-        return self._make_request("UserUpdate", "POST",
-                                {"mode": mode, "identity": identity, "profile": profile},
-                                request_metadata)
+        return self._make_request(
+            "UserUpdate",
+            "POST",
+            {"mode": mode, "identity": identity, "profile": profile},
+            request_metadata,
+        )
 
-    def delete_user(self, mode: str, identity: str,
-                   request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def delete_user(
+        self,
+        mode: str,
+        identity: str,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         """
         Delete a user from DatabunkerPro.
 
@@ -139,9 +167,13 @@ def delete_user(self, mode: str, identity: str,
         Returns:
             Dict[str, Any]: The API response
         """
-        return self._make_request("UserDelete", "POST", {"mode": mode, "identity": identity}, request_metadata)
+        return self._make_request(
+            "UserDelete", "POST", {"mode": mode, "identity": identity}, request_metadata
+        )
 
-    def get_system_stats(self, request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def get_system_stats(
+        self, request_metadata: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
         """
         Get system statistics from DatabunkerPro.
 
@@ -151,11 +183,17 @@ def get_system_stats(self, request_metadata: Optional[Dict[str, Any]] = None) ->
         Returns:
             Dict[str, Any]: The API response containing system statistics
         """
-        return self._make_request("SystemGetSystemStats", "POST", None, request_metadata)
+        return self._make_request(
+            "SystemGetSystemStats", "POST", None, request_metadata
+        )
 
-    def create_token(self, token_type: str, record: str,
-                    options: Optional[Dict[str, Any]] = None,
-                    request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def create_token(
+        self,
+        token_type: str,
+        record: str,
+        options: Optional[Dict[str, Any]] = None,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         """
         Create a token for sensitive data.
 
@@ -173,8 +211,9 @@ def create_token(self, token_type: str, record: str,
             data.update(options)
         return self._make_request("TokenCreate", "POST", data, request_metadata)
 
-    def get_token(self, token: str,
-                 request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def get_token(
+        self, token: str, request_metadata: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
         """
         Get token information from DatabunkerPro.
 
@@ -185,10 +224,13 @@ def get_token(self, token: str,
         Returns:
             Dict[str, Any]: The API response
         """
-        return self._make_request("TokenGet", "POST", {"token": token}, request_metadata)
+        return self._make_request(
+            "TokenGet", "POST", {"token": token}, request_metadata
+        )
 
-    def delete_token(self, token: str,
-                    request_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    def delete_token(
+        self, token: str, request_metadata: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
         """
         Delete a token from DatabunkerPro.
 
@@ -199,4 +241,6 @@ def delete_token(self, token: str,
         Returns:
             Dict[str, Any]: The API response
         """
-        return self._make_request("TokenDelete", "POST", {"token": token}, request_metadata) 
+        return self._make_request(
+            "TokenDelete", "POST", {"token": token}, request_metadata
+        )

setup.py

@@ -37,4 +37,4 @@
             "mypy>=0.910",
         ],
     },
-) 
+)

tests/test_api.py

@@ -8,31 +8,38 @@
 import requests
 from databunkerpro import DatabunkerproAPI
 
+
 class TestDatabunkerproAPI(unittest.TestCase):
     """Test cases for the DatabunkerPro API client."""
 
     @classmethod
     def setUpClass(cls):
         """Set up test environment before running tests."""
         # Try to get credentials from environment first
-        cls.api_url = os.getenv('DATABUNKER_API_URL', 'https://pro.databunker.org')
-        cls.api_token = os.getenv('DATABUNKER_API_TOKEN', '')
-        cls.tenant_name = os.getenv('DATABUNKER_TENANT_NAME', '')
+        cls.api_url = os.getenv("DATABUNKER_API_URL", "https://pro.databunker.org")
+        cls.api_token = os.getenv("DATABUNKER_API_TOKEN", "")
+        cls.tenant_name = os.getenv("DATABUNKER_TENANT_NAME", "")
 
         # If credentials are not in environment, try to get them from sandbox server
         if not all([cls.api_token, cls.tenant_name]):
             try:
-                response = requests.get('https://databunker.org/api/newtenant.php', verify=False)
+                response = requests.get(
+                    "https://databunker.org/api/newtenant.php", verify=False
+                )
                 if response.ok:
                     data = response.json()
-                    if data and data.get('status') == 'ok':
-                        cls.tenant_name = data['tenantname']
-                        cls.api_token = data['xtoken']
-                        print("\nSuccessfully connected to DatabunkerPro sandbox server")
+                    if data and data.get("status") == "ok":
+                        cls.tenant_name = data["tenantname"]
+                        cls.api_token = data["xtoken"]
+                        print(
+                            "\nSuccessfully connected to DatabunkerPro sandbox server"
+                        )
                         print(f"Tenant: {cls.tenant_name}")
                         print(f"API URL: {cls.api_url}")
                     else:
-                        raise unittest.SkipTest("Failed to get credentials from sandbox server")
+                        raise unittest.SkipTest(
+                            "Failed to get credentials from sandbox server"
+                        )
                 else:
                     raise unittest.SkipTest("Failed to connect to sandbox server")
             except Exception as e:
@@ -44,62 +51,62 @@ def setUpClass(cls):
         # Test connection
         try:
             result = cls.api.get_system_stats()
-            if not (isinstance(result, dict) and result.get('status') == 'ok'):
+            if not (isinstance(result, dict) and result.get("status") == "ok"):
                 raise unittest.SkipTest("Failed to connect to DatabunkerPro server")
         except Exception as e:
-            raise unittest.SkipTest(f"Failed to connect to DatabunkerPro server: {str(e)}")
+            raise unittest.SkipTest(
+                f"Failed to connect to DatabunkerPro server: {str(e)}"
+            )
 
     def test_create_user(self):
         """Test user creation."""
         user_data = {
-            'email': f'test{random.randint(1000, 999999)}@example.com',
-            'name': f'Test User {random.randint(1000, 999999)}',
-            'phone': str(random.randint(1000, 999999))
+            "email": f"test{random.randint(1000, 999999)}@example.com",
+            "name": f"Test User {random.randint(1000, 999999)}",
+            "phone": str(random.randint(1000, 999999)),
         }
         result = self.api.create_user(user_data)
         self.assertIsInstance(result, dict)
-        self.assertEqual(result.get('status'), 'ok')
-        self.assertIn('token', result)
-        return user_data['email']
+        self.assertEqual(result.get("status"), "ok")
+        self.assertIn("token", result)
+        return user_data["email"]
 
     def test_get_user(self):
         """Test user retrieval."""
         email = self.test_create_user()
-        result = self.api.get_user('email', email)
+        result = self.api.get_user("email", email)
         self.assertIsInstance(result, dict)
-        self.assertEqual(result.get('status'), 'ok')
-        self.assertIn('profile', result)
+        self.assertEqual(result.get("status"), "ok")
+        self.assertIn("profile", result)
         return email
 
     def test_update_user(self):
         """Test user update."""
         email = self.test_get_user()
-        update_data = {
-            'name': 'Updated Test User',
-            'phone': '+9876543210'
-        }
-        result = self.api.update_user('email', email, update_data)
+        update_data = {"name": "Updated Test User", "phone": "+9876543210"}
+        result = self.api.update_user("email", email, update_data)
         self.assertIsInstance(result, dict)
-        self.assertEqual(result.get('status'), 'ok')
+        self.assertEqual(result.get("status"), "ok")
 
         # Verify the update
-        updated_user = self.api.get_user('email', email)
-        self.assertEqual(updated_user['profile']['name'], update_data['name'])
-        self.assertEqual(updated_user['profile']['phone'], update_data['phone'])
+        updated_user = self.api.get_user("email", email)
+        self.assertEqual(updated_user["profile"]["name"], update_data["name"])
+        self.assertEqual(updated_user["profile"]["phone"], update_data["phone"])
 
     def test_delete_user(self):
         """Test user deletion."""
         email = self.test_create_user()
-        result = self.api.delete_user('email', email)
+        result = self.api.delete_user("email", email)
         self.assertIsInstance(result, dict)
-        self.assertEqual(result.get('status'), 'ok')
+        self.assertEqual(result.get("status"), "ok")
 
     def test_get_system_stats(self):
         """Test getting system statistics."""
         result = self.api.get_system_stats()
         self.assertIsInstance(result, dict)
-        self.assertEqual(result.get('status'), 'ok')
-        self.assertIn('stats', result)
+        self.assertEqual(result.get("status"), "ok")
+        self.assertIn("stats", result)
+
 
-if __name__ == '__main__':
-    unittest.main() 
+if __name__ == "__main__":
+    unittest.main()