diff --git a/packages/cli/tests/bindings-test/src/test_analytics_engine.py b/packages/cli/tests/bindings-test/src/test_analytics_engine.py index 371228e..1b99a1f 100644 --- a/packages/cli/tests/bindings-test/src/test_analytics_engine.py +++ b/packages/cli/tests/bindings-test/src/test_analytics_engine.py @@ -1,5 +1,5 @@ import pytest -from workers._workers import _BindingWrapper +from workers.rpc import _BindingWrapper @pytest.mark.asyncio diff --git a/packages/cli/tests/bindings-test/src/test_assets.py b/packages/cli/tests/bindings-test/src/test_assets.py index 350a43b..1b37e56 100644 --- a/packages/cli/tests/bindings-test/src/test_assets.py +++ b/packages/cli/tests/bindings-test/src/test_assets.py @@ -1,5 +1,5 @@ import pytest -from workers._workers import _FetcherWrapper +from workers.rpc import _FetcherWrapper @pytest.mark.asyncio diff --git a/packages/cli/tests/bindings-test/src/test_images.py b/packages/cli/tests/bindings-test/src/test_images.py index fc687d7..496b1f1 100644 --- a/packages/cli/tests/bindings-test/src/test_images.py +++ b/packages/cli/tests/bindings-test/src/test_images.py @@ -3,7 +3,7 @@ import js import pytest from pyodide.ffi import create_proxy, to_js -from workers._workers import _BindingWrapper +from workers.rpc import _BindingWrapper PNG_1X1 = base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/5+hHgAHggJ/PchI7wAAAABJRU5ErkJggg==" diff --git a/packages/cli/tests/bindings-test/src/test_mtls.py b/packages/cli/tests/bindings-test/src/test_mtls.py index 00510b8..e3dcbc7 100644 --- a/packages/cli/tests/bindings-test/src/test_mtls.py +++ b/packages/cli/tests/bindings-test/src/test_mtls.py @@ -1,5 +1,5 @@ import pytest -from workers._workers import _FetcherWrapper +from workers.rpc import _FetcherWrapper # cannot test mTLS locally so we just check the binding is properly wrapped diff --git a/packages/cli/tests/bindings-test/src/test_ratelimit.py b/packages/cli/tests/bindings-test/src/test_ratelimit.py index 0023a81..340ec82 100644 --- a/packages/cli/tests/bindings-test/src/test_ratelimit.py +++ b/packages/cli/tests/bindings-test/src/test_ratelimit.py @@ -1,5 +1,5 @@ import pytest -from workers._workers import _BindingWrapper +from workers.rpc import _BindingWrapper @pytest.mark.asyncio diff --git a/packages/cli/tests/bindings-test/src/test_vectorize.py b/packages/cli/tests/bindings-test/src/test_vectorize.py index 38826ad..ddb42af 100644 --- a/packages/cli/tests/bindings-test/src/test_vectorize.py +++ b/packages/cli/tests/bindings-test/src/test_vectorize.py @@ -1,5 +1,5 @@ import pytest -from workers._workers import _BindingWrapper +from workers.rpc import _BindingWrapper @pytest.mark.asyncio diff --git a/packages/cli/tests/workerd-test/durable-object-inheritance/worker.py b/packages/cli/tests/workerd-test/durable-object-inheritance/worker.py index afe9e43..07306de 100644 --- a/packages/cli/tests/workerd-test/durable-object-inheritance/worker.py +++ b/packages/cli/tests/workerd-test/durable-object-inheritance/worker.py @@ -3,7 +3,8 @@ # _wrap_subclass must not double-wrap ctx and env when the hierarchy is >1 deep. from workers import DurableObject, WorkerEntrypoint -from workers._workers import DurableObjectContext, _EnvWrapper +from workers.entrypoints import DurableObjectContext +from workers.rpc import _EnvWrapper def assert_wrapped_once(obj): diff --git a/packages/cli/tests/workerd-test/sdk/tests/test_sdk.py b/packages/cli/tests/workerd-test/sdk/tests/test_sdk.py index 5fa39de..40fb975 100644 --- a/packages/cli/tests/workerd-test/sdk/tests/test_sdk.py +++ b/packages/cli/tests/workerd-test/sdk/tests/test_sdk.py @@ -13,7 +13,7 @@ env, fetch, ) -from workers._workers import _EnvWrapper +from workers.rpc import _EnvWrapper # TODO: Right now the `fetch` that's available on a binding is the JS fetch. # We may wish to rewrite it to be the same as the `fetch` defined in diff --git a/packages/runtime-sdk/src/workers/__init__.py b/packages/runtime-sdk/src/workers/__init__.py index ab7b070..eba91d0 100644 --- a/packages/runtime-sdk/src/workers/__init__.py +++ b/packages/runtime-sdk/src/workers/__init__.py @@ -1,16 +1,15 @@ -from ._workers import ( +from .blob import Blob, BlobEnding, BlobValue, File +from .entrypoints import ( DurableObject, WorkerEntrypoint, WorkflowEntrypoint, - _EnvWrapper, handler, ) -from .blob import Blob, BlobEnding, BlobValue, File from .fetch import fetch from .formdata import FormData, FormDataValue from .request import Request from .response import FetchResponse, Response -from .rpc import python_from_rpc, python_to_rpc +from .rpc import _EnvWrapper, python_from_rpc, python_to_rpc from .types import ( Body, Context, diff --git a/packages/runtime-sdk/src/workers/_workers.py b/packages/runtime-sdk/src/workers/entrypoints.py similarity index 62% rename from packages/runtime-sdk/src/workers/_workers.py rename to packages/runtime-sdk/src/workers/entrypoints.py index 064e248..50131e1 100644 --- a/packages/runtime-sdk/src/workers/_workers.py +++ b/packages/runtime-sdk/src/workers/entrypoints.py @@ -1,172 +1,28 @@ -# This module defines a Workers API for Python. It is similar to the API provided by -# JS Workers, but with changes and additions to be more idiomatic to the Python -# programming language. import functools import inspect from asyncio import create_task, gather from typing import TYPE_CHECKING, Any import _cloudflare_compat_flags - -# Get globals modules and import function from the entrypoint-helper import _pyodide_entrypoint_helper import js from js import Object -from pyodide.ffi import ( - JsProxy, - create_once_callable, - to_js, -) +from pyodide.ffi import create_once_callable, to_js -from .fetch import fetch from .request import Request -from .rpc import python_from_rpc, python_to_rpc -from .utils import ( - _JS_PASSTHROUGH_TYPES, - _from_js_error, - _get_js_constructor_name, - _is_js_instance, +from .rpc import ( + _BindingWrapper, + _EnvWrapper, + _idempotent_new, + python_from_rpc, + python_to_rpc, ) +from .utils import _from_js_error, _is_js_instance if TYPE_CHECKING: from js import DurableObjectState, Env, ExecutionContext -def _idempotent_new(cls, obj): - """Set __new__ on a class to this so cls is idempotent: - - >>> a = A(x) - >>> b = A(a) - >>> assert a is b - - For this to work, start the __init__ function with: - - if obj is self: - return - - to prevent double-init. - """ - if isinstance(obj, cls): - return obj - return object.__new__(cls) - - -class _BindingWrapper: - __new__ = _idempotent_new - - def __init__(self, binding): - if binding is self: - return - self._binding = binding - - @property - def _real_name(self): - js_name = _get_js_constructor_name(self._binding) - if not js_name: - # Should not happen, but just in case - return type(self).__name__ - return js_name - - def _should_wrap_nested_attribute(self, jsobj) -> bool: - if not isinstance(jsobj, JsProxy): - return False - - # TODO: This allowlist approach is a workaround. The long-term fix is to - # add dedicated Python wrappers for these types in python_from_rpc so they - # never reach _BindingWrapper in the first place. - js_type = _get_js_constructor_name(jsobj) - return js_type and js_type not in _JS_PASSTHROUGH_TYPES - - def _convert_result(self, result): - converted = python_from_rpc(result) - - # After python_from_rpc, some objects may still be JsProxy objects. - # We need to wrap them with _BindingWrapper (or a subclass of it) again - # to ensure that accessing attributes on them will be properly converted. - if self._should_wrap_nested_attribute(converted): - return self.__class__(converted) - if isinstance(converted, list): - return [ - self.__class__(item) - if self._should_wrap_nested_attribute(item) - else item - for item in converted - ] - return converted - - @staticmethod - def _convert_args(args, kwargs): - js_args = [python_to_rpc(arg) for arg in args] - js_kwargs = {k: python_to_rpc(v) for k, v in kwargs.items()} - return js_args, js_kwargs - - def _getattr_helper(self, name): - attr = getattr(self._binding, name) - - if not callable(attr): - return self._convert_result(attr) - - def wrapper(*args, **kwargs): - js_args, js_kwargs = self._convert_args(args, kwargs) - result = attr(*js_args, **js_kwargs) - if hasattr(result, "then") and callable(result.then): - - async def await_and_convert(): - return self._convert_result(await result) - - return await_and_convert() - return self._convert_result(result) - - return wrapper - - def __getattr__(self, name): - result = self._getattr_helper(name) - setattr(self, name, result) - return result - - def __getitem__(self, key): - if isinstance(key, int): - return self._convert_result(self._binding[key]) - return self._convert_result(getattr(self._binding, key)) - - def __iter__(self): - binding = self._binding - if not hasattr(binding, "__iter__"): - raise TypeError(f"'{self._real_name}' object is not iterable") - for item in binding: - yield self._convert_result(item) - - def __len__(self): - binding = self._binding - if not hasattr(binding, "length"): - raise TypeError(f"'{self._real_name}' object has no len()") - return binding.length - - -class _FetcherWrapper(_BindingWrapper): - def fetch(self, *args, **kwargs): - return fetch(*args, fetcher=self._binding.fetch, **kwargs) - - -class _DurableObjectNamespaceWrapper: - def __init__(self, binding): - self._binding = binding - - def __getattr__(self, name): - return getattr(self._binding, name) - - def get(self, *args, **kwargs): - return _FetcherWrapper(self._binding.get(*args, **kwargs)) - - def getByName(self, *args, **kwargs): - return _FetcherWrapper(self._binding.getByName(*args, **kwargs)) - - def jurisdiction(self, *args, **kwargs): - return _DurableObjectNamespaceWrapper( - self._binding.jurisdiction(*args, **kwargs) - ) - - class DurableObjectAbort(BaseException): pass @@ -212,103 +68,6 @@ def abort(self, reason: str | None = None): raise DurableObjectAbort(reason or "Durable Object abort requested") -class _WorkflowInstanceWrapper(_BindingWrapper): - # status/pause/resume/restart/terminate share their JS names and are handled by - # the _BindingWrapper, which already converts arguments and results. - # Only send_event needs the snake_case -> camelCase mapping for backward compatibility - - async def send_event(self, *args, **kwargs): - js_args, js_kwargs = self._convert_args(args, kwargs) - return self._convert_result( - await self._binding.sendEvent(*js_args, **js_kwargs) - ) - - -class _WorkflowBindingWrapper(_BindingWrapper): - async def get(self, *args, **kwargs): - js_args, js_kwargs = self._convert_args(args, kwargs) - return _WorkflowInstanceWrapper(await self._binding.get(*js_args, **js_kwargs)) - - async def create(self, *args, **kwargs): - js_args, js_kwargs = self._convert_args(args, kwargs) - return _WorkflowInstanceWrapper( - await self._binding.create(*js_args, **js_kwargs) - ) - - async def create_batch(self, *args, **kwargs): - js_args, js_kwargs = self._convert_args(args, kwargs) - return [ - _WorkflowInstanceWrapper(w) - for w in await self._binding.createBatch(*js_args, **js_kwargs) - ] - - -class _EnvWrapper: - _BINDING_TYPES = { - "KvNamespace", - "R2Bucket", - "D1Database", - "WorkerQueue", - "Ai", - "VectorizeIndexImpl", - "AnalyticsEngineDataset", - "LocalAnalyticsEngineDataset", - "ImagesBindingImpl", - "HostedImagesBindingImpl", - "Ratelimit", - } - - __new__ = _idempotent_new - - def __init__(self, env: Any): - if env is self: - return - self._env = env - - def _getattr_helper(self, name): - binding = getattr(self._env, name) - if _is_js_instance(binding, "Fetcher"): - return _FetcherWrapper(binding) - - if _is_js_instance(binding, "DurableObjectNamespace"): - return _DurableObjectNamespaceWrapper(binding) - - if _is_js_instance(binding, "WorkflowImpl"): - return _WorkflowBindingWrapper(binding) - - if _is_js_instance(binding, self._BINDING_TYPES): - return _BindingWrapper(binding) - - return binding - - def __getattr__(self, name): - result = self._getattr_helper(name) - setattr(self, name, result) - return result - - -def handler(func): - """ - When applied to handlers such as `on_fetch` it will rewrite arguments passed in to native Python - types defined in this module. For example, the `request` argument to `on_fetch` gets converted - to an instance of the Request class defined in this module. - """ - - @functools.wraps(func) - def wrapper(*args, **kwargs): - # TODO: support transforming kwargs - if len(args) > 0 and _is_js_instance(args[0], "Request"): - args = (Request(args[0]), *args[1:]) - - # Wrap `env` so that bindings can be used without to_js. - if len(args) > 1: - args = (args[0], _EnvWrapper(args[1]), *args[2:]) - - return func(*args, **kwargs) - - return wrapper - - class _WorkflowStepWrapper: __new__ = _idempotent_new @@ -485,6 +244,28 @@ async def _closure(): return result +def handler(func): + """ + When applied to handlers such as `on_fetch` it will rewrite arguments passed in to native Python + types defined in this module. For example, the `request` argument to `on_fetch` gets converted + to an instance of the Request class defined in this module. + """ + + @functools.wraps(func) + def wrapper(*args, **kwargs): + # TODO: support transforming kwargs + if len(args) > 0 and _is_js_instance(args[0], "Request"): + args = (Request(args[0]), *args[1:]) + + # Wrap `env` so that bindings can be used without to_js. + if len(args) > 1: + args = (args[0], _EnvWrapper(args[1]), *args[2:]) + + return func(*args, **kwargs) + + return wrapper + + def _wrap_class(cls): # Override the class __init__ so that we can wrap the `env` in the constructor. original_init = cls.__dict__.get("__init__") @@ -533,6 +314,22 @@ async def wrapped_run(self, event=None, step=None, /, *args, **kwargs): cls.run = wrapped_run +def _wrap_queue_handler(cls): + queue_fn = getattr(cls, "queue", None) + if queue_fn is None: + return + + @functools.wraps(queue_fn) + async def wrapped_queue(self, batch, *args, **kwargs): + wrapped_batch = _BindingWrapper(batch) + result = queue_fn(self, wrapped_batch, *args, **kwargs) + if inspect.iscoroutine(result): + result = await result + return result + + cls.queue = wrapped_queue + + @_wrap_class class DurableObject: """ @@ -584,19 +381,3 @@ def __init__(self, ctx: "ExecutionContext", env: "Env"): def __init_subclass__(cls, **_kwargs: Any): _wrap_class(cls) _wrap_workflow_step(cls) - - -def _wrap_queue_handler(cls): - queue_fn = getattr(cls, "queue", None) - if queue_fn is None: - return - - @functools.wraps(queue_fn) - async def wrapped_queue(self, batch, *args, **kwargs): - wrapped_batch = _BindingWrapper(batch) - result = queue_fn(self, wrapped_batch, *args, **kwargs) - if inspect.iscoroutine(result): - result = await result - return result - - cls.queue = wrapped_queue diff --git a/packages/runtime-sdk/src/workers/rpc.py b/packages/runtime-sdk/src/workers/rpc.py index 6f00fbf..bafccde 100644 --- a/packages/runtime-sdk/src/workers/rpc.py +++ b/packages/runtime-sdk/src/workers/rpc.py @@ -6,10 +6,17 @@ from pyodide.ffi import JsProxy, create_proxy, to_js from .blob import Blob, File +from .fetch import fetch from .formdata import FormData from .request import Request from .response import Response -from .utils import _is_iterable, _is_js_instance, jsnull +from .utils import ( + _JS_PASSTHROUGH_TYPES, + _get_js_constructor_name, + _is_iterable, + _is_js_instance, + jsnull, +) def _python_from_rpc_default_converter(value, convert, cache): @@ -108,9 +115,6 @@ def python_from_rpc(obj: "JsProxy"): def _raise_on_disabled_type(value): - # Lazy import: _BindingWrapper is defined in _workers.py - from ._workers import _BindingWrapper - if isinstance(value, _BindingWrapper): return @@ -137,9 +141,6 @@ def _raise_on_disabled_type(value): def _python_to_rpc_default_converter(obj, convert, cache): - # Lazy import: _BindingWrapper is defined in _workers.py - from ._workers import _BindingWrapper - if obj is None: return jsnull @@ -184,10 +185,6 @@ def python_to_rpc(value) -> JsProxy: This method is used for Workers RPC in Python to convert Python objects to JavaScript. As such it does not support serializing all Python object types. """ - # Lazy import: _BindingWrapper is defined in _workers.py - # TODO: refactor more to avoid circular imports - from ._workers import _BindingWrapper - if value is None: return jsnull @@ -206,3 +203,218 @@ def python_to_rpc(value) -> JsProxy: ) return result + + +# --------------------------------------------------------------------------- +# Binding wrappers +# --------------------------------------------------------------------------- + + +def _idempotent_new(cls, obj): + """Set __new__ on a class to this so cls is idempotent: + + >>> a = A(x) + >>> b = A(a) + >>> assert a is b + + For this to work, start the __init__ function with: + + if obj is self: + return + + to prevent double-init. + """ + if isinstance(obj, cls): + return obj + return object.__new__(cls) + + +class _BindingWrapper: + __new__ = _idempotent_new + + def __init__(self, binding): + if binding is self: + return + self._binding = binding + + @property + def _real_name(self): + js_name = _get_js_constructor_name(self._binding) + if not js_name: + # Should not happen, but just in case + return type(self).__name__ + return js_name + + def _should_wrap_nested_attribute(self, jsobj) -> bool: + if not isinstance(jsobj, JsProxy): + return False + + # TODO: This allowlist approach is a workaround. The long-term fix is to + # add dedicated Python wrappers for these types in python_from_rpc so they + # never reach _BindingWrapper in the first place. + js_type = _get_js_constructor_name(jsobj) + return js_type and js_type not in _JS_PASSTHROUGH_TYPES + + def _convert_result(self, result): + converted = python_from_rpc(result) + + # After python_from_rpc, some objects may still be JsProxy objects. + # We need to wrap them with _BindingWrapper (or a subclass of it) again + # to ensure that accessing attributes on them will be properly converted. + if self._should_wrap_nested_attribute(converted): + return self.__class__(converted) + if isinstance(converted, list): + return [ + self.__class__(item) + if self._should_wrap_nested_attribute(item) + else item + for item in converted + ] + return converted + + @staticmethod + def _convert_args(args, kwargs): + js_args = [python_to_rpc(arg) for arg in args] + js_kwargs = {k: python_to_rpc(v) for k, v in kwargs.items()} + return js_args, js_kwargs + + def _getattr_helper(self, name): + attr = getattr(self._binding, name) + + if not callable(attr): + return self._convert_result(attr) + + def wrapper(*args, **kwargs): + js_args, js_kwargs = self._convert_args(args, kwargs) + result = attr(*js_args, **js_kwargs) + if hasattr(result, "then") and callable(result.then): + + async def await_and_convert(): + return self._convert_result(await result) + + return await_and_convert() + return self._convert_result(result) + + return wrapper + + def __getattr__(self, name): + result = self._getattr_helper(name) + setattr(self, name, result) + return result + + def __getitem__(self, key): + if isinstance(key, int): + return self._convert_result(self._binding[key]) + return self._convert_result(getattr(self._binding, key)) + + def __iter__(self): + binding = self._binding + if not hasattr(binding, "__iter__"): + raise TypeError(f"'{self._real_name}' object is not iterable") + for item in binding: + yield self._convert_result(item) + + def __len__(self): + binding = self._binding + if not hasattr(binding, "length"): + raise TypeError(f"'{self._real_name}' object has no len()") + return binding.length + + +class _FetcherWrapper(_BindingWrapper): + def fetch(self, *args, **kwargs): + return fetch(*args, fetcher=self._binding.fetch, **kwargs) + + +class _DurableObjectNamespaceWrapper: + def __init__(self, binding): + self._binding = binding + + def __getattr__(self, name): + return getattr(self._binding, name) + + def get(self, *args, **kwargs): + return _FetcherWrapper(self._binding.get(*args, **kwargs)) + + def getByName(self, *args, **kwargs): + return _FetcherWrapper(self._binding.getByName(*args, **kwargs)) + + def jurisdiction(self, *args, **kwargs): + return _DurableObjectNamespaceWrapper( + self._binding.jurisdiction(*args, **kwargs) + ) + + +class _WorkflowInstanceWrapper(_BindingWrapper): + # status/pause/resume/restart/terminate share their JS names and are handled by + # the _BindingWrapper, which already converts arguments and results. + # Only send_event needs the snake_case -> camelCase mapping for backward compatibility + + async def send_event(self, *args, **kwargs): + js_args, js_kwargs = self._convert_args(args, kwargs) + return self._convert_result( + await self._binding.sendEvent(*js_args, **js_kwargs) + ) + + +class _WorkflowBindingWrapper(_BindingWrapper): + async def get(self, *args, **kwargs): + js_args, js_kwargs = self._convert_args(args, kwargs) + return _WorkflowInstanceWrapper(await self._binding.get(*js_args, **js_kwargs)) + + async def create(self, *args, **kwargs): + js_args, js_kwargs = self._convert_args(args, kwargs) + return _WorkflowInstanceWrapper( + await self._binding.create(*js_args, **js_kwargs) + ) + + async def create_batch(self, *args, **kwargs): + js_args, js_kwargs = self._convert_args(args, kwargs) + return [ + _WorkflowInstanceWrapper(w) + for w in await self._binding.createBatch(*js_args, **js_kwargs) + ] + + +class _EnvWrapper: + _BINDING_TYPES = { + "KvNamespace", + "R2Bucket", + "D1Database", + "WorkerQueue", + "Ai", + "VectorizeIndexImpl", + "AnalyticsEngineDataset", + "LocalAnalyticsEngineDataset", + "ImagesBindingImpl", + "HostedImagesBindingImpl", + "Ratelimit", + } + + __new__ = _idempotent_new + + def __init__(self, env): + if env is self: + return + self._env = env + + def _getattr_helper(self, name): + binding = getattr(self._env, name) + if _is_js_instance(binding, "Fetcher"): + return _FetcherWrapper(binding) + + if _is_js_instance(binding, "DurableObjectNamespace"): + return _DurableObjectNamespaceWrapper(binding) + + if _is_js_instance(binding, "WorkflowImpl"): + return _WorkflowBindingWrapper(binding) + + if _is_js_instance(binding, self._BINDING_TYPES): + return _BindingWrapper(binding) + + return binding + + def __getattr__(self, name): + result = self._getattr_helper(name) + setattr(self, name, result) + return result