Move service_name to Application from wire

Author: WiZeYAR

src/asyncapi_python/contrib/codec/__init__.py

@@ -2,5 +2,4 @@
 
 from .registry import CodecRegistry
 
-
 __all__ = ["CodecRegistry"]

src/asyncapi_python/contrib/codec/json.py

@@ -1,6 +1,6 @@
 import json
-from typing import Type, ClassVar
 from types import ModuleType
+from typing import ClassVar, Type
 
 from pydantic import BaseModel, ValidationError
 

src/asyncapi_python/contrib/codec/registry.py

@@ -1,7 +1,9 @@
-from typing import ClassVar, Any
 from types import ModuleType
-from asyncapi_python.kernel.codec import CodecFactory, Codec
+from typing import Any, ClassVar
+
+from asyncapi_python.kernel.codec import Codec, CodecFactory
 from asyncapi_python.kernel.document.message import Message
+
 from .json import JsonCodecFactory
 
 

src/asyncapi_python/contrib/wire/amqp/consumer.py

@@ -6,10 +6,10 @@
 try:
     from aio_pika import ExchangeType  # type: ignore[import-not-found]
     from aio_pika.abc import (  # type: ignore[import-not-found]
-        AbstractConnection,
         AbstractChannel,
-        AbstractQueue,
+        AbstractConnection,
         AbstractExchange,
+        AbstractQueue,
     )
 except ImportError as e:
     raise ImportError(

src/asyncapi_python/contrib/wire/amqp/factory.py

@@ -1,7 +1,8 @@
 """AMQP wire factory implementation"""
 
 import secrets
-from typing import Optional, Callable, Any, cast
+from typing import Any, Callable, Optional, cast
+
 from typing_extensions import Unpack
 
 try:
@@ -13,11 +14,11 @@
     ) from e
 
 from asyncapi_python.kernel.wire import AbstractWireFactory, EndpointParams
-from asyncapi_python.kernel.wire.typing import Producer, Consumer
+from asyncapi_python.kernel.wire.typing import Consumer, Producer
 
-from .message import AmqpWireMessage, AmqpIncomingMessage
-from .producer import AmqpProducer
 from .consumer import AmqpConsumer
+from .message import AmqpIncomingMessage, AmqpWireMessage
+from .producer import AmqpProducer
 from .resolver import resolve_amqp_config
 
 
@@ -135,8 +136,12 @@ async def create_consumer(
         # Generate operation name from available information
         operation_name = self._generate_operation_name(kwargs)
 
+        # Use provided app_id if available, otherwise use instance app_id
+        # This allows application-level control over queue naming
+        app_id = kwargs.get("app_id", self._app_id)
+
         # Resolve AMQP configuration using pattern matching
-        config = resolve_amqp_config(kwargs, operation_name, self._app_id)
+        config = resolve_amqp_config(kwargs, operation_name, app_id)
 
         connection = await self._get_connection()
 
@@ -154,8 +159,12 @@ async def create_producer(
         # Generate operation name from available information
         operation_name = self._generate_operation_name(kwargs)
 
+        # Use provided app_id if available, otherwise use instance app_id
+        # This allows application-level control over queue naming
+        app_id = kwargs.get("app_id", self._app_id)
+
         # Resolve AMQP configuration using pattern matching
-        config = resolve_amqp_config(kwargs, operation_name, self._app_id)
+        config = resolve_amqp_config(kwargs, operation_name, app_id)
 
         connection = await self._get_connection()
 

src/asyncapi_python/contrib/wire/amqp/resolver.py

@@ -1,12 +1,13 @@
 """Binding resolution with comprehensive pattern matching"""
 
 from typing import Any
-from asyncapi_python.kernel.wire import EndpointParams
-from asyncapi_python.kernel.document.channel import Channel
+
 from asyncapi_python.kernel.document.bindings import AmqpChannelBinding
+from asyncapi_python.kernel.document.channel import Channel
+from asyncapi_python.kernel.wire import EndpointParams
 
-from .config import AmqpConfig, AmqpBindingType
-from .utils import validate_parameters_strict, substitute_parameters
+from .config import AmqpBindingType, AmqpConfig
+from .utils import substitute_parameters, validate_parameters_strict
 
 
 def resolve_amqp_config(
@@ -57,17 +58,32 @@ def resolve_amqp_config(
                 },
             )
 
-        # Reply channel with explicit address - shared channel with filtering
+        # Reply channel with explicit address - check if direct queue or topic exchange
         case (True, _, address, _) if address:
             resolved_address = substitute_parameters(address, param_values)
-            return AmqpConfig(
-                queue_name=f"reply-{app_id}",  # App-specific reply queue
-                exchange_name=resolved_address,  # Shared exchange for replies
-                exchange_type="topic",  # Enable pattern matching for filtering
-                routing_key=app_id,  # Filter messages by app_id
-                binding_type=AmqpBindingType.REPLY,
-                queue_properties={"durable": True, "exclusive": False},
-            )
+            # If address starts with "reply-", treat it as a direct queue name (RPC pattern)
+            if resolved_address.startswith("reply-"):
+                return AmqpConfig(
+                    queue_name=resolved_address,  # Use address as queue name
+                    exchange_name="",  # Default exchange for direct routing
+                    routing_key=resolved_address,  # Route directly to queue
+                    binding_type=AmqpBindingType.REPLY,
+                    queue_properties={
+                        "durable": False,
+                        "exclusive": True,
+                        "auto_delete": True,
+                    },
+                )
+            else:
+                # Topic-based reply pattern - shared exchange with filtering
+                return AmqpConfig(
+                    queue_name=f"reply-{app_id}",  # App-specific reply queue
+                    exchange_name=resolved_address,  # Shared exchange for replies
+                    exchange_type="topic",  # Enable pattern matching for filtering
+                    routing_key=app_id,  # Filter messages by app_id
+                    binding_type=AmqpBindingType.REPLY,
+                    queue_properties={"durable": True, "exclusive": False},
+                )
 
         # Reply channel with binding - defer to binding resolution
         case (True, binding, _, _) if binding and binding.type == "queue":

src/asyncapi_python/contrib/wire/amqp/utils.py

@@ -3,6 +3,7 @@
 # TODO: This thing should be general wire utils, not tied to specific wire
 
 import re
+
 from asyncapi_python.kernel.document.channel import Channel
 
 

src/asyncapi_python/kernel/application.py

@@ -1,34 +1,44 @@
 import asyncio
-from typing import TypedDict, Any
-from typing_extensions import Unpack, Required, NotRequired
+from typing import Any, TypedDict
+
+from typing_extensions import NotRequired, Required, Unpack
 
 from asyncapi_python.kernel.document.operation import Operation
 from asyncapi_python.kernel.wire import AbstractWireFactory
+
+from .codec import CodecFactory
 from .endpoint import AbstractEndpoint, EndpointFactory
 from .endpoint.abc import EndpointParams
-from .codec import CodecFactory
 
 
 class BaseApplication:
     class Inputs(TypedDict):
         wire_factory: Required[AbstractWireFactory[Any, Any]]
         codec_factory: Required[CodecFactory[Any, Any]]
+        service_name: NotRequired[str]
         endpoint_params: NotRequired[EndpointParams]
 
     def __init__(self, **kwargs: Unpack[Inputs]) -> None:
         self.__endpoints: set[AbstractEndpoint] = set()
         self.__wire_factory: AbstractWireFactory[Any, Any] = kwargs["wire_factory"]
         self.__codec_factory: CodecFactory[Any, Any] = kwargs["codec_factory"]
+        self.__service_name: str = kwargs.get("service_name", "app")
         self.__endpoint_params: EndpointParams = kwargs.get("endpoint_params", {})
         self._stop_event: asyncio.Event | None = None
         self._monitor_task: asyncio.Task[None] | None = None
         self._exception_future: asyncio.Future[Exception] | None = None
 
+    @property
+    def service_name(self) -> str:
+        """Get the service name for this application"""
+        return self.__service_name
+
     def _register_endpoint(self, op: Operation) -> AbstractEndpoint:
         endpoint = EndpointFactory.create(
             operation=op,
             wire_factory=self.__wire_factory,
             codec_factory=self.__codec_factory,
+            service_name=self.__service_name,
             endpoint_params=self.__endpoint_params,
         )
         self.__endpoints.add(endpoint)

src/asyncapi_python/kernel/codec.py

@@ -3,6 +3,7 @@
 from typing import Generic, Protocol
 
 from asyncapi_python.kernel.document.message import Message
+
 from .typing import T_DecodedPayload, T_EncodedPayload
 
 

src/asyncapi_python/kernel/document/__init__.py

@@ -1,3 +1,10 @@
+from .bindings import (
+    AmqpChannelBinding,
+    AmqpExchange,
+    AmqpExchangeType,
+    AmqpOperationBinding,
+    AmqpQueue,
+)
 from .channel import AddressParameter, Channel, ChannelBindings
 from .common import ExternalDocs, Server, Tag
 from .message import (
@@ -15,13 +22,6 @@
     OperationTrait,
     SecurityScheme,
 )
-from .bindings import (
-    AmqpChannelBinding,
-    AmqpOperationBinding,
-    AmqpExchange,
-    AmqpQueue,
-    AmqpExchangeType,
-)
 
 __all__ = [
     # channel