Move service name to endpoint params

Author: WiZeYAR

src/asyncapi_python/kernel/application.py

@@ -15,30 +15,22 @@ 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/endpoint/abc.py

@@ -10,12 +10,11 @@
 from ..typing import BatchConfig, Handler, T_Input, T_Output
 
 
-class EndpointParams(TypedDict):
+class EndpointParams(TypedDict, total=False):
     """Optional parameters for endpoint configuration"""
 
-    disable_handler_validation: NotRequired[
-        bool
-    ]  # Opt-out of handler enforcement for testing
+    service_name: str  # Service name for generating app_id
+    disable_handler_validation: bool  # Opt-out of handler enforcement for testing
 
 
 class HandlerParams(TypedDict):
@@ -31,7 +30,6 @@ class Inputs(TypedDict):
         operation: Required[Operation]
         wire_factory: Required[AbstractWireFactory[Any, Any]]
         codec_factory: Required[CodecFactory[Any, Any]]
-        service_name: NotRequired[str]  # Service name for app_id generation
         endpoint_params: NotRequired[EndpointParams]  # Optional endpoint configuration
 
     class StartParams(TypedDict):
@@ -43,7 +41,6 @@ class StartParams(TypedDict):
     def __init__(self, **kwargs: Unpack[Inputs]):
         self._operation = kwargs["operation"]
         self._wire = kwargs["wire_factory"]
-        self._service_name = kwargs.get("service_name", "app")
         codec_factory = kwargs["codec_factory"]
         # Endpoint sets its own defaults - empty dict if not provided
         self._endpoint_params = kwargs.get("endpoint_params", {})

src/asyncapi_python/kernel/endpoint/rpc_client.py

@@ -45,16 +45,19 @@ async def start(self, **params: Unpack[AbstractEndpoint.StartParams]) -> None:
 
         # Ensure global reply handling is set up (only happens once)
         await global_reply_handler.ensure_reply_handler(
-            self._wire, self._operation, self._service_name
+            self._wire, self._operation, self._endpoint_params
         )
 
+        # Extract service_name from endpoint_params for app_id
+        service_name = self._endpoint_params.get("service_name", "app")
+
         # Create instance-specific producer for sending requests
         self._producer = await self._wire.create_producer(
             channel=self._operation.channel,
             parameters={},
             op_bindings=self._operation.bindings,
             is_reply=False,
-            app_id=self._service_name,
+            app_id=service_name,
         )
 
         # Start producer

src/asyncapi_python/kernel/endpoint/rpc_reply_handler.py

@@ -5,7 +5,7 @@
 from typing import Any
 
 from asyncapi_python.kernel.document import Channel, Operation
-from asyncapi_python.kernel.wire import AbstractWireFactory, Consumer
+from asyncapi_python.kernel.wire import AbstractWireFactory, Consumer, EndpointParams
 
 from ..typing import IncomingMessage
 
@@ -29,16 +29,19 @@ async def ensure_reply_handler(
         self,
         wire_factory: AbstractWireFactory[Any, Any],
         operation: Operation,
-        service_name: str = "app",
+        endpoint_params: EndpointParams,
     ) -> None:
         """Ensure reply consumer and task are running
 
         Args:
             wire_factory: Wire factory for creating consumer
             operation: Operation definition
-            service_name: Service name for generating consistent app_id
+            endpoint_params: Endpoint parameters including service_name
         """
         if self._reply_consumer is None:
+            # Extract service_name from endpoint_params
+            service_name = endpoint_params.get("service_name", "app")
+
             # Generate app_id with service name + random hex (same format as AmqpWire)
             random_hex = secrets.token_hex(4)  # 4 bytes = 8 hex chars
             app_id = f"{service_name}-{random_hex}"

tests/integration/scenarios/fan_in_logging.py

@@ -27,11 +27,14 @@ def __init__(
         wire_factory: AbstractWireFactory,
         codec_factory: CodecFactory,
     ):
+        # Pass service_name via endpoint_params
+        endpoint_params = {"service_name": service_name}
         super().__init__(
             wire_factory=wire_factory,
             codec_factory=codec_factory,
-            service_name=service_name,
+            endpoint_params=endpoint_params,
         )
+        self.service_name = service_name  # Store for use in _setup_endpoints
         self._setup_endpoints()
 
     def _setup_endpoints(self):

tests/integration/scenarios/fan_out_broadcasting.py

@@ -108,11 +108,14 @@ def __init__(
         wire_factory: AbstractWireFactory,
         codec_factory: CodecFactory,
     ):
+        # Pass service_name via endpoint_params
+        endpoint_params = {"service_name": service_name}
         super().__init__(
             wire_factory=wire_factory,
             codec_factory=codec_factory,
-            service_name=service_name,
+            endpoint_params=endpoint_params,
         )
+        self.service_name = service_name  # Store for use in _setup_endpoints
         self._setup_endpoints()
 
     def _setup_endpoints(self):