feat: implement discoverability methods

Author: azmeuk

doc/changelog.rst

@@ -13,6 +13,7 @@ Changelog
 Added
 ^^^^^
 - Implement :meth:`~scim2_client.BaseSCIMClient.register_naive_resource_types`.
+- Implement :meth:`~scim2_client.BaseSyncSCIMClient.discover` methods.
 
 [0.3.3] - 2024-11-29
 --------------------

scim2_client/client.py

@@ -1,3 +1,4 @@
+import asyncio
 import sys
 from collections.abc import Collection
 from dataclasses import dataclass
@@ -29,6 +30,7 @@
     "Accept": "application/scim+json",
     "Content-Type": "application/scim+json",
 }
+CONFIG_RESOURCES = (ResourceType, Schema, ServiceProviderConfig)
 
 
 @dataclass
@@ -153,18 +155,27 @@ def __init__(
         check_response_payload: bool = True,
         raise_scim_errors: bool = True,
     ):
-        self.resource_models = tuple(
-            set(resource_models or []) | {ResourceType, Schema, ServiceProviderConfig}
-        )
+        self.resource_models = tuple(set(resource_models or []) | set(CONFIG_RESOURCES))
         self.resource_types = resource_types
         self.check_request_payload = check_request_payload
         self.check_response_payload = check_response_payload
         self.raise_scim_errors = raise_scim_errors
 
+    def get_resource_model(self, name: str) -> Optional[type[Resource]]:
+        """Get a registered model by its name or its schema."""
+        for resource_model in self.resource_models:
+            schema = resource_model.model_fields["schemas"].default[0]
+            if schema == name or schema.split(":")[-1] == name:
+                return resource_model
+        return None
+
     def check_resource_model(
         self, resource_model: type[Resource], payload=None
     ) -> None:
-        if resource_model not in self.resource_models:
+        if (
+            resource_model not in self.resource_models
+            and resource_model not in CONFIG_RESOURCES
+        ):
             raise SCIMRequestError(
                 f"Unknown resource type: '{resource_model}'", source=payload
             )
@@ -202,7 +213,7 @@ def register_naive_resource_types(self):
         self.resource_types = [
             ResourceType.from_resource(model)
             for model in self.resource_models
-            if model not in (ResourceType, Schema, ServiceProviderConfig)
+            if model not in CONFIG_RESOURCES
         ]
 
     def check_response(
@@ -759,6 +770,31 @@ def replace(
         """
         raise NotImplementedError()
 
+    def discover(self):
+        """Dynamically discover the server models :class:`~scim2_models.Schema` and :class:`~scim2_models.ResourceType`.
+
+        This is a shortcut for :meth:`BaseSyncSCIMClient.discover_models`
+        and :meth:`BaseSyncSCIMClient.discover_resource_types`.
+        """
+        self.discover_models()
+        self.discover_resource_types()
+
+    def discover_models(self):
+        """Dynamically register resource models by reading the server :class:`~scim2_models.Schema` endpoint.
+
+        Internally it performs a request to the SCIM server to get all the schemas,
+        generate classes from those schemas, and register them in the client.
+        """
+        schemas = self.query(Schema)
+        self.resource_models = tuple(
+            Resource.from_schema(schema) for schema in schemas.resources
+        )
+
+    def discover_resource_types(self):
+        """Dynamically register resource types by reading the server :class:`~scim2_models.ResourceType` endpoint."""
+        schemas = self.query(ResourceType)
+        self.resource_types = schemas.resources
+
 
 class BaseAsyncSCIMClient(BaseSCIMClient):
     """Base class for asynchronous request clients."""
@@ -1015,3 +1051,30 @@ async def replace(
             the response payload.
         """
         raise NotImplementedError()
+
+    async def discover(self):
+        """Dynamically discover the server models :class:`~scim2_models.Schema` and :class:`~scim2_models.ResourceType`.
+
+        This is a shortcut for the parallel execution of :meth:`BaseAsyncSCIMClient.discover_models`
+        and :meth:`BaseAsyncSCIMClient.discover_resource_types`.
+        """
+        models_task = asyncio.create_task(self.discover_models())
+        resources_task = asyncio.create_task(self.discover_resource_types())
+        await models_task
+        await resources_task
+
+    async def discover_models(self):
+        """Dynamically register resource models by reading the server :class:`~scim2_models.Schema` endpoint.
+
+        Internally it performs a request to the SCIM server to get all the schemas,
+        generate classes from those schemas, and register them in the client.
+        """
+        schemas = await self.query(Schema)
+        self.resource_models = tuple(
+            Resource.from_schema(schema) for schema in schemas.resources
+        )
+
+    async def discover_resource_types(self):
+        """Dynamically register resource types by reading the server :class:`~scim2_models.ResourceType` endpoint."""
+        schemas = await self.query(ResourceType)
+        self.resource_types = schemas.resources

tests/engines/test_httpx.py

@@ -5,9 +5,7 @@
 import pytest
 from httpx import AsyncClient
 from httpx import Client
-from scim2_models import ResourceType
 from scim2_models import SearchRequest
-from scim2_models import User
 
 from scim2_client.engines.httpx import AsyncSCIMClient
 from scim2_client.engines.httpx import SyncSCIMClient
@@ -16,14 +14,18 @@
 scim2_server = pytest.importorskip("scim2_server")
 from scim2_server.backend import InMemoryBackend  # noqa: E402
 from scim2_server.provider import SCIMProvider  # noqa: E402
+from scim2_server.utils import load_default_resource_types  # noqa: E402
+from scim2_server.utils import load_default_schemas  # noqa: E402
 
 
 @pytest.fixture(scope="session")
 def server():
     backend = InMemoryBackend()
     provider = SCIMProvider(backend)
-    provider.register_schema(User.to_schema())
-    provider.register_resource_type(ResourceType.from_resource(User))
+    for schema in load_default_schemas().values():
+        provider.register_schema(schema)
+    for resource_type in load_default_resource_types().values():
+        provider.register_resource_type(resource_type)
     host = "localhost"
     port = portpicker.pick_unused_port()
     httpd = wsgiref.simple_server.make_server(host, port, provider)
@@ -40,11 +42,9 @@ def server():
 def test_sync_engine(server):
     host, port = server
     client = Client(base_url=f"http://{host}:{port}")
-    scim_client = SyncSCIMClient(
-        client,
-        resource_models=[User],
-        resource_types=[ResourceType.from_resource(User)],
-    )
+    scim_client = SyncSCIMClient(client)
+    scim_client.discover()
+    User = scim_client.get_resource_model("User")
 
     request_user = User(user_name="foo", display_name="bar")
     response_user = scim_client.create(request_user)
@@ -77,11 +77,9 @@ def test_sync_engine(server):
 async def test_async_engine(server):
     host, port = server
     client = AsyncClient(base_url=f"http://{host}:{port}")
-    scim_client = AsyncSCIMClient(
-        client,
-        resource_models=(User,),
-        resource_types=[ResourceType.from_resource(User)],
-    )
+    scim_client = AsyncSCIMClient(client)
+    await scim_client.discover()
+    User = scim_client.get_resource_model("User")
 
     request_user = User(user_name="foo", display_name="bar")
     response_user = await scim_client.create(request_user)

tests/engines/test_werkzeug.py

@@ -1,5 +1,4 @@
 import pytest
-from scim2_models import ResourceType
 from scim2_models import SearchRequest
 from scim2_models import User
 from werkzeug.wrappers import Request
@@ -12,26 +11,29 @@
 scim2_server = pytest.importorskip("scim2_server")
 from scim2_server.backend import InMemoryBackend  # noqa: E402
 from scim2_server.provider import SCIMProvider  # noqa: E402
+from scim2_server.utils import load_default_resource_types  # noqa: E402
+from scim2_server.utils import load_default_schemas  # noqa: E402
 
 
 @pytest.fixture
 def scim_provider():
     provider = SCIMProvider(InMemoryBackend())
-    provider.register_schema(User.to_schema())
-    provider.register_resource_type(ResourceType.from_resource(User))
+    for schema in load_default_schemas().values():
+        provider.register_schema(schema)
+    for resource_type in load_default_resource_types().values():
+        provider.register_resource_type(resource_type)
     return provider
 
 
 @pytest.fixture
 def scim_client(scim_provider):
-    return TestSCIMClient(
-        app=scim_provider,
-        resource_models=(User,),
-        resource_types=[ResourceType.from_resource(User)],
-    )
+    client = TestSCIMClient(app=scim_provider)
+    client.discover()
+    return client
 
 
 def test_werkzeug_engine(scim_client):
+    User = scim_client.get_resource_model("User")
     request_user = User(user_name="foo", display_name="bar")
     response_user = scim_client.create(request_user)
     assert response_user.user_name == "foo"

tests/test_utils.py

@@ -34,3 +34,16 @@ class Foobar(Resource):
 
     with pytest.raises(SCIMRequestError):
         client.resource_endpoint(Foobar)
+
+
+def test_get_resource_model():
+    client = SyncSCIMClient(
+        None,
+        resource_models=[User[EnterpriseUser]],
+    )
+    assert client.get_resource_model("User") == User[EnterpriseUser]
+    assert (
+        client.get_resource_model("urn:ietf:params:scim:schemas:core:2.0:User")
+        == User[EnterpriseUser]
+    )
+    assert client.get_resource_model("Group") is None