multiprocess compute done

CHANGELOG.md

@@ -22,10 +22,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Security 
 -->
 
-## [Unreleased] - YYYY-MM-DD
+## [0.1.0a7] - 2026-09-07
 
 ### Added
 - `SQLStorage` - SQL based storage module.
+- `MPCompute` - Multiprocess compute module that offloads compute operations to a worker process pool.
 - `StorageModule` and `ComputeModule` now automatically translate exceptions raised in their methods into the method's expected result body.
 - Full class and method docstrings for `StorageModule` and `ComputeModule`, including success and error return examples, call examples with the full config body, and notes on the automatic exception translation.
     - `ComputeModule` docstring notes that a compute module must handle all errors returned from storage.
@@ -35,8 +36,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `ComputeModule` and `StorageModule` base classes now inherit from ABC. 
 - `DictStorage` now stores storage latch `created_at` as an ISO 8601 string instead of a `datetime` object.
 
-### Deprecated
-
 ### Removed
 - `NotImplementedError` since base classes now use auto checks from ABC.
 
@@ -46,8 +45,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
     - The non-list (`get_*`) identity lookup in `validate_request` now populates the identity lookup and returns the correct identity error message.
     - `validate_batch_request` no longer raises `KeyError` on the non-list identity lookup path and no longer silently succeeds for an unregistered root definition in the list path.
 
-### Security 
-
 
 ## [0.1.0a6] - 2026-08-27
 

src/authzee/compute/compute_module.py

@@ -90,15 +90,15 @@ class ComputeModule(metaclass=_ComputeMeta):
     than implemented on compute:
 
     - The definition and grant persistence and retrieval configs:
-      `GetContextDefConfig`, `PutContextDefConfig`, `DeleteContextDefConfig`,
-      `GetIdentityDefConfig`, `PutIdentityDefConfig`, `DeleteIdentityDefConfig`,
-      `GetResourceDefConfig`, `PutResourceDefConfig`, `DeleteResourceDefConfig`,
-      `GetGrantConfig`, `EnactConfig`, and `RepealConfig`, along with the
-      standalone `ListContextDefsConfig`, `ListIdentityDefsConfig`,
-      `ListResourceDefsConfig`, `ListGrantsConfig`, and `ListGrantRefsConfig` as
-      top-level (non-embedded) configs.
+    `GetContextDefConfig`, `PutContextDefConfig`, `DeleteContextDefConfig`,
+    `GetIdentityDefConfig`, `PutIdentityDefConfig`, `DeleteIdentityDefConfig`,
+    `GetResourceDefConfig`, `PutResourceDefConfig`, `DeleteResourceDefConfig`,
+    `GetGrantConfig`, `EnactConfig`, and `RepealConfig`, along with the
+    standalone `ListContextDefsConfig`, `ListIdentityDefsConfig`,
+    `ListResourceDefsConfig`, `ListGrantsConfig`, and `ListGrantRefsConfig` as
+    top-level (non-embedded) configs.
     - The storage latch configs: `CreateLatchConfig`, `GetLatchConfig`,
-      `SetLatchConfig`, `DeleteLatchConfig`, and `CleanupLatchesConfig`.
+    `SetLatchConfig`, `DeleteLatchConfig`, and `CleanupLatchesConfig`.
 
     A compute module does still cause several of these storage calls to run (for
     example listing grants during an audit or authorize); when it does, it uses the

src/authzee/compute/mp_compute.py

@@ -0,0 +1,343 @@
+"""Multiprocess compute module for Authzee.
+
+All requests are offloaded to a worker process pool.
+"""
+
+__all__ = [
+    "MPCompute"
+]
+
+import asyncio
+from concurrent.futures import ProcessPoolExecutor
+import multiprocessing
+from typing import Any, Callable, Type
+
+from authzee.compute.compute_module import ComputeModule
+from authzee.compute.in_process_compute import InProcessCompute
+from authzee.module_locality import ModuleLocality
+from authzee.storage.storage_module import StorageModule
+from authzee.types.authzee import *
+from authzee.types.config import (
+    AuditConfig,
+    AuthorizeConfig,
+    BatchAuditConfig,
+    BatchAuthorizeConfig,
+    ComputeConstructConfig,
+    ComputeDestroyConfig,
+    ComputeShutdownConfig,
+    ComputeStartConfig,
+    ValidateBatchRequestConfig,
+    ValidateContextDefConfig,
+    ValidateGrantConfig,
+    ValidateIdentityDefConfig,
+    ValidateRequestConfig,
+    ValidateResourceDefConfig
+)
+
+
+class MPCompute(ComputeModule):
+    """Multiprocess Compute Module.
+
+    Parameters
+    ----------
+    max_workers : int | None
+        Maximum number of worker processes. If None, defaults to number of machine processors.
+    worker_compute : Type[ComputeModule]
+        The type of the compute module for each worker process to use.
+    worker_kwargs : dict[str, Any]
+        KWArgs to pass when creating compute modules for the worker processes.
+
+    Examples
+    --------
+    ```python
+    from authzee import (
+        Authzee,
+        DictStorage,
+        InProcessCompute,
+        jmespath_execute,
+        MPCompute
+    )
+
+
+    storage_dict = {}
+    authz = Authzee(
+        execute=jmespath_execute,
+        compute_type=MPCompute,
+        compute_kwargs={
+            "max_workers": None,
+            "worker_compute": InProcessCompute,
+            "worker_kwargs": {}
+        },
+        storage_type=DictStorage,
+        storage_kwargs={
+            "storage_dict": storage_dict
+        },
+        config={ # optional - AuthzeeConfigOverride | None - All keys are optional
+            "authzee": {
+                "raise_errors": True
+            }
+            # "method_name": {<method config>}
+        }
+    )
+    """
+
+
+    def __init__(
+        self,
+        max_workers: int | None,
+        worker_compute: Type[ComputeModule],
+        worker_kwargs: dict[str, Any]
+    ):
+        self._max_workers = max_workers
+        self._worker_compute = worker_compute
+        self._worker_kwargs = worker_kwargs
+        self._executor = None
+
+
+    async def start(
+        self,
+        execute: Callable[[str, Any], Any],
+        storage_type: Type[StorageModule],
+        storage_kwargs: dict[str, Any],
+        config: ComputeStartConfig
+    ) -> GenericResult:
+        await super().start(
+            execute=execute,
+            storage_type=storage_type,
+            storage_kwargs=storage_kwargs,
+            config=config
+        )
+        self.locality = ModuleLocality.SYSTEM
+        self.has_parallel_paging = False
+        self._executor = ProcessPoolExecutor(
+            max_workers=self._max_workers,
+            mp_context=multiprocessing.get_context("spawn"),
+            initializer=_executor_start,
+            initargs=(
+                self._worker_compute,
+                self._worker_kwargs,
+                execute,
+                storage_type,
+                storage_kwargs,
+                config
+            )
+        )
+
+        return {
+            "error": None
+        }
+
+
+    async def shutdown(self, config: ComputeShutdownConfig) -> GenericResult:
+        if self._executor is not None:
+            self._executor.shutdown(wait=True)
+            self._executor = None
+
+        return {
+            "error": None
+        }
+
+
+    async def construct(self, config: ComputeConstructConfig) -> GenericResult:
+        return {
+            "error": None
+        }
+
+
+    async def destroy(self, config: ComputeDestroyConfig) -> GenericResult:
+        return {
+            "error": None
+        }
+
+
+    async def validate_context_def(
+        self,
+        context_def: ContextDef,
+        config: ValidateContextDefConfig
+    ) -> GenericResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "validate_context_def",
+            {
+                "context_def": context_def,
+                "config": config
+            }
+        )
+
+
+    async def validate_identity_def(
+        self,
+        identity_def: IdentityDef,
+        config: ValidateIdentityDefConfig
+    ) -> GenericResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "validate_identity_def",
+            {
+                "identity_def": identity_def,
+                "config": config
+            }
+        )
+
+
+    async def validate_resource_def(
+        self,
+        resource_def: ResourceDef,
+        config: ValidateResourceDefConfig
+    ) -> GenericResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "validate_resource_def",
+            {
+                "resource_def": resource_def,
+                "config": config
+            }
+        )
+
+
+    async def validate_grant(
+        self,
+        grant: Grant,
+        config: ValidateGrantConfig
+    ) -> GenericResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "validate_grant",
+            {
+                "grant": grant,
+                "config": config
+            }
+        )
+
+
+    async def validate_request(
+        self,
+        request: AuthzeeRequest,
+        config: ValidateRequestConfig
+    ) -> GenericResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "validate_request",
+            {
+                "request": request,
+                "config": config
+            }
+        )
+
+
+    async def validate_batch_request(
+        self,
+        batch_request: AuthzeeBatchRequest,
+        config: ValidateBatchRequestConfig
+    ) -> ValidateBatchRequestResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "validate_batch_request",
+            {
+                "batch_request": batch_request,
+                "config": config
+            }
+        )
+
+
+    async def audit(
+        self,
+        request: AuthzeeRequest,
+        page_ref: str | None,
+        config: AuditConfig
+    ) -> AuditResultPage:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "audit",
+            {
+                "request": request,
+                "page_ref": page_ref,
+                "config": config
+            }
+        )
+
+
+    async def authorize(
+        self,
+        request: AuthzeeRequest,
+        config: AuthorizeConfig
+    ) -> AuthorizeResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "authorize",
+            {
+                "request": request,
+                "config": config
+            }
+        )
+
+
+    async def batch_audit(
+        self,
+        batch_request: AuthzeeBatchRequest,
+        page_ref: str | None,
+        config: BatchAuditConfig
+    ) -> BatchAuditResultPage:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "batch_audit",
+            {
+                "batch_request": batch_request,
+                "page_ref": page_ref,
+                "config": config
+            }
+        )
+
+
+    async def batch_authorize(
+        self,
+        batch_request: AuthzeeBatchRequest,
+        config: BatchAuthorizeConfig
+    ) -> BatchAuthorizeResult:
+        return await asyncio.get_running_loop().run_in_executor(
+            self._executor,
+            _executor_run,
+            "batch_authorize",
+            {
+                "batch_request": batch_request,
+                "config": config
+            }
+        )
+
+
+def _executor_start(
+    worker_compute: Type[ComputeModule],
+    worker_kwargs: dict[str, Any],
+    execute: Callable[[str, Any], Any],
+    storage_type: Type[StorageModule],
+    storage_kwargs: dict[str, Any],
+    config: ComputeStartConfig
+) -> None:
+    global _authzee_compute
+    _authzee_compute = worker_compute(**worker_kwargs)
+
+    return asyncio.run(
+        _authzee_compute.start(
+            execute=execute,
+            storage_type=storage_type,
+            storage_kwargs=storage_kwargs,
+            config=config
+        )
+    )
+
+
+def _executor_run(method: str, method_kwargs: dict[str, Any]) -> Any:
+    global _authzee_compute
+
+    return asyncio.run(
+        getattr(_authzee_compute, method)(**method_kwargs)
+    )