diff --git a/dapr/actor/client/proxy.py b/dapr/actor/client/proxy.py index 699ef7447..1a8909ecc 100644 --- a/dapr/actor/client/proxy.py +++ b/dapr/actor/client/proxy.py @@ -14,16 +14,18 @@ """ from abc import ABC, abstractmethod -from typing import Any, Callable, Dict, Optional, Type +from typing import Any, Callable, Dict, Optional, Type, Union from dapr.actor.actor_interface import ActorInterface from dapr.actor.id import ActorId from dapr.actor.runtime._type_utils import get_dispatchable_attrs_from_interface from dapr.clients import DaprActorClientBase, DaprActorHttpClient from dapr.clients.retry import RetryPolicy -from dapr.conf import settings +from dapr.conf import _FromSettings from dapr.serializers import DefaultJSONSerializer, Serializer +_HTTP_TIMEOUT_FROM_SETTINGS = _FromSettings('DAPR_HTTP_TIMEOUT_SECONDS') + # Actor factory Callable type hint. ACTOR_FACTORY_CALLBACK = Callable[[ActorInterface, str, str], 'ActorProxy'] @@ -49,9 +51,11 @@ class ActorProxyFactory(ActorFactoryBase): def __init__( self, message_serializer=DefaultJSONSerializer(), - http_timeout_seconds: int = settings.DAPR_HTTP_TIMEOUT_SECONDS, + http_timeout_seconds: Union[int, None, _FromSettings] = _HTTP_TIMEOUT_FROM_SETTINGS, retry_policy: Optional[RetryPolicy] = None, ): + if isinstance(http_timeout_seconds, _FromSettings): + http_timeout_seconds = int(http_timeout_seconds.resolve()) # TODO: support serializer for state store later self._dapr_client = DaprActorHttpClient( message_serializer, timeout=http_timeout_seconds, retry_policy=retry_policy diff --git a/dapr/actor/runtime/grpc_host.py b/dapr/actor/runtime/grpc_host.py index 93cbcdca4..be3d1f24e 100644 --- a/dapr/actor/runtime/grpc_host.py +++ b/dapr/actor/runtime/grpc_host.py @@ -16,7 +16,7 @@ import asyncio import logging import os -from typing import Awaitable, Callable, Optional, Protocol, Set, Type, TypeVar +from typing import Awaitable, Callable, Optional, Protocol, Set, Type, TypeVar, Union import grpc.aio # type: ignore from grpc import StatusCode # type: ignore[attr-defined] @@ -43,7 +43,7 @@ from dapr.actor.runtime.runtime import ActorRuntime from dapr.clients.grpc._channel import create_aio_channel from dapr.clients.grpc.dapr_actor_grpc_client import DaprActorGrpcClient -from dapr.conf import settings +from dapr.conf import _FromSettings, settings from dapr.proto import api_service_v1, api_v1 from dapr.serializers import DefaultJSONSerializer, Serializer @@ -60,6 +60,7 @@ _RECONNECT_DELAY_SECONDS = 1.0 _DRAIN_TIMEOUT_SECONDS = 5.0 _HANDSHAKE_TIMEOUT_SECONDS = 60.0 +_HTTP_TIMEOUT_FROM_SETTINGS = _FromSettings('DAPR_HTTP_TIMEOUT_SECONDS') _GRPC_AIO_EOF = grpc.aio.EOF # type: ignore[attr-defined] @@ -144,7 +145,7 @@ class ActorGrpcHost: def __init__( self, address: Optional[str] = None, - timeout_seconds: int = settings.DAPR_HTTP_TIMEOUT_SECONDS, + timeout_seconds: Union[int, None, _FromSettings] = _HTTP_TIMEOUT_FROM_SETTINGS, app_port: Optional[int] = None, ): """Creates the host. @@ -153,14 +154,17 @@ def __init__( address (str, optional): daprd gRPC address. Defaults to the same resolution as the Dapr clients (``DAPR_GRPC_ENDPOINT``, ``DAPR_RUNTIME_HOST``/``DAPR_GRPC_PORT``). - timeout_seconds (int): per-call timeout for the actors' outbound - operations (state, reminders, timers, invocations). + timeout_seconds (int, optional): per-call timeout for the actors' outbound + operations (state, reminders, timers, invocations). Defaults to + ``DAPR_HTTP_TIMEOUT_SECONDS``, read when the host is created. app_port (int, optional): port for the empty gRPC app-channel listener the host starts so daprd accepts the stream. Defaults to the ``APP_PORT`` env var (set by ``dapr run``). Pass ``0`` to disable it when you run your own gRPC app server. """ self._address = address + if isinstance(timeout_seconds, _FromSettings): + timeout_seconds = int(timeout_seconds.resolve()) self._timeout_seconds = timeout_seconds self._app_port = self._resolve_app_port(app_port) self._channel: Optional[grpc.aio.Channel] = None diff --git a/dapr/actor/runtime/runtime.py b/dapr/actor/runtime/runtime.py index 6f5462748..d87a24c1d 100644 --- a/dapr/actor/runtime/runtime.py +++ b/dapr/actor/runtime/runtime.py @@ -14,7 +14,7 @@ """ import asyncio -from typing import Callable, Dict, List, Optional, Type +from typing import Callable, Dict, List, Optional, Type, Union from dapr.actor.id import ActorId from dapr.actor.runtime._type_information import ActorTypeInformation @@ -25,9 +25,11 @@ from dapr.clients.base import DaprActorClientBase from dapr.clients.http.dapr_actor_http_client import DaprActorHttpClient from dapr.common.reentrancy_context import reentrancy_ctx -from dapr.conf import settings +from dapr.conf import _FromSettings from dapr.serializers import DefaultJSONSerializer, Serializer +_HTTP_TIMEOUT_FROM_SETTINGS = _FromSettings('DAPR_HTTP_TIMEOUT_SECONDS') + class ActorRuntime: """The class that creates instances of :class:`Actor` and @@ -45,7 +47,7 @@ async def register_actor( actor: Type[Actor], message_serializer: Serializer = DefaultJSONSerializer(), state_serializer: Serializer = DefaultJSONSerializer(), - http_timeout_seconds: int = settings.DAPR_HTTP_TIMEOUT_SECONDS, + http_timeout_seconds: Union[int, None, _FromSettings] = _HTTP_TIMEOUT_FROM_SETTINGS, actor_factory: Optional[Callable[['ActorRuntimeContext', ActorId], 'Actor']] = None, actor_client: Optional[DaprActorClientBase] = None, ) -> None: @@ -56,13 +58,16 @@ async def register_actor( message_serializer (:class:`Serializer`): A serializer that serializes message between actors. state_serializer (:class:`Serializer`): Serializer that serializes state values. - http_timeout_seconds (:int:): a configurable timeout value + http_timeout_seconds (int, optional): timeout for the default HTTP actor client. + Defaults to ``DAPR_HTTP_TIMEOUT_SECONDS``, read when the actor is registered. actor_factory (Callable, optional): factory to create Actor objects. actor_client (:class:`DaprActorClientBase`, optional): the client used for the actor's outbound operations (state, reminders, timers, invocations). Defaults to the HTTP client; :class:`ActorGrpcHost` supplies a gRPC one. """ type_info = ActorTypeInformation.create(actor) + if isinstance(http_timeout_seconds, _FromSettings): + http_timeout_seconds = int(http_timeout_seconds.resolve()) if actor_client is None: actor_client = DaprActorHttpClient(message_serializer, timeout=http_timeout_seconds) ctx = ActorRuntimeContext( diff --git a/dapr/clients/grpc/dapr_actor_grpc_client.py b/dapr/clients/grpc/dapr_actor_grpc_client.py index c01ee0187..85e310fb2 100644 --- a/dapr/clients/grpc/dapr_actor_grpc_client.py +++ b/dapr/clients/grpc/dapr_actor_grpc_client.py @@ -50,7 +50,7 @@ class DaprActorGrpcClient(DaprActorClientBase): def __init__( self, - timeout: int = 60, + timeout: Optional[int] = 60, address: Optional[str] = None, channel: Optional[grpc.aio.Channel] = None, message_serializer: Serializer = DefaultJSONSerializer(), @@ -59,7 +59,7 @@ def __init__( """Creates the gRPC actor client. Args: - timeout (int): per-call timeout in seconds. + timeout (int, optional): per-call timeout in seconds; None means no timeout. address (str, optional): Dapr runtime gRPC address; defaults to the same resolution as the Dapr clients (DAPR_GRPC_ENDPOINT et al). channel (grpc.aio.Channel, optional): externally owned channel to diff --git a/dapr/clients/http/dapr_actor_http_client.py b/dapr/clients/http/dapr_actor_http_client.py index bed3c97e4..d35f85441 100644 --- a/dapr/clients/http/dapr_actor_http_client.py +++ b/dapr/clients/http/dapr_actor_http_client.py @@ -32,7 +32,7 @@ class DaprActorHttpClient(DaprActorClientBase): def __init__( self, message_serializer: 'Serializer', - timeout: int = 60, + timeout: Optional[int] = 60, headers_callback: Optional[Callable[[], Dict[str, str]]] = None, retry_policy: Optional[RetryPolicy] = None, ): diff --git a/dapr/clients/retry.py b/dapr/clients/retry.py index 58283b376..81bb92a7e 100644 --- a/dapr/clients/retry.py +++ b/dapr/clients/retry.py @@ -15,18 +15,21 @@ import asyncio import time -from typing import Callable, List, Optional +from typing import Callable, List, Union from grpc import RpcError, StatusCode # type: ignore -from dapr.conf import settings +from dapr.conf import _FromSettings + +_MAX_RETRIES_FROM_SETTINGS = _FromSettings('DAPR_API_MAX_RETRIES') class RetryPolicy: """RetryPolicy holds the retry policy configuration for a gRPC client. Args: - max_attempts (int): The maximum number of retry attempts. + max_attempts (int, optional): The maximum number of retry attempts. Defaults to + ``DAPR_API_MAX_RETRIES``, read when the policy is constructed. initial_backoff (int): The initial backoff duration. max_backoff (int): The maximum backoff duration. backoff_multiplier (float): The backoff multiplier. @@ -36,7 +39,7 @@ class RetryPolicy: def __init__( self, - max_attempts: Optional[int] = settings.DAPR_API_MAX_RETRIES, + max_attempts: Union[int, None, _FromSettings] = _MAX_RETRIES_FROM_SETTINGS, initial_backoff: int = 1, max_backoff: int = 20, backoff_multiplier: float = 1.5, @@ -46,6 +49,9 @@ def __init__( StatusCode.DEADLINE_EXCEEDED, ], ): + if isinstance(max_attempts, _FromSettings): + max_attempts = int(max_attempts.resolve()) + if max_attempts is None or max_attempts < -1: raise ValueError('max_attempts must be greater than or equal to -1') self.max_attempts = max_attempts diff --git a/dapr/conf/__init__.py b/dapr/conf/__init__.py index 7fbe5f2f7..d909f0947 100644 --- a/dapr/conf/__init__.py +++ b/dapr/conf/__init__.py @@ -14,27 +14,58 @@ """ import os +from typing import Any from dapr.conf import global_settings class Settings: - def __init__(self): - for setting in dir(global_settings): - default_value = getattr(global_settings, setting) - env_variable = os.environ.get(setting) - if env_variable: - val = ( - type(default_value)(env_variable) if default_value is not None else env_variable - ) - setattr(self, setting, val) - else: - setattr(self, setting, default_value) + """Settings resolved from the environment each time they are read. + + A value assigned on the instance (``settings.DAPR_HTTP_PORT = 3500``) takes + precedence; otherwise the ``DAPR_*`` environment variable is used when set, + falling back to the default in :mod:`dapr.conf.global_settings`. + """ def __getattr__(self, name): - if name not in dir(global_settings): + # Only reached when the attribute was not assigned on the instance. + if name.startswith('__') or not hasattr(global_settings, name): raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'") - return getattr(self, name) + default_value = getattr(global_settings, name) + env_variable = os.environ.get(name) + if not env_variable: + return default_value + if default_value is None: + return env_variable + try: + return type(default_value)(env_variable) + except ValueError as error: + raise ValueError(f'Invalid value for {name}: {env_variable!r}') from error + + def __dir__(self): + return sorted( + {*super().__dir__(), *(n for n in dir(global_settings) if not n.startswith('__'))} + ) settings = Settings() + + +class _FromSettings: + """Sentinel default that reads a setting when the call is made, not at import. + + Lets a parameter tell "omitted" apart from an explicit ``None``. + + Args: + name: Name of the setting in :mod:`dapr.conf.global_settings`. + """ + + def __init__(self, name: str) -> None: + self.name = name + + def __repr__(self) -> str: + return self.name + + def resolve(self) -> Any: + """Returns the current value of the setting.""" + return getattr(settings, self.name) diff --git a/tests/conf/settings_test.py b/tests/conf/settings_test.py new file mode 100644 index 000000000..25b603abf --- /dev/null +++ b/tests/conf/settings_test.py @@ -0,0 +1,161 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2026 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +import os +import unittest +from unittest import mock + +from dapr.actor.client.proxy import ActorProxyFactory +from dapr.actor.runtime.config import ActorRuntimeConfig +from dapr.actor.runtime.grpc_host import ActorGrpcHost +from dapr.actor.runtime.runtime import ActorRuntime +from dapr.clients.grpc._channel import resolve_grpc_endpoint +from dapr.clients.retry import RetryPolicy +from dapr.conf import Settings, global_settings, settings +from tests.actor.fake_actor_classes import FakeSimpleActor + + +class SettingsReadEnvAtAccessTests(unittest.TestCase): + def setUp(self): + patcher = mock.patch.dict(os.environ) + patcher.start() + self.addCleanup(patcher.stop) + names = ( + 'DAPR_GRPC_ENDPOINT', + 'DAPR_GRPC_PORT', + 'DAPR_API_MAX_RETRIES', + 'DAPR_HTTP_ENDPOINT', + 'DAPR_HTTP_TIMEOUT_SECONDS', + ) + for name in names: + os.environ.pop(name, None) + # Other tests assign on the shared instance; hide those so env is visible. + assigned = {n: settings.__dict__.pop(n) for n in names if n in settings.__dict__} + self.addCleanup(settings.__dict__.update, assigned) + + def test_env_set_after_import_is_seen(self): + self.assertIsNone(settings.DAPR_GRPC_ENDPOINT) + + os.environ['DAPR_GRPC_ENDPOINT'] = 'grpc-example.example.com:443' + + self.assertEqual('grpc-example.example.com:443', settings.DAPR_GRPC_ENDPOINT) + + def test_env_change_and_removal_is_seen(self): + os.environ['DAPR_GRPC_PORT'] = '1234' + self.assertEqual(1234, settings.DAPR_GRPC_PORT) + + os.environ['DAPR_GRPC_PORT'] = '5678' + self.assertEqual(5678, settings.DAPR_GRPC_PORT) + + del os.environ['DAPR_GRPC_PORT'] + self.assertEqual(global_settings.DAPR_GRPC_PORT, settings.DAPR_GRPC_PORT) + + def test_empty_env_uses_default(self): + os.environ['DAPR_GRPC_PORT'] = '' + self.assertEqual(global_settings.DAPR_GRPC_PORT, settings.DAPR_GRPC_PORT) + + def test_explicit_assignment_wins_over_env(self): + local = Settings() + os.environ['DAPR_GRPC_PORT'] = '1234' + local.DAPR_GRPC_PORT = 4321 + + self.assertEqual(4321, local.DAPR_GRPC_PORT) + os.environ['DAPR_GRPC_PORT'] = '9999' + self.assertEqual(4321, local.DAPR_GRPC_PORT) + + del local.DAPR_GRPC_PORT + self.assertEqual(9999, local.DAPR_GRPC_PORT) + + def test_unknown_setting_raises(self): + with self.assertRaises(AttributeError): + settings.DAPR_DOES_NOT_EXIST # noqa: B018 + + def test_grpc_endpoint_resolution_uses_env_set_after_import(self): + os.environ['DAPR_GRPC_ENDPOINT'] = 'grpc-example.example.com:443' + + self.assertEqual( + 'dns:grpc-example.example.com:443', + resolve_grpc_endpoint(None).endpoint, + ) + + def test_retry_policy_default_reads_env_at_construction(self): + os.environ['DAPR_API_MAX_RETRIES'] = '7' + + self.assertEqual(7, RetryPolicy().max_attempts) + self.assertEqual(3, RetryPolicy(max_attempts=3).max_attempts) + + def test_retry_policy_zero_from_env(self): + os.environ['DAPR_API_MAX_RETRIES'] = '0' + self.assertEqual(0, RetryPolicy().max_attempts) + + def test_invalid_env_value_names_the_setting(self): + os.environ['DAPR_GRPC_PORT'] = 'abc' + with self.assertRaisesRegex(ValueError, 'DAPR_GRPC_PORT'): + settings.DAPR_GRPC_PORT # noqa: B018 + + def test_dir_lists_settings(self): + self.assertIn('DAPR_GRPC_ENDPOINT', dir(settings)) + + def test_http_endpoint_from_env(self): + os.environ['DAPR_HTTP_ENDPOINT'] = 'http://example.com:3500' + self.assertEqual('http://example.com:3500', settings.DAPR_HTTP_ENDPOINT) + + def test_actor_timeout_defaults_read_env_at_call(self): + os.environ['DAPR_HTTP_TIMEOUT_SECONDS'] = '7' + with mock.patch('dapr.actor.client.proxy.DaprActorHttpClient') as client: + ActorProxyFactory() + self.assertEqual(7, client.call_args.kwargs['timeout']) + self.assertEqual(7, ActorGrpcHost(app_port=0)._timeout_seconds) + self.assertEqual(3, ActorGrpcHost(timeout_seconds=3, app_port=0)._timeout_seconds) + + def test_actor_timeout_none_is_passed_through(self): + os.environ['DAPR_HTTP_TIMEOUT_SECONDS'] = '7' + with mock.patch('dapr.actor.client.proxy.DaprActorHttpClient') as client: + ActorProxyFactory(http_timeout_seconds=None) + self.assertIsNone(client.call_args.kwargs['timeout']) + host = ActorGrpcHost(timeout_seconds=None, app_port=0) + self.assertIsNone(host._timeout_seconds) + + +class RegisterActorTimeoutTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + env_patcher = mock.patch.dict(os.environ) + env_patcher.start() + self.addCleanup(env_patcher.stop) + managers_patcher = mock.patch.object(ActorRuntime, '_actor_managers', {}) + managers_patcher.start() + self.addCleanup(managers_patcher.stop) + config_patcher = mock.patch.object(ActorRuntime, '_actor_config', ActorRuntimeConfig()) + config_patcher.start() + self.addCleanup(config_patcher.stop) + names = ('DAPR_HTTP_TIMEOUT_SECONDS',) + assigned = {n: settings.__dict__.pop(n) for n in names if n in settings.__dict__} + self.addCleanup(settings.__dict__.update, assigned) + + async def test_register_actor_default_reads_env_at_call(self): + os.environ['DAPR_HTTP_TIMEOUT_SECONDS'] = '7' + with mock.patch('dapr.actor.runtime.runtime.DaprActorHttpClient') as client: + await ActorRuntime.register_actor(FakeSimpleActor) + self.assertEqual(7, client.call_args.kwargs['timeout']) + + async def test_register_actor_explicit_timeout_wins(self): + os.environ['DAPR_HTTP_TIMEOUT_SECONDS'] = '7' + with mock.patch('dapr.actor.runtime.runtime.DaprActorHttpClient') as client: + await ActorRuntime.register_actor(FakeSimpleActor, http_timeout_seconds=3) + self.assertEqual(3, client.call_args.kwargs['timeout']) + + +if __name__ == '__main__': + unittest.main()