Move Organization operations to the top-level (#68)

Author: maxdeviant

tests/test_organizations.py

@@ -0,0 +1,83 @@
+import json
+from requests import Response
+
+import pytest
+
+import workos
+from workos.organizations import Organizations
+
+
+class TestOrganizations(object):
+    @pytest.fixture(autouse=True)
+    def setup(self, set_api_key):
+        self.organizations = Organizations()
+
+    @pytest.fixture
+    def mock_organization(self):
+        return {
+            "name": "Test Organization",
+            "object": "organization",
+            "id": "org_01EHT88Z8J8795GZNQ4ZP1J81T",
+            "domains": [
+                {
+                    "domain": "example.com",
+                    "object": "organization_domain",
+                    "id": "org_domain_01EHT88Z8WZEFWYPM6EC9BX2R8",
+                }
+            ],
+        }
+
+    @pytest.fixture
+    def mock_organizations(self):
+        return {
+            "object": "list",
+            "data": [
+                {
+                    "object": "organization",
+                    "id": "org_01EHQMYV6MBK39QC5PZXHY59C3",
+                    "name": "example.com",
+                    "domains": [
+                        {
+                            "object": "organization_domain",
+                            "id": "org_domain_01EHQMYV71XT8H31WE5HF8YK4A",
+                            "domain": "example.com",
+                        }
+                    ],
+                },
+                {
+                    "object": "organization",
+                    "id": "org_01EHQMVDTC2GRAHFCCRNTSKH46",
+                    "name": "example2.com",
+                    "domains": [
+                        {
+                            "object": "organization_domain",
+                            "id": "org_domain_01EHQMVDTZVA27PK614ME4YK7V",
+                            "domain": "example2.com",
+                        }
+                    ],
+                },
+            ],
+            "listMetadata": {"before": "before-id", "after": None},
+        }
+
+    def test_list_organizations(self, mock_organizations, mock_request_method):
+        mock_response = Response()
+        mock_response.status_code = 200
+        mock_response.response_dict = mock_organizations
+        mock_request_method("get", mock_response, 200)
+        response = self.organizations.list_organizations()
+        assert response.status_code == 200
+        assert len(response.response_dict["data"]) == 2
+
+    def test_create_organization(self, mock_organization, mock_request_method):
+        organization = {"domains": ["example.com"], "name": "Test Organization"}
+        mock_response = Response()
+        mock_response.status_code = 201
+        mock_response.response_dict = mock_organization
+        mock_request_method("post", mock_response, 201)
+
+        result = self.organizations.create_organization(organization)
+        subject = result.response_dict
+
+        assert subject["id"] == "org_01EHT88Z8J8795GZNQ4ZP1J81T"
+        assert subject["name"] == "Test Organization"

tests/test_portal.py

@@ -16,67 +16,6 @@ def setup(self, set_api_key):
     def mock_portal_link(self):
         return {"link": "https://id.workos.com/portal/launch?secret=secret"}
 
-    @pytest.fixture
-    def mock_organization(self):
-        return {
-            "name": "Test Organization",
-            "object": "organization",
-            "id": "org_01EHT88Z8J8795GZNQ4ZP1J81T",
-            "domains": [
-                {
-                    "domain": "example.com",
-                    "object": "organization_domain",
-                    "id": "org_domain_01EHT88Z8WZEFWYPM6EC9BX2R8",
-                }
-            ],
-        }
-
-    @pytest.fixture
-    def mock_organizations(self):
-        return {
-            "object": "list",
-            "data": [
-                {
-                    "object": "organization",
-                    "id": "org_01EHQMYV6MBK39QC5PZXHY59C3",
-                    "name": "example.com",
-                    "domains": [
-                        {
-                            "object": "organization_domain",
-                            "id": "org_domain_01EHQMYV71XT8H31WE5HF8YK4A",
-                            "domain": "example.com",
-                        }
-                    ],
-                },
-                {
-                    "object": "organization",
-                    "id": "org_01EHQMVDTC2GRAHFCCRNTSKH46",
-                    "name": "example2.com",
-                    "domains": [
-                        {
-                            "object": "organization_domain",
-                            "id": "org_domain_01EHQMVDTZVA27PK614ME4YK7V",
-                            "domain": "example2.com",
-                        }
-                    ],
-                },
-            ],
-            "listMetadata": {"before": "before-id", "after": None},
-        }
-
-    def test_create_organization(self, mock_organization, mock_request_method):
-        organization = {"domains": ["example.com"], "name": "Test Organization"}
-        mock_response = Response()
-        mock_response.status_code = 201
-        mock_response.response_dict = mock_organization
-        mock_request_method("post", mock_response, 201)
-
-        result = self.portal.create_organization(organization)
-        subject = result.response_dict
-
-        assert subject["id"] == "org_01EHT88Z8J8795GZNQ4ZP1J81T"
-        assert subject["name"] == "Test Organization"
-
     def test_generate_link_sso(self, mock_portal_link, mock_request_method):
         mock_response = Response()
         mock_response.status_code = 201
@@ -98,12 +37,3 @@ def test_generate_link_dsync(self, mock_portal_link, mock_request_method):
         subject = result.response_dict
 
         assert subject["link"] == "https://id.workos.com/portal/launch?secret=secret"
-
-    def test_list_organizations(self, mock_organizations, mock_request_method):
-        mock_response = Response()
-        mock_response.status_code = 200
-        mock_response.response_dict = mock_organizations
-        mock_request_method("get", mock_response, 200)
-        response = self.portal.list_organizations()
-        assert response.status_code == 200
-        assert len(response.response_dict["data"]) == 2

workos/organizations.py

@@ -0,0 +1,63 @@
+import workos
+from workos.utils.request import RequestHelper, REQUEST_METHOD_GET, REQUEST_METHOD_POST
+from workos.utils.validation import ORGANIZATIONS_MODULE, validate_settings
+
+ORGANIZATIONS_PATH = "organizations"
+RESPONSE_LIMIT = 10
+
+
+class Organizations(object):
+    @validate_settings(ORGANIZATIONS_MODULE)
+    def __init__(self):
+        pass
+
+    @property
+    def request_helper(self):
+        if not getattr(self, "_request_helper", None):
+            self._request_helper = RequestHelper()
+        return self._request_helper
+
+    def list_organizations(
+        self, domains=None, limit=RESPONSE_LIMIT, before=None, after=None
+    ):
+        """Retrieve a list of organizations that have connections configured within your WorkOS dashboard.
+
+        Kwargs:
+            domains (list): Filter organizations to only return those that are associated with the provided domains. (Optional)
+            limit (int): Maximum number of records to return. (Optional)
+            before (str): Pagination cursor to receive records before a provided Organization ID. (Optional)
+            after (str): Pagination cursor to receive records after a provided Organization ID. (Optional)
+
+        Returns:
+            dict: Organizations response from WorkOS.
+        """
+        params = {
+            "domains": domains,
+            "limit": limit,
+            "before": before,
+            "after": after,
+        }
+        return self.request_helper.request(
+            ORGANIZATIONS_PATH,
+            method=REQUEST_METHOD_GET,
+            params=params,
+            token=workos.api_key,
+        )
+
+    def create_organization(self, organization):
+        """Create an organization
+
+        Args:
+            organization (dict) - An organization object
+                organization[domains] (list) - List of domains that belong to the organization
+                organization[name] (str) - A unique, descriptive name for the organization
+
+        Returns:
+            dict: Created Organization response from WorkOS.
+        """
+        return self.request_helper.request(
+            ORGANIZATIONS_PATH,
+            method=REQUEST_METHOD_POST,
+            params=organization,
+            token=workos.api_key,
+        )

workos/portal.py

@@ -1,10 +1,8 @@
 import workos
-from workos.utils.request import RequestHelper, REQUEST_METHOD_GET, REQUEST_METHOD_POST
+from workos.utils.request import RequestHelper, REQUEST_METHOD_POST
 from workos.utils.validation import PORTAL_MODULE, validate_settings
 
-ORGANIZATIONS_PATH = "organizations"
 PORTAL_GENERATE_PATH = "portal/generate_link"
-RESPONSE_LIMIT = 10
 
 
 class Portal(object):
@@ -18,24 +16,6 @@ def request_helper(self):
             self._request_helper = RequestHelper()
         return self._request_helper
 
-    def create_organization(self, organization):
-        """Create an organization
-
-        Args:
-            organization (dict) - An organization object
-                organization[domains] (list) - List of domains that belong to the organization
-                organization[name] (str) - A unique, descriptive name for the organization
-
-        Returns:
-            dict: Created Organization response from WorkOS.
-        """
-        return self.request_helper.request(
-            ORGANIZATIONS_PATH,
-            method=REQUEST_METHOD_POST,
-            params=organization,
-            token=workos.api_key,
-        )
-
     def generate_link(self, intent, organization, return_url=None):
         """Generate a link to grant access to an organization's Admin Portal
 
@@ -60,30 +40,3 @@ def generate_link(self, intent, organization, return_url=None):
             params=params,
             token=workos.api_key,
         )
-
-    def list_organizations(
-        self, domains=None, limit=RESPONSE_LIMIT, before=None, after=None
-    ):
-        """Retrieve a list of organizations that have connections configured within your WorkOS dashboard.
-
-        Kwargs:
-            domains (list): Filter organizations to only return those that are associated with the provided domains. (Optional)
-            limit (int): Maximum number of records to return. (Optional)
-            before (str): Pagination cursor to receive records before a provided Organization ID. (Optional)
-            after (str): Pagination cursor to receive records after a provided Organization ID. (Optional)
-
-        Returns:
-            dict: Organizations response from WorkOS.
-        """
-        params = {
-            "domains": domains,
-            "limit": limit,
-            "before": before,
-            "after": after,
-        }
-        return self.request_helper.request(
-            ORGANIZATIONS_PATH,
-            method=REQUEST_METHOD_GET,
-            params=params,
-            token=workos.api_key,
-        )

workos/utils/validation.py

@@ -5,13 +5,15 @@
 
 AUDIT_TRAIL_MODULE = "AuditTrail"
 DIRECTORY_SYNC_MODULE = "DirectorySync"
+ORGANIZATIONS_MODULE = "Organizations"
 PASSWORDLESS_MODULE = "Passwordless"
 PORTAL_MODULE = "Portal"
 SSO_MODULE = "SSO"
 
 REQUIRED_SETTINGS_FOR_MODULE = {
     AUDIT_TRAIL_MODULE: ["api_key",],
     DIRECTORY_SYNC_MODULE: ["api_key",],
+    ORGANIZATIONS_MODULE: ["api_key"],
     PASSWORDLESS_MODULE: ["api_key",],
     PORTAL_MODULE: ["api_key",],
     SSO_MODULE: ["api_key", "client_id",],