Remove runtime type checking for list types

Author: tribble

workos/directory_sync.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, TypedDict, Union
 import workos
 from workos.utils.pagination_order import PaginationOrder
 from workos.utils.request import (
@@ -17,6 +17,21 @@
 
 
 RESPONSE_LIMIT = 10
+DirectoryListFilters = TypedDict(
+    "DirectoryListFilters",
+    {"search": Optional[str], "organization": Optional[str], "domain": Optional[str]},
+    total=False,
+)
+DirectoryUserListFilters = TypedDict(
+    "DirectoryUserListFilters",
+    {"group": Optional[str], "directory": Optional[str]},
+    total=False,
+)
+DirectoryGroupListFilters = TypedDict(
+    "DirectoryGroupListFilters",
+    {"user": Optional[str], "directory": Optional[str]},
+    total=False,
+)
 
 
 class DirectorySync:
@@ -40,7 +55,7 @@ def list_users(
         before: Optional[str] = None,
         after: Optional[str] = None,
         order: PaginationOrder = "desc",
-    ) -> WorkOsListResource[DirectoryUser]:
+    ) -> WorkOsListResource[DirectoryUser, DirectoryUserListFilters]:
         """Gets a list of provisioned Users for a Directory.
 
         Note, either 'directory' or 'group' must be provided.
@@ -57,30 +72,31 @@ def list_users(
             dict: Directory Users response from WorkOS.
         """
 
-        params = {
+        list_params: ListArgs = {
             "limit": limit,
             "before": before,
             "after": after,
             "order": order,
         }
 
+        filter_params: DirectoryUserListFilters = {}
         if group is not None:
-            params["group"] = group
+            filter_params["group"] = group
         if directory is not None:
-            params["directory"] = directory
+            filter_params["directory"] = directory
 
         response = self.request_helper.request(
             "directory_users",
             method=REQUEST_METHOD_GET,
-            params=params,
+            params={**list_params, **filter_params},
             token=workos.api_key,
         )
 
         return WorkOsListResource(
             list_method=self.list_users,
-            # TODO: Should we even bother with this validation?
-            list_args=ListArgs.model_validate(params),
-            **ListPage[DirectoryUser](**response).model_dump()
+            list_args=list_params,
+            filter_params=filter_params,
+            **ListPage[DirectoryUser](**response).model_dump(),
         )
 
     def list_groups(
@@ -91,7 +107,7 @@ def list_groups(
         before: Optional[str] = None,
         after: Optional[str] = None,
         order: PaginationOrder = "desc",
-    ) -> WorkOsListResource[DirectoryGroup]:
+    ) -> WorkOsListResource[DirectoryGroup, DirectoryGroupListFilters]:
         """Gets a list of provisioned Groups for a Directory .
 
         Note, either 'directory' or 'user' must be provided.
@@ -107,29 +123,30 @@ def list_groups(
         Returns:
             dict: Directory Groups response from WorkOS.
         """
-        params = {
+        list_params: ListArgs = {
             "limit": limit,
             "before": before,
             "after": after,
             "order": order,
         }
+        filter_params: DirectoryGroupListFilters = {}
         if user is not None:
-            params["user"] = user
+            filter_params["user"] = user
         if directory is not None:
-            params["directory"] = directory
+            filter_params["directory"] = directory
 
         response = self.request_helper.request(
             "directory_groups",
             method=REQUEST_METHOD_GET,
-            params=params,
+            params={**list_params, **filter_params},
             token=workos.api_key,
         )
 
         return WorkOsListResource(
             list_method=self.list_groups,
-            # TODO: Should we even bother with this validation?
-            list_args=ListArgs.model_validate(params),
-            **ListPage[DirectoryGroup](**response).model_dump()
+            list_args=list_params,
+            filter_params=filter_params,
+            **ListPage[DirectoryGroup](**response).model_dump(),
         )
 
     def get_user(self, user: str):
@@ -193,7 +210,7 @@ def list_directories(
         after: Optional[str] = None,
         organization: Optional[str] = None,
         order: PaginationOrder = "desc",
-    ) -> WorkOsListResource[Directory]:
+    ) -> WorkOsListResource[Directory, DirectoryListFilters]:
         """Gets details for existing Directories.
 
         Args:
@@ -209,27 +226,28 @@ def list_directories(
             dict: Directories response from WorkOS.
         """
 
-        params = {
-            "domain": domain,
-            "organization": organization,
-            "search": search,
+        list_params: ListArgs = {
             "limit": limit,
             "before": before,
             "after": after,
             "order": order,
         }
-
+        filter_params: DirectoryListFilters = {
+            "domain": domain,
+            "search": search,
+            "organization": organization,
+        }
         response = self.request_helper.request(
             "directories",
             method=REQUEST_METHOD_GET,
-            params=params,
+            params={**list_params, **filter_params},
             token=workos.api_key,
         )
         return WorkOsListResource(
             list_method=self.list_directories,
-            # TODO: Should we even bother with this validation?
-            list_args=ListArgs.model_validate(params),
-            **ListPage[Directory](**response).model_dump()
+            list_args=list_params,
+            filter_params=filter_params,
+            **ListPage[Directory](**response).model_dump(),
         )
 
     def delete_directory(self, directory: str):

workos/organizations.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import List, Optional, TypedDict, Union
 import workos
 from workos.utils.pagination_order import PaginationOrder
 from workos.utils.request import (
@@ -17,6 +17,11 @@
 
 ORGANIZATIONS_PATH = "organizations"
 RESPONSE_LIMIT = 10
+OrganizationListFilters = TypedDict(
+    "OrganizationListFilters",
+    {"domains": Optional[List[str]]},
+    total=False,
+)
 
 
 class Organizations:
@@ -37,7 +42,7 @@ def list_organizations(
         before: Optional[str] = None,
         after: Optional[str] = None,
         order: PaginationOrder = "desc",
-    ) -> WorkOsListResource[Organization]:
+    ) -> WorkOsListResource[Organization, OrganizationListFilters]:
         """Retrieve a list of organizations that have connections configured within your WorkOS dashboard.
 
         Kwargs:
@@ -51,25 +56,26 @@ def list_organizations(
             dict: Organizations response from WorkOS.
         """
 
-        params = {
-            "domains": domains,
+        list_params: ListArgs = {
             "limit": limit,
             "before": before,
             "after": after,
             "order": order,
         }
 
+        filter_params: OrganizationListFilters = {"domains": domains}
+
         response = self.request_helper.request(
             ORGANIZATIONS_PATH,
             method=REQUEST_METHOD_GET,
-            params=params,
+            params={**list_params, **filter_params},
             token=workos.api_key,
         )
 
-        return WorkOsListResource[Organization](
+        return WorkOsListResource[Organization, OrganizationListFilters](
             list_method=self.list_organizations,
-            # TODO: Should we even bother with this validation?
-            list_args=ListArgs.model_validate(params),
+            list_args=list_params,
+            filter_params=filter_params,
             **ListPage[Organization](**response).model_dump()
         )
 

workos/resources/list.py

@@ -1,20 +1,27 @@
 from abc import abstractmethod
 from typing import (
+    Dict,
     List,
     Any,
     Literal,
+    Mapping,
+    Protocol,
+    Tuple,
     TypeVar,
     Generic,
     Callable,
     Iterator,
     Optional,
+    TypedDict,
 )
 
 from workos.resources.base import WorkOSBaseResource
 from workos.resources.directory_sync import Directory, DirectoryGroup, DirectoryUser
 from workos.resources.organizations import Organization
+from operator import itemgetter
+from pydantic import BaseModel, Field
 
-from pydantic import BaseModel, Extra, Field
+from workos.resources.workos_model import WorkOSModel
 
 # TODO: THIS OLD RESOURCE GOES AWAY
 
@@ -117,56 +124,55 @@ def auto_paging_iter(self):
     DirectoryGroup,
     DirectoryUser,
 )
+FilterParams = TypeVar("FilterParams")
 
 
 class ListMetadata(BaseModel):
     after: Optional[str] = None
     before: Optional[str] = None
 
 
-class ListPage(BaseModel, Generic[ListableResource]):
+class ListPage(WorkOSModel, Generic[ListableResource]):
     object: Literal["list"]
     data: List[ListableResource]
     list_metadata: ListMetadata
 
 
-class ListArgs(BaseModel, extra="allow"):
-    limit: Optional[int] = 10
-    before: Optional[str] = None
-    after: Optional[str] = None
-    order: Literal["asc", "desc"] = "desc"
-
-    class Config:
-        extra = "allow"
+class ListArgs(TypedDict):
+    limit: int
+    before: Optional[str]
+    after: Optional[str]
+    order: Literal["asc", "desc"]
 
 
-class WorkOsListResource(BaseModel, Generic[ListableResource]):
+class WorkOsListResource(
+    WorkOSModel,
+    Generic[ListableResource, FilterParams],
+):
     object: Literal["list"]
     data: List[ListableResource]
     list_metadata: ListMetadata
 
-    # These fields end up exposed in the types. Does we care?
     list_method: Callable = Field(exclude=True)
     list_args: ListArgs = Field(exclude=True)
+    filter_params: FilterParams = Field(exclude=True)
 
     def auto_paging_iter(self) -> Iterator[ListableResource]:
-        next_page: WorkOsListResource[ListableResource]
+        next_page: WorkOsListResource[ListableResource, FilterParams]
 
         after = self.list_metadata.after
-        order = self.list_args.order
-
-        fixed_pagination_params = {"order": order, "limit": self.list_args.limit}
-        filter_params = self.list_args.model_dump(
-            exclude={"after", "before", "order", "limit"}
-        )
 
+        fixed_pagination_params = {
+            "order": self.list_args["order"],
+            "limit": self.list_args["limit"],
+        }
         index: int = 0
 
         while True:
             if index >= len(self.data):
                 if after is not None:
                     next_page = self.list_method(
-                        after=after, **fixed_pagination_params, **filter_params
+                        after=after, **fixed_pagination_params, **self.filter_params
                     )
                     self.data = next_page.data
                     after = next_page.list_metadata.after