Document web framework integrations

.pre-commit-config.yaml

@@ -5,7 +5,9 @@ repos:
     hooks:
       - id: ruff-check
         args: [--fix, --exit-non-zero-on-fix]
+        exclude: ^doc/guides/_examples/
       - id: ruff-format
+        exclude: ^doc/guides/_examples/
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v6.0.0
     hooks:
@@ -19,7 +21,7 @@ repos:
     rev: v1.19.1
     hooks:
       - id: mypy
-        exclude: ^(tests/|conftest.py)
+        exclude: ^(tests/|conftest\.py|doc/)
         additional_dependencies:
           - pydantic[email]>=2.7.0
   - repo: https://github.com/codespell-project/codespell

doc/conf.py

@@ -43,6 +43,7 @@
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),
     "pydantic": ("https://docs.pydantic.dev/latest/", None),
+    "flask": ("https://flask.palletsprojects.com/en/stable/", None),
 }
 
 # -- Options for HTML output ----------------------------------------------

doc/guides/__init__.py

@@ -0,0 +1 @@
+"""Documentation guides package."""

doc/guides/_examples/__init__.py

@@ -0,0 +1 @@
+"""Runnable examples used by the documentation."""

doc/guides/_examples/django_example.py

@@ -0,0 +1,172 @@
+import json
+from http import HTTPStatus
+
+from django.http import HttpResponse
+from django.urls import path
+from django.urls import register_converter
+from django.utils.decorators import method_decorator
+from django.views import View
+from django.views.decorators.csrf import csrf_exempt
+from pydantic import ValidationError
+
+from scim2_models import Context
+from scim2_models import Error
+from scim2_models import ListResponse
+from scim2_models import PatchOp
+from scim2_models import SearchRequest
+from scim2_models import UniquenessException
+from scim2_models import User
+
+from .integrations import delete_record
+from .integrations import from_scim_user
+from .integrations import get_record
+from .integrations import list_records
+from .integrations import save_record
+from .integrations import to_scim_user
+
+# -- setup-start --
+def scim_response(payload, status=HTTPStatus.OK):
+    """Build a Django response with the SCIM media type."""
+    return HttpResponse(
+        payload,
+        status=status,
+        content_type="application/scim+json",
+    )
+# -- setup-end --
+
+
+# -- refinements-start --
+# -- converters-start --
+class UserConverter:
+    regex = "[^/]+"
+
+    def to_python(self, id):
+        try:
+            return get_record(id)
+        except KeyError:
+            raise ValueError
+
+    def to_url(self, record):
+        return record["id"]
+
+
+register_converter(UserConverter, "user")
+# -- converters-end --
+
+
+# -- validation-helper-start --
+def scim_validation_error(error):
+    """Turn Pydantic validation errors into a SCIM error response."""
+    scim_error = Error.from_validation_error(error.errors()[0])
+    return scim_response(scim_error.model_dump_json(), scim_error.status)
+# -- validation-helper-end --
+
+
+# -- uniqueness-helper-start --
+def scim_uniqueness_error(error):
+    """Turn uniqueness errors into a SCIM 409 response."""
+    scim_error = UniquenessException(detail=str(error)).to_error()
+    return scim_response(scim_error.model_dump_json(), HTTPStatus.CONFLICT)
+# -- uniqueness-helper-end --
+
+
+# -- error-handler-start --
+def handler404(request, exception):
+    """Turn Django 404 errors into SCIM error responses."""
+    scim_error = Error(status=404, detail=str(exception))
+    return scim_response(scim_error.model_dump_json(), HTTPStatus.NOT_FOUND)
+# -- error-handler-end --
+# -- refinements-end --
+
+
+# -- endpoints-start --
+# -- single-resource-start --
+@method_decorator(csrf_exempt, name="dispatch")
+class UserView(View):
+    """Handle GET, PATCH and DELETE on one SCIM user resource."""
+
+    def get(self, request, app_record):
+        scim_user = to_scim_user(app_record)
+        return scim_response(
+            scim_user.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
+        )
+
+    def delete(self, request, app_record):
+        delete_record(app_record["id"])
+        return scim_response("", HTTPStatus.NO_CONTENT)
+
+    def patch(self, request, app_record):
+        try:
+            patch = PatchOp[User].model_validate(
+                json.loads(request.body),
+                scim_ctx=Context.RESOURCE_PATCH_REQUEST,
+            )
+        except ValidationError as error:
+            return scim_validation_error(error)
+
+        scim_user = to_scim_user(app_record)
+        patch.patch(scim_user)
+
+        updated_record = from_scim_user(scim_user)
+        try:
+            save_record(updated_record)
+        except ValueError as error:
+            return scim_uniqueness_error(error)
+
+        return scim_response(
+            scim_user.model_dump_json(scim_ctx=Context.RESOURCE_PATCH_RESPONSE)
+        )
+# -- single-resource-end --
+
+
+# -- collection-start --
+@method_decorator(csrf_exempt, name="dispatch")
+class UsersView(View):
+    """Handle GET and POST on the SCIM users collection."""
+
+    def get(self, request):
+        try:
+            req = SearchRequest.model_validate(request.GET)
+        except ValidationError as error:
+            return scim_validation_error(error)
+        all_records = list_records()
+        page = all_records[req.start_index_0 : req.stop_index_0]
+        resources = [to_scim_user(record) for record in page]
+        response = ListResponse[User](
+            total_results=len(all_records),
+            start_index=req.start_index or 1,
+            items_per_page=len(resources),
+            resources=resources,
+        )
+        return scim_response(
+            response.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
+        )
+
+    def post(self, request):
+        try:
+            request_user = User.model_validate(
+                json.loads(request.body),
+                scim_ctx=Context.RESOURCE_CREATION_REQUEST,
+            )
+        except ValidationError as error:
+            return scim_validation_error(error)
+
+        app_record = from_scim_user(request_user)
+        try:
+            save_record(app_record)
+        except ValueError as error:
+            return scim_uniqueness_error(error)
+
+        response_user = to_scim_user(app_record)
+        return scim_response(
+            response_user.model_dump_json(scim_ctx=Context.RESOURCE_CREATION_RESPONSE),
+            HTTPStatus.CREATED,
+        )
+
+
+urlpatterns = [
+    path("scim/v2/Users", UsersView.as_view(), name="scim_users"),
+    path("scim/v2/Users/<user:app_record>", UserView.as_view(), name="scim_user"),
+]
+# -- collection-end --
+# -- endpoints-end --

doc/guides/_examples/flask_example.py

@@ -0,0 +1,166 @@
+from http import HTTPStatus
+
+from flask import Blueprint
+from flask import request
+from pydantic import ValidationError
+from werkzeug.routing import BaseConverter
+from werkzeug.routing import ValidationError as RoutingValidationError
+
+from scim2_models import Context
+from scim2_models import Error
+from scim2_models import ListResponse
+from scim2_models import PatchOp
+from scim2_models import SearchRequest
+from scim2_models import UniquenessException
+from scim2_models import User
+
+from .integrations import delete_record
+from .integrations import from_scim_user
+from .integrations import get_record
+from .integrations import list_records
+from .integrations import save_record
+from .integrations import to_scim_user
+
+# -- setup-start --
+bp = Blueprint("scim", __name__, url_prefix="/scim/v2")
+
+
+@bp.after_request
+def add_scim_content_type(response):
+    """Expose every endpoint with the SCIM media type."""
+    response.headers["Content-Type"] = "application/scim+json"
+    return response
+# -- setup-end --
+
+
+# -- refinements-start --
+# -- converters-start --
+class UserConverter(BaseConverter):
+    """Resolve a user identifier to an application record."""
+
+    def to_python(self, id):
+        try:
+            return get_record(id)
+        except KeyError:
+            raise RoutingValidationError()
+
+    def to_url(self, record):
+        return record["id"]
+
+
+@bp.record_once
+def _register_converter(state):
+    state.app.url_map.converters["user"] = UserConverter
+# -- converters-end --
+
+
+# -- error-handlers-start --
+@bp.errorhandler(ValidationError)
+def handle_validation_error(error):
+    """Turn Pydantic validation errors into SCIM error responses."""
+    scim_error = Error.from_validation_error(error.errors()[0])
+    return scim_error.model_dump_json(), scim_error.status
+
+
+@bp.errorhandler(404)
+def handle_not_found(error):
+    """Turn Flask 404 errors into SCIM error responses."""
+    scim_error = Error(status=404, detail=str(error.description))
+    return scim_error.model_dump_json(), HTTPStatus.NOT_FOUND
+
+
+@bp.errorhandler(ValueError)
+def handle_value_error(error):
+    """Turn uniqueness errors into SCIM 409 responses."""
+    scim_error = UniquenessException(detail=str(error)).to_error()
+    return scim_error.model_dump_json(), HTTPStatus.CONFLICT
+# -- error-handlers-end --
+# -- refinements-end --
+
+
+# -- endpoints-start --
+# -- single-resource-start --
+# -- get-user-start --
+@bp.get("/Users/<user:app_record>")
+def get_user(app_record):
+    """Return one SCIM user."""
+    scim_user = to_scim_user(app_record)
+    return (
+        scim_user.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
+        HTTPStatus.OK,
+    )
+# -- get-user-end --
+
+
+# -- patch-user-start --
+@bp.patch("/Users/<user:app_record>")
+def patch_user(app_record):
+    """Apply a SCIM PatchOp to an existing user."""
+    scim_user = to_scim_user(app_record)
+    patch = PatchOp[User].model_validate(
+        request.get_json(),
+        scim_ctx=Context.RESOURCE_PATCH_REQUEST,
+    )
+    patch.patch(scim_user)
+
+    updated_record = from_scim_user(scim_user)
+    save_record(updated_record)
+
+    return (
+        scim_user.model_dump_json(scim_ctx=Context.RESOURCE_PATCH_RESPONSE),
+        HTTPStatus.OK,
+    )
+# -- patch-user-end --
+
+
+# -- delete-user-start --
+@bp.delete("/Users/<user:app_record>")
+def delete_user(app_record):
+    """Delete an existing user."""
+    delete_record(app_record["id"])
+    return "", HTTPStatus.NO_CONTENT
+# -- delete-user-end --
+# -- single-resource-end --
+
+
+# -- collection-start --
+# -- list-users-start --
+@bp.get("/Users")
+def list_users():
+    """Return one page of users as a SCIM ListResponse."""
+    req = SearchRequest.model_validate(request.args)
+    all_records = list_records()
+    page = all_records[req.start_index_0 : req.stop_index_0]
+    resources = [to_scim_user(record) for record in page]
+    response = ListResponse[User](
+        total_results=len(all_records),
+        start_index=req.start_index or 1,
+        items_per_page=len(resources),
+        resources=resources,
+    )
+    return (
+        response.model_dump_json(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
+        HTTPStatus.OK,
+    )
+# -- list-users-end --
+
+
+# -- create-user-start --
+@bp.post("/Users")
+def create_user():
+    """Validate a SCIM creation payload and store the new user."""
+    request_user = User.model_validate(
+        request.get_json(),
+        scim_ctx=Context.RESOURCE_CREATION_REQUEST,
+    )
+    app_record = from_scim_user(request_user)
+    save_record(app_record)
+
+    response_user = to_scim_user(app_record)
+    return (
+        response_user.model_dump_json(scim_ctx=Context.RESOURCE_CREATION_RESPONSE),
+        HTTPStatus.CREATED,
+    )
+# -- create-user-end --
+# -- collection-end --
+# -- endpoints-end --