Add configurations for subscriber and publisher

faststream/__init__.py

@@ -1,5 +1,6 @@
 """A Python framework for building services interacting with Apache Kafka, RabbitMQ, NATS and Redis."""
 
+from faststream._internal.configs.settings import Settings
 from faststream._internal.testing.app import TestApp
 from faststream._internal.utils import apply_types
 from faststream.annotations import ContextRepo, Logger
@@ -27,6 +28,7 @@
     "PublishCommand",
     "PublishType",
     "Response",
+    "Settings",
     "SourceType",
     "StreamMessage",
     "TestApp",

faststream/_internal/broker/broker.py

@@ -105,11 +105,20 @@ def _setup_logger(self) -> None:
 
         self.config.logger._setup(self.config.fd_config.context)
 
+    def setup_logger(self) -> None:
+        self._setup_logger()
+
+    def resolve_settings(self) -> None:
+        for sub in self.subscribers:
+            sub.resolve_settings()
+
+        for pub in self.publishers:
+            pub.resolve_settings()
+
     async def connect(self) -> ConnectionType:
         """Connect to a remote server."""
         if self._connection is None:
             self._connection = await self._connect()
-            self._setup_logger()
 
         return self._connection
 

faststream/_internal/configs/__init__.py

@@ -1,5 +1,6 @@
 from .broker import BrokerConfig, BrokerConfigType, ConfigComposition
 from .endpoint import PublisherUsecaseConfig, SubscriberUsecaseConfig
+from .settings import Settings, make_settings_container
 from .specification import (
     PublisherSpecificationConfig,
     SpecificationConfig as SubscriberSpecificationConfig,
@@ -11,6 +12,8 @@
     "ConfigComposition",
     "PublisherSpecificationConfig",
     "PublisherUsecaseConfig",
+    "Settings",
     "SubscriberSpecificationConfig",
     "SubscriberUsecaseConfig",
+    "make_settings_container",
 )

faststream/_internal/configs/broker.py

@@ -8,6 +8,8 @@
 from faststream._internal.logger import LoggerState
 from faststream._internal.producer import ProducerProto, ProducerUnset
 
+from .settings import SettingsContainer, make_settings_container
+
 if TYPE_CHECKING:
     from fast_depends.dependencies import Dependant
 
@@ -19,6 +21,8 @@ class BrokerConfig:
     prefix: str = ""
     include_in_schema: bool | None = True
 
+    settings: SettingsContainer = field(default_factory=make_settings_container)
+
     broker_middlewares: Sequence["BrokerMiddleware[Any]"] = ()
     broker_parser: Optional["CustomCallable"] = None
     broker_decoder: Optional["CustomCallable"] = None

faststream/_internal/configs/settings.py

@@ -0,0 +1,64 @@
+from collections.abc import Mapping, MutableMapping
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+
+@dataclass(slots=True)
+class Settings:
+    key: str
+
+
+class SettingsContainer(Protocol):
+    def resolve(self, item: Any) -> Any:
+        pass
+
+
+class RealSettingsContainer(SettingsContainer):
+    def __init__(self, settings: Mapping[str, Any]) -> None:
+        self._items = settings
+
+    def resolve(self, item: Any) -> Any:
+        if isinstance(item, Settings):
+            return self._items[item.key]
+        self._resolve_child(item)
+        return item
+
+    def _resolve_child(
+        self,
+        item: Any,
+        seen: set[Any] | None = None,
+    ) -> None:
+        if seen is None:
+            seen = set()
+
+        if id(item) in seen:
+            return
+
+        seen.add(id(item))
+
+        if isinstance(item, MutableMapping):
+            for key, value in item.items():
+                if isinstance(value, Settings):
+                    item[key] = self._items[value.key]
+                self._resolve_child(value, seen)
+
+        else:
+            for attr_name in dir(item):
+                if not attr_name.startswith("_"):
+                    attr = getattr(item, attr_name)
+                    if isinstance(attr, Settings):
+                        setattr(item, attr_name, self._items[attr.key])
+                    self._resolve_child(attr, seen)
+
+
+class FakeSettingsContainer(SettingsContainer):
+    def resolve(self, item: Any) -> Any:
+        return item
+
+
+def make_settings_container(
+    settings: Mapping[str, Any] | None = None,
+) -> SettingsContainer:
+    if not settings:
+        return FakeSettingsContainer()
+    return RealSettingsContainer(settings)

faststream/rabbit/broker/broker.py

@@ -1,5 +1,5 @@
 import logging
-from collections.abc import Iterable, Sequence
+from collections.abc import Iterable, Mapping, Sequence
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -16,6 +16,7 @@
 
 from faststream.__about__ import SERVICE_NAME
 from faststream._internal.broker import BrokerUsecase
+from faststream._internal.configs import make_settings_container
 from faststream._internal.constants import EMPTY
 from faststream._internal.context.repository import ContextRepo
 from faststream._internal.di import FastDependsConfig
@@ -108,6 +109,7 @@ def __init__(
         # FastDepends args
         apply_types: bool = True,
         serializer: Optional["SerializerProto"] = EMPTY,
+        settings: Mapping[str, Any] | None = None,
         provider: Optional["Provider"] = None,
         context: Optional["ContextRepo"] = None,
     ) -> None:
@@ -141,6 +143,7 @@ def __init__(
             log_level: Service messages log level.
             apply_types: Whether to use FastDepends or not.
             serializer: FastDepends-compatible serializer to validate incoming messages.
+            settings: Container for configuration publisher and subscriber.
             provider: Provider for FastDepends.
             context: Context for FastDepends.
         """
@@ -185,6 +188,7 @@ def __init__(
             # Basic args
             routers=routers,
             config=RabbitBrokerConfig(
+                settings=make_settings_container(settings),
                 channel_manager=cm,
                 producer=producer,
                 declarer=declarer,
@@ -274,6 +278,9 @@ async def close(
     async def start(self) -> None:
         """Connect broker to RabbitMQ and startup all subscribers."""
         await self.connect()
+        # can merge it into one operation, something like br.initialize or br.initialize_pipe
+        self.resolve_settings()
+        self.setup_logger()
         await self.declare_queue(RABBIT_REPLY)
         await super().start()
 

faststream/rabbit/broker/registrator.py

@@ -5,6 +5,7 @@
 from typing_extensions import deprecated, override
 
 from faststream._internal.broker.registrator import Registrator
+from faststream._internal.configs import Settings
 from faststream._internal.constants import EMPTY
 from faststream.exceptions import SetupError
 from faststream.middlewares import AckPolicy
@@ -38,11 +39,11 @@ class RabbitRegistrator(Registrator[IncomingMessage, RabbitBrokerConfig]):
     @override
     def subscriber(  # type: ignore[override]
         self,
-        queue: Union[str, "RabbitQueue"],
-        exchange: Union[str, "RabbitExchange", None] = None,
+        queue: Union[str, "RabbitQueue", Settings],
+        exchange: Union[str, "RabbitExchange", Settings, None] = None,
         *,
-        channel: Optional["Channel"] = None,
-        consume_args: dict[str, Any] | None = None,
+        channel: Optional["Channel"] | Settings = None,
+        consume_args: dict[str, Any] | Settings | None = None,
         no_ack: Annotated[
             bool,
             deprecated(
@@ -120,16 +121,16 @@ def subscriber(  # type: ignore[override]
     @override
     def publisher(  # type: ignore[override]
         self,
-        queue: Union["RabbitQueue", str] = "",
-        exchange: Union["RabbitExchange", str, None] = None,
+        queue: Union["RabbitQueue", str, Settings] = "",
+        exchange: Union["RabbitExchange", str, Settings, None] = None,
         *,
-        routing_key: str = "",
-        mandatory: bool = True,
-        immediate: bool = False,
+        routing_key: str | Settings = "",
+        mandatory: bool | Settings = True,
+        immediate: bool | Settings = False,
+        persist: bool | Settings = False,
+        reply_to: str | Settings | None = None,
+        priority: int | Settings | None = None,
         timeout: "TimeoutType" = None,
-        persist: bool = False,
-        reply_to: str | None = None,
-        priority: int | None = None,
         persistent: bool = True,
         # specific
         middlewares: Annotated[
@@ -145,12 +146,12 @@ def publisher(  # type: ignore[override]
         schema: Any | None = None,
         include_in_schema: bool = True,
         # message args
-        headers: Optional["HeadersType"] = None,
-        content_type: str | None = None,
-        content_encoding: str | None = None,
-        expiration: Optional["DateType"] = None,
-        message_type: str | None = None,
-        user_id: str | None = None,
+        headers: Optional["HeadersType"] | Settings = None,
+        content_type: str | Settings | None = None,
+        content_encoding: str | Settings | None = None,
+        expiration: Optional["DateType"] | Settings = None,
+        message_type: str | Settings | None = None,
+        user_id: str | Settings | None = None,
     ) -> "RabbitPublisher":
         """Creates long-living and AsyncAPI-documented publisher object.
 

faststream/rabbit/publisher/factory.py

@@ -1,5 +1,7 @@
 from collections.abc import Sequence
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Union
+
+from faststream._internal.configs.settings import Settings
 
 from .config import RabbitPublisherConfig, RabbitPublisherSpecificationConfig
 from .specification import RabbitPublisherSpecification
@@ -15,10 +17,10 @@
 
 def create_publisher(
     *,
-    routing_key: str,
-    queue: "RabbitQueue",
-    exchange: "RabbitExchange",
-    message_kwargs: "PublishKwargs",
+    routing_key: str | Settings,
+    queue: Union["RabbitQueue", Settings],
+    exchange: Union["RabbitExchange", Settings],
+    message_kwargs: Union["PublishKwargs", Settings],
     # Broker args
     config: "RabbitBrokerConfig",
     # Publisher args

faststream/rabbit/publisher/usecase.py

@@ -1,5 +1,5 @@
 from collections.abc import Iterable
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 from typing_extensions import Unpack, override
 
@@ -79,7 +79,17 @@ def routing(
 
         return routing_key
 
+    def resolve_settings(self) -> None:
+        resolver: Callable[..., Any] = self._outer_config.settings.resolve
+        self.routing_key = resolver(self.routing_key)
+        self.queue = RabbitQueue.validate(resolver(self.queue))
+        self.exchange = RabbitExchange.validate(resolver(self.exchange))
+        self.headers = resolver(self.headers)
+        self.reply_to = resolver(self.reply_to)
+        self.timeout = resolver(self.timeout)
+
     async def start(self) -> None:
+        """Starts the consumer for the RabbitMQ queue."""
         if self.exchange is not None:
             await self._outer_config.declarer.declare_exchange(self.exchange)
         return await super().start()

faststream/rabbit/schemas/queue.py

@@ -79,6 +79,15 @@ def add_prefix(self, prefix: str) -> "RabbitQueue":
 
         return new_q
 
+    def set_routing(self):
+        re, routing_key = compile_path(
+            self.routing_key,
+            replace_symbol="*",
+            patch_regex=lambda x: x.replace(r"\#", ".+"),
+        )
+        self.path_regex = re
+        self.routing_key = routing_key
+
     @overload
     def __init__(
         self,
@@ -164,11 +173,7 @@ def __init__(
         :param bind_arguments: Queue-exchange binding options.
         :param routing_key: Explicit binding routing key. Uses name if not present.
         """
-        re, routing_key = compile_path(
-            routing_key,
-            replace_symbol="*",
-            patch_regex=lambda x: x.replace(r"\#", ".+"),
-        )
+        re = None
 
         if queue_type is QueueType.QUORUM or queue_type is QueueType.STREAM:
             if durable is EMPTY:

faststream/rabbit/subscriber/factory.py

@@ -1,6 +1,7 @@
 import warnings
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any, Optional, Union
 
+from faststream._internal.configs.settings import Settings
 from faststream._internal.constants import EMPTY
 from faststream._internal.endpoint.subscriber.call_item import CallsCollection
 from faststream.exceptions import SetupError
@@ -20,10 +21,10 @@
 
 def create_subscriber(
     *,
-    queue: "RabbitQueue",
-    exchange: "RabbitExchange",
-    consume_args: dict[str, Any] | None,
-    channel: Optional["Channel"],
+    queue: Union["RabbitQueue", Settings],
+    exchange: Union["RabbitExchange", Settings],
+    consume_args: dict[str, Any] | Settings | None,
+    channel: Optional["Channel"] | Settings,
     # Subscriber args
     no_reply: bool,
     ack_policy: "AckPolicy",