Add basic implementation of amqp_params

Author: WiZeYAR

.devcontainer/devcontainer.json

@@ -25,6 +25,7 @@
 				"github.vscode-github-actions"
 			],
 			"settings": {
+				"python.defaultInterpreterPath": "/workspaces/${localWorkspaceFolderBasename}/.venv/bin/python",
 				"yaml.schemas": {
 					"https://asyncapi.com/schema-store/3.0.0-without-$id.json": [
 						"file:///workspaces/asyncapi-python/examples/*.yaml"

src/asyncapi_python/amqp/__init__.py

@@ -19,9 +19,11 @@
 from .operation import Operation
 from .utils import union_model
 from .error import Rejection, RejectedError
+from .params import AmqpParams
 
 __all__ = [
     "channel_pool",
+    "AmqpParams",
     "AmqpPool",
     "BaseApplication",
     "Router",

src/asyncapi_python/amqp/base_application.py

@@ -23,6 +23,7 @@
 from .endpoint import EndpointParams
 from .connection import channel_pool
 from .utils import encode_message, decode_message
+from .params import AmqpParams
 from typing import Generic, Optional, TypeVar
 
 
@@ -61,6 +62,7 @@ def __init__(
         amqp_uri: str,
         producer_factory: type[P],
         consumer_factory: type[C],
+        amqp_params: AmqpParams,
     ):
         self.__params = EndpointParams(
             pool=channel_pool(amqp_uri),
@@ -69,6 +71,7 @@ def __init__(
             register_correlation_id=self.__register_correlation_id,
             stop_application=self.stop,
             app_id=str(uuid4()),
+            amqp_params=amqp_params,
         )
         self.__reply_futures: dict[
             str,

src/asyncapi_python/amqp/connection.py

@@ -14,6 +14,7 @@
 
 
 from functools import cache
+from typing import TypeAlias
 from aio_pika.robust_connection import (
     AbstractRobustConnection,
     AbstractRobustChannel,

src/asyncapi_python/amqp/endpoint/base.py

@@ -31,6 +31,7 @@
 from ..error import Rejection, RejectedError
 from ..connection import AmqpPool
 from ..operation import Operation
+from ..params import AmqpParams
 from aio_pika.abc import (
     AbstractRobustChannel,
     AbstractRobustQueue,
@@ -64,6 +65,7 @@ class EndpointParams:
     ]
     app_id: str
     stop_application: Callable[[], Awaitable[None]]
+    amqp_params: AmqpParams
 
     @property
     def reply_queue_name(self) -> str:

src/asyncapi_python/amqp/endpoint/receiver.py

@@ -45,6 +45,8 @@ async def start(self) -> None:
         print("start", self._op)
         if self._fn:
             async with self._params.pool.acquire() as ch:
+                if self._params.amqp_params.prefetch_count:
+                    await ch.set_qos(prefetch_count=self._params.amqp_params.prefetch_count)
                 q = self._queue = await self._declare(ch)
                 self._consumer_tag = await q.consume(self._consumer)
             return

src/asyncapi_python/amqp/params.py

@@ -0,0 +1,4 @@
+from typing import TypedDict
+
+class AmqpParams(TypedDict, total=False):
+    prefetch_count: str

src/asyncapi_python_codegen/generators/amqp/templates/application.py.j2

@@ -13,9 +13,9 @@
    limitations under the License. #}
 from .consumer import _Router_0 as Consumer
 from .producer import _Router_0 as Producer
-from asyncapi_python.amqp.base_application import BaseApplication
+from asyncapi_python.amqp import BaseApplication, AmqpParams
 
 
 class Application(BaseApplication[Producer, Consumer]):
-    def __init__(self, amqp_uri: str):
-        super().__init__(amqp_uri, Producer, Consumer)
+    def __init__(self, amqp_uri: str, amqp_params: AmqpParams = {}):
+        super().__init__(amqp_uri, Producer, Consumer, amqp_params)