diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 87b2c5be..a9d38443 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -56,6 +56,7 @@ jobs: - amgi-kafka-event-source-mapping - amgi-nats - amgi-paho-mqtt + - amgi-pulsar - amgi-redis - amgi-sqs-event-source-mapping - amgi-types diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index a5ea9a09..7112a782 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -11,7 +11,7 @@ repos: rev: v1.19.1 hooks: - id: mypy - additional_dependencies: ["pydantic==2.12.5", "nats-py>=2.10.0", "types-docutils", "types-PyYAML", "sphinx>=7.4.7", "starlette==1.2.1"] + additional_dependencies: ["pydantic==2.12.5", "nats-py>=2.10.0", "pulsar-client>=3.13.0", "types-docutils", "types-PyYAML", "sphinx>=7.4.7", "starlette==1.2.1"] - repo: https://github.com/PyCQA/autoflake rev: v2.3.3 hooks: diff --git a/README.md b/README.md index 255207f2..5c211fc4 100644 --- a/README.md +++ b/README.md @@ -202,6 +202,11 @@ The following AMGI servers are currently available: - **[amgi-nats](https://pypi.org/project/amgi-nats/)** - Contains [NATS](https://nats.io/) push and pull server implementations +### Pulsar + +- **[amgi-pulsar](https://pypi.org/project/amgi-pulsar/)** - [Apache Pulsar](https://pulsar.apache.org/) server + implementation + ### Redis - **[amgi-redis](https://pypi.org/project/amgi-redis/)** - [Redis](https://redis.io/) server implementation diff --git a/docs/specifications/bindings.rst b/docs/specifications/bindings.rst index b754edea..8e761066 100644 --- a/docs/specifications/bindings.rst +++ b/docs/specifications/bindings.rst @@ -9,14 +9,24 @@ A message sent to an application can only contain one protocol in its binding ob can include multiple protocol bindings. The server simply ignores the binding if it cannot handle it. This is to allow for progressive migrations of an application -******* +******** Kafka -******* +******** The Kafka binding object has the following key: -- ``message["bindings"]["keys"]`` (:py:obj:`~typing.NotRequired` [ :py:obj:`~typing.Optional` [:py:obj:`bytes` ]]) - - The :py:obj:`bytes` of the Kafka key, this can be also be :py:obj:`None`. If missing, it defaults to :py:obj:`None`. +- ``message["bindings"]["kafka"]["key"]`` (:py:obj:`~typing.NotRequired` [ :py:obj:`~typing.Optional` [:py:obj:`bytes` ]]) - + The :py:obj:`bytes` of the Kafka key, this can also be :py:obj:`None`. If missing, it defaults to :py:obj:`None`. + +******** + Pulsar +******** + +The Pulsar binding object has the following key: + +- ``message["bindings"]["pulsar"]["key"]`` (:py:obj:`~typing.NotRequired` [ :py:obj:`~typing.Optional` [:py:obj:`str` ]]) - + The :py:obj:`str` of the Pulsar partition key, this can also be :py:obj:`None`. If missing, it defaults to + :py:obj:`None`. .. _asyncapi: https://www.asyncapi.com/en diff --git a/example/README.md b/example/README.md index 01818578..7a201cdd 100644 --- a/example/README.md +++ b/example/README.md @@ -88,6 +88,25 @@ Or via NATS pull subscriptions with: asyncfast run amgi-nats-pull main:app input_channel ``` +## Pulsar + +Run the Pulsar compose file with: + +```commandline +docker compose --file apache-pulsar/docker-compose.yaml up --detach +``` + +This includes: + +- A Pulsar standalone cluster running on `pulsar://localhost:6650` +- Pulsar HTTP service ([`http://localhost:8080/`](http://localhost:8080/)) + +Connect the app via Pulsar with: + +```commandline +asyncfast run amgi-pulsar main:app input_channel +``` + ## SQS Run the SQS compose file with: diff --git a/example/apache-pulsar/docker-compose.yaml b/example/apache-pulsar/docker-compose.yaml new file mode 100644 index 00000000..7494f05f --- /dev/null +++ b/example/apache-pulsar/docker-compose.yaml @@ -0,0 +1,8 @@ +services: + pulsar: + image: ghcr.io/asyncfast/apachepulsar/pulsar:4.0.5 + command: bin/pulsar standalone + restart: unless-stopped + ports: + - "6650:6650" + - "8080:8080" diff --git a/example/pyproject.toml b/example/pyproject.toml index 804a7584..376a44f6 100644 --- a/example/pyproject.toml +++ b/example/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "amgi-aiokafka", "amgi-nats", "amgi-paho-mqtt", + "amgi-pulsar", "asyncfast", ] @@ -25,4 +26,5 @@ sources.amgi-aio-pika = { workspace = true } sources.amgi-aiokafka = { workspace = true } sources.amgi-nats = { workspace = true } sources.amgi-paho-mqtt = { workspace = true } +sources.amgi-pulsar = { workspace = true } sources.asyncfast = { workspace = true } diff --git a/packages/amgi-pulsar/LICENSE b/packages/amgi-pulsar/LICENSE new file mode 100644 index 00000000..5da9ba59 --- /dev/null +++ b/packages/amgi-pulsar/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 Jack Burridge + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/packages/amgi-pulsar/README.md b/packages/amgi-pulsar/README.md new file mode 100644 index 00000000..1aaed86f --- /dev/null +++ b/packages/amgi-pulsar/README.md @@ -0,0 +1,58 @@ +# amgi-pulsar + +amgi-pulsar is an [AMGI](https://amgi.readthedocs.io/en/latest/) compatible server project, supporting running AMGI +applications against [Apache Pulsar](https://pulsar.apache.org/) topics. + +The server consumes messages from a shared subscription, which is created if it does not exist. The subscription type +can be configured with `consumer_type`; `Shared` (the default) provides no ordering guarantees between consumers, while +`Exclusive`, `Failover` and `KeyShared` are also available. Messages are only acknowledged when the application sends a +`message.ack` event; a `message.nack` event triggers a redelivery. Messages that remain unacknowledged for longer than +`unacked_messages_timeout_ms` (30000 by default) are also redelivered. + +## Installation + +``` +pip install amgi-pulsar==0.47.0 +``` + +## Example + +This example uses [AsyncFast](https://pypi.org/project/asyncfast/): + +```python +from dataclasses import dataclass + +from amgi_pulsar import run +from asyncfast import AsyncFast + +app = AsyncFast() + + +@dataclass +class Order: + item_ids: list[str] + + +@app.channel("order.topic") +async def order_topic(order: Order) -> None: + # Makes an order + ... + + +if __name__ == "__main__": + run(app, "order.topic") +``` + +Or the application could be run via the commandline: + +```commandline +asyncfast run amgi-pulsar main:app order.topic +``` + +## Contact + +For questions or suggestions, please contact [jack.burridge@mail.com](mailto:jack.burridge@mail.com). + +## License + +Copyright 2025 AMGI diff --git a/packages/amgi-pulsar/pyproject.toml b/packages/amgi-pulsar/pyproject.toml new file mode 100644 index 00000000..80d80695 --- /dev/null +++ b/packages/amgi-pulsar/pyproject.toml @@ -0,0 +1,49 @@ +[build-system] +build-backend = "uv_build" +requires = [ + "uv-build>=0.8.14,<0.9.0", +] + +[project] +name = "amgi-pulsar" +version = "0.49.0" +description = "AMGI Apache Pulsar Server" +readme = "README.md" +license = "MIT" +license-files = [ + "LICENSE", +] +authors = [ + { name = "jack.burridge", email = "jack.burridge@mail.com" }, +] +requires-python = ">=3.10" +classifiers = [ + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", +] +dependencies = [ + "amgi-common==0.49.0", + "amgi-types==0.49.0", + "pulsar-client>=3.10.0", + "typing-extensions>=4.15.0; python_full_version<'3.11'", +] +entry-points.amgi_server.amgi-pulsar = "amgi_pulsar:_run_cli" + +[dependency-groups] +dev = [ + "pytest>=8.4.1", + "pytest-asyncio>=1.3.0", + "pytest-cov>=7.0.0", + "pytest-timeout>=2.4.0", + "test-utils", + "testcontainers>=4.13.0", +] + +[tool.uv] +sources.amgi-common.workspace = true +sources.amgi-types.workspace = true +sources.test-utils.workspace = true diff --git a/packages/amgi-pulsar/src/amgi_pulsar/__init__.py b/packages/amgi-pulsar/src/amgi_pulsar/__init__.py new file mode 100644 index 00000000..38131a29 --- /dev/null +++ b/packages/amgi-pulsar/src/amgi_pulsar/__init__.py @@ -0,0 +1,260 @@ +import asyncio +import logging +import sys +from asyncio import Task +from collections.abc import Awaitable +from collections.abc import Callable +from collections.abc import Sequence +from types import TracebackType +from typing import Any +from typing import AsyncContextManager + +import pulsar.asyncio +from amgi_common import Lifespan +from amgi_common import server_serve +from amgi_common import Stoppable +from amgi_types import AMGIApplication +from amgi_types import AMGIReceiveEvent +from amgi_types import AMGISendEvent +from amgi_types import MessageScope +from amgi_types import MessageSendEvent + +if sys.version_info >= (3, 11): + from typing import Self +else: + from typing_extensions import Self + +__all__ = ["MessageSend", "Server", "run"] + +_MessageSendT = Callable[[MessageSendEvent], Awaitable[None]] +_MessageSendManagerT = AsyncContextManager[_MessageSendT] + +logger = logging.getLogger("amgi-pulsar.error") + + +def run( + app: AMGIApplication, + *topics: str, + service_url: str = "pulsar://localhost:6650", + subscription_name: str = "amgi", + negative_ack_redelivery_delay_ms: int = 60000, + consumer_type: pulsar.ConsumerType = pulsar.ConsumerType.Shared, + unacked_messages_timeout_ms: int = 30000, + message_send: _MessageSendManagerT | None = None, +) -> None: + server = Server( + app, + *topics, + service_url=service_url, + subscription_name=subscription_name, + negative_ack_redelivery_delay_ms=negative_ack_redelivery_delay_ms, + consumer_type=consumer_type, + unacked_messages_timeout_ms=unacked_messages_timeout_ms, + message_send=message_send, + ) + server_serve(server) + + +def _run_cli( + app: AMGIApplication, + topics: list[str], + service_url: str = "pulsar://localhost:6650", + subscription_name: str = "amgi", +) -> None: + run(app, *topics, service_url=service_url, subscription_name=subscription_name) + + +async def _receive() -> AMGIReceiveEvent: + raise RuntimeError("Receive should not be called") + + +def _encode_headers(properties: dict[str, str] | None) -> list[tuple[bytes, bytes]]: + if not properties: + return [] + return [(key.encode(), value.encode()) for key, value in properties.items()] + + +def _decode_headers( + headers: Sequence[tuple[bytes, bytes]], +) -> dict[str, str]: + return {key.decode(): value.decode() for key, value in headers} + + +class MessageSend: + def __init__(self, service_url: str) -> None: + self._service_url = service_url + self._client: pulsar.asyncio.Client | None = None + self._producers: dict[str, pulsar.asyncio.Producer] = {} + self._producers_lock = asyncio.Lock() + + async def __aenter__(self) -> Self: + self._client = pulsar.asyncio.Client(self._service_url) + return self + + async def __call__(self, event: MessageSendEvent) -> None: + producer = await self._get_producer(event["address"]) + key = event.get("bindings", {}).get("pulsar", {}).get("key") + + await producer.send( + event.get("payload") or b"", + properties=_decode_headers(event["headers"]), + partition_key=key, + ) + + async def _get_producer(self, topic: str) -> pulsar.asyncio.Producer: + producer = self._producers.get(topic) + if producer is not None: + return producer + + async with self._producers_lock: + producer = self._producers.get(topic) + if producer is None: + producer = await self._require_client().create_producer(topic) + self._producers[topic] = producer + return producer + + def _require_client(self) -> pulsar.asyncio.Client: + if self._client is None: + raise RuntimeError("MessageSend not initialized") + return self._client + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + if self._client is not None: + self._producers.clear() + client, self._client = self._client, None + await client.close() + + +class _Send: + def __init__( + self, + consumer: pulsar.asyncio.Consumer, + message: pulsar.Message, + message_send: _MessageSendT, + ) -> None: + self._consumer = consumer + self._message = message + self._message_send = message_send + + async def __call__(self, event: AMGISendEvent) -> None: + if event["type"] == "message.ack": + await self._consumer.acknowledge(self._message) + elif event["type"] == "message.nack": + await self._consumer.negative_acknowledge(self._message) + elif event["type"] == "message.send": + await self._message_send(event) + + +class Server: + def __init__( + self, + app: AMGIApplication, + *topics: str, + service_url: str, + subscription_name: str, + negative_ack_redelivery_delay_ms: int = 60000, + consumer_type: pulsar.ConsumerType = pulsar.ConsumerType.Shared, + unacked_messages_timeout_ms: int = 30000, + message_send: _MessageSendManagerT | None = None, + ) -> None: + self._app = app + self._topics = topics + self._service_url = service_url + self._subscription_name = subscription_name + self._negative_ack_redelivery_delay_ms = negative_ack_redelivery_delay_ms + self._consumer_type = consumer_type + self._unacked_messages_timeout_ms = unacked_messages_timeout_ms + self._message_send = message_send or MessageSend(service_url) + self._stoppable = Stoppable() + self._tasks = set[Task[None]]() + + async def serve(self) -> None: + client = pulsar.asyncio.Client(self._service_url) + try: + async with self._message_send as message_send: + consumers = [ + await self._subscribe(client, topic) for topic in self._topics + ] + async with Lifespan(self._app) as state: + topic_tasks = [ + asyncio.create_task( + self._topic_loop(consumer, topic, message_send, state) + ) + for consumer, topic in zip(consumers, self._topics) + ] + try: + await asyncio.gather(*topic_tasks) + except BaseException: + for topic_task in topic_tasks: + topic_task.cancel() + await asyncio.gather(*topic_tasks, return_exceptions=True) + handler_tasks = list(self._tasks) + for handler_task in handler_tasks: + handler_task.cancel() + await asyncio.gather(*handler_tasks, return_exceptions=True) + raise + await asyncio.gather(*self._tasks, return_exceptions=True) + finally: + await client.close() + + async def _subscribe( + self, client: pulsar.asyncio.Client, topic: str + ) -> pulsar.asyncio.Consumer: + return await client.subscribe( + topic, + self._subscription_name, + consumer_type=self._consumer_type, + negative_ack_redelivery_delay_ms=self._negative_ack_redelivery_delay_ms, + unacked_messages_timeout_ms=self._unacked_messages_timeout_ms, + ) + + async def _topic_loop( + self, + consumer: pulsar.asyncio.Consumer, + topic: str, + message_send: _MessageSendT, + state: dict[str, Any], + ) -> None: + loop = asyncio.get_running_loop() + async for message in self._stoppable.call(consumer.receive): + task = loop.create_task( + self._handle_message(consumer, topic, message, message_send, state) + ) + self._tasks.add(task) + task.add_done_callback(self._task_done) + + def _task_done(self, task: Task[None]) -> None: + self._tasks.discard(task) + if task.cancelled(): + return + if (exception := task.exception()) is not None: + logger.exception("Message task failed", exc_info=exception) + + async def _handle_message( + self, + consumer: pulsar.asyncio.Consumer, + topic: str, + message: pulsar.Message, + message_send: _MessageSendT, + state: dict[str, Any], + ) -> None: + # The C client returns "" for keyless messages and cannot distinguish an + # explicitly empty key from a missing one, so map both to None + scope: MessageScope = { + "type": "message", + "amgi": {"version": "2.0", "spec_version": "2.0"}, + "address": topic, + "headers": _encode_headers(message.properties()), + "payload": message.data(), + "bindings": {"pulsar": {"key": message.partition_key() or None}}, + "state": state.copy(), + } + await self._app(scope, _receive, _Send(consumer, message, message_send)) + + def stop(self) -> None: + self._stoppable.stop() diff --git a/packages/amgi-pulsar/src/amgi_pulsar/py.typed b/packages/amgi-pulsar/src/amgi_pulsar/py.typed new file mode 100644 index 00000000..e69de29b diff --git a/packages/amgi-pulsar/tests_amgi_pulsar/__init__.py b/packages/amgi-pulsar/tests_amgi_pulsar/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/amgi-pulsar/tests_amgi_pulsar/test_pulsar_message.py b/packages/amgi-pulsar/tests_amgi_pulsar/test_pulsar_message.py new file mode 100644 index 00000000..e7756642 --- /dev/null +++ b/packages/amgi-pulsar/tests_amgi_pulsar/test_pulsar_message.py @@ -0,0 +1,261 @@ +import asyncio +import logging +from collections.abc import Awaitable +from collections.abc import Callable + +import pulsar.asyncio +import pytest +from amgi_pulsar import _decode_headers +from amgi_pulsar import _encode_headers +from amgi_pulsar import MessageSend +from amgi_pulsar import Server +from amgi_types import AMGIReceiveCallable +from amgi_types import AMGISendCallable +from amgi_types import MessageSendEvent +from amgi_types import Scope + + +class _FakeClient: + def __init__(self) -> None: + self.created_topics: list[str] = [] + + async def create_producer(self, topic: str) -> object: + self.created_topics.append(topic) + await asyncio.sleep(0) + return object() + + +class _FakeMessage: + def data(self) -> bytes: + return b"" + + def properties(self) -> dict[str, str]: + return {} + + def partition_key(self) -> str: + return "" + + +class _FakeConsumer: + def __init__(self, messages: list[_FakeMessage]) -> None: + self._messages = list(messages) + + async def receive(self) -> _FakeMessage: + if self._messages: + message = self._messages.pop(0) + await asyncio.sleep(0) + return message + await asyncio.Event().wait() + raise AssertionError # pragma: no cover + + +async def test_message_send_requires_context_manager() -> None: + message_send = MessageSend("pulsar://localhost:6650") + + with pytest.raises(RuntimeError, match="MessageSend not initialized"): + await message_send({"type": "message.send", "address": "topic", "headers": []}) + + +async def test_message_send_creates_one_producer_per_topic() -> None: + message_send = MessageSend("pulsar://localhost:6650") + fake_client = _FakeClient() + message_send._client = fake_client + + first, second = await asyncio.gather( + message_send._get_producer("topic"), + message_send._get_producer("topic"), + ) + third = await message_send._get_producer("topic") + + assert first is second is third + assert fake_client.created_topics == ["topic"] + + +def test_encode_headers_handles_empty_and_values() -> None: + assert _encode_headers(None) == [] + assert _encode_headers({}) == [] + assert _encode_headers({"name": "value"}) == [(b"name", b"value")] + + +def test_decode_headers_handles_empty_and_values() -> None: + assert _decode_headers([]) == {} + assert _decode_headers([(b"name", b"value")]) == {"name": "value"} + + +async def test_message_handler_failure_is_logged( + caplog: pytest.LogCaptureFixture, +) -> None: + handled_count = 0 + second_message_handled = asyncio.Event() + + async def app( + scope: Scope, + receive: AMGIReceiveCallable, + send: AMGISendCallable, + ) -> None: + nonlocal handled_count + handled_count += 1 + if handled_count == 2: + second_message_handled.set() + raise RuntimeError("Handler failed") + + async def message_send(event: MessageSendEvent) -> None: + raise AssertionError # pragma: no cover + + server = Server( + app, "topic", service_url="pulsar://localhost:6650", subscription_name="amgi" + ) + consumer = _FakeConsumer([_FakeMessage(), _FakeMessage()]) + + with caplog.at_level(logging.ERROR): + topic_loop = asyncio.create_task( + server._topic_loop(consumer, "topic", message_send, {}) + ) + await asyncio.wait_for(second_message_handled.wait(), timeout=5) + server.stop() + await topic_loop + + assert handled_count == 2 + records = [ + record for record in caplog.records if record.name == "amgi-pulsar.error" + ] + assert len(records) == 2 + for record in records: + assert record.levelno == logging.ERROR + assert record.exc_info is not None + assert isinstance(record.exc_info[1], RuntimeError) + + +class _FailingConsumer: + def __init__(self, handler_started: asyncio.Event | None = None) -> None: + self._handler_started = handler_started + self._delivered = False + + async def receive(self) -> _FakeMessage: + if self._handler_started is not None and not self._delivered: + self._delivered = True + return _FakeMessage() + if self._handler_started is not None: + await self._handler_started.wait() + raise RuntimeError("Receive failed") + + +class _FakeServeClient: + def __init__(self, consumers: list[object]) -> None: + self._consumers = consumers + self.closed = False + + async def subscribe( + self, topic: str, subscription_name: str, **kwargs: object + ) -> object: + return self._consumers.pop(0) + + async def close(self) -> None: + self.closed = True + + +class _FakeMessageSend: + async def __aenter__(self) -> Callable[[MessageSendEvent], Awaitable[None]]: + async def send(event: MessageSendEvent) -> None: + raise AssertionError # pragma: no cover + + return send + + async def __aexit__(self, *args: object) -> None: + return None + + +async def test_serve_cancels_topic_loops_when_one_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + client = _FakeServeClient([_FailingConsumer(), _FakeConsumer([])]) + monkeypatch.setattr(pulsar.asyncio, "Client", lambda url: client) + + async def app( + scope: Scope, + receive: AMGIReceiveCallable, + send: AMGISendCallable, + ) -> None: + raise RuntimeError("Lifespan unsupported") + + server = Server( + app, + "first", + "second", + service_url="pulsar://localhost:6650", + subscription_name="amgi", + message_send=_FakeMessageSend(), + ) + + with pytest.raises(RuntimeError, match="Receive failed"): + await server.serve() + + assert client.closed + + +async def test_cancelled_message_task_is_not_logged( + caplog: pytest.LogCaptureFixture, +) -> None: + async def app( + scope: Scope, + receive: AMGIReceiveCallable, + send: AMGISendCallable, + ) -> None: + raise AssertionError # pragma: no cover + + server = Server( + app, + "topic", + service_url="pulsar://localhost:6650", + subscription_name="amgi", + ) + + async def wait_forever() -> None: + await asyncio.Event().wait() + + task = asyncio.create_task(wait_forever()) + await asyncio.sleep(0) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + with caplog.at_level(logging.ERROR): + server._task_done(task) + + assert not caplog.records + + +async def test_serve_cancels_running_handlers_when_topic_loop_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + handler_started = asyncio.Event() + handler_cancelled = asyncio.Event() + client = _FakeServeClient([_FailingConsumer(handler_started)]) + monkeypatch.setattr(pulsar.asyncio, "Client", lambda url: client) + + async def app( + scope: Scope, + receive: AMGIReceiveCallable, + send: AMGISendCallable, + ) -> None: + if scope["type"] == "lifespan": + raise RuntimeError("Lifespan unsupported") + handler_started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + handler_cancelled.set() + raise + + server = Server( + app, + "topic", + service_url="pulsar://localhost:6650", + subscription_name="amgi", + message_send=_FakeMessageSend(), + ) + + with pytest.raises(RuntimeError, match="Receive failed"): + await server.serve() + + assert handler_cancelled.is_set() + assert client.closed diff --git a/packages/amgi-pulsar/tests_amgi_pulsar/test_pulsar_message_integration.py b/packages/amgi-pulsar/tests_amgi_pulsar/test_pulsar_message_integration.py new file mode 100644 index 00000000..ac0a5e20 --- /dev/null +++ b/packages/amgi-pulsar/tests_amgi_pulsar/test_pulsar_message_integration.py @@ -0,0 +1,340 @@ +import asyncio +import time +from collections.abc import AsyncGenerator +from typing import Generator +from uuid import uuid4 + +import pulsar.asyncio +import pytest +from amgi_pulsar import _run_cli +from amgi_pulsar import MessageSend +from amgi_pulsar import run +from amgi_pulsar import Server +from test_utils import assert_run_can_terminate +from test_utils import MockApp +from testcontainers.core.generic import DockerContainer + + +@pytest.fixture(scope="module") +def pulsar_container() -> Generator[DockerContainer, None, None]: + container = DockerContainer(image="ghcr.io/asyncfast/apachepulsar/pulsar:4.0.5") + container.with_command("bin/pulsar standalone") + container.with_exposed_ports(6650, 8080) + with container: + wait_until_ready(container) + yield container + + +def service_url(container: DockerContainer) -> str: + return f"pulsar://{container.get_container_host_ip()}:{container.get_exposed_port(6650)}" + + +def wait_until_ready(container: DockerContainer) -> None: + deadline = time.monotonic() + 180 + while time.monotonic() < deadline: + try: + client = pulsar.Client( + service_url(container), + operation_timeout_seconds=2, + connection_timeout_ms=2000, + ) + try: + client.create_producer("readiness-probe").close() + finally: + client.close() + return + except (OSError, pulsar.PulsarException): + time.sleep(2) + raise RuntimeError("Pulsar container failed to become ready") # pragma: no cover + + +@pytest.fixture +def topic() -> str: + return f"receive-{uuid4()}" + + +@pytest.fixture +def subscription_name() -> str: + return f"amgi-{uuid4()}" + + +@pytest.fixture +async def app( + pulsar_container: DockerContainer, topic: str, subscription_name: str +) -> AsyncGenerator[MockApp, None]: + app = MockApp() + server = Server( + app, + topic, + service_url=service_url(pulsar_container), + subscription_name=subscription_name, + ) + async with app.lifespan(server=server): + yield app + + +@pytest.mark.integration +async def test_message( + app: MockApp, topic: str, pulsar_container: DockerContainer +) -> None: + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + await producer.send(b"value") + await client.close() + + async with app.call() as (scope, receive, send): + assert scope["type"] == "message" + assert scope["address"] == topic + assert scope["amgi"] == {"version": "2.0", "spec_version": "2.0"} + assert scope["headers"] == [] + assert scope["payload"] == b"value" + assert scope["bindings"] == {"pulsar": {"key": None}} + assert scope["state"] == {} + await send({"type": "message.ack"}) + + +@pytest.mark.integration +async def test_message_uses_properties_and_key( + app: MockApp, topic: str, pulsar_container: DockerContainer +) -> None: + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + await producer.send(b"payload", properties={"name": "value"}, partition_key="key") + await client.close() + + async with app.call() as (scope, receive, send): + assert scope["type"] == "message" + assert scope["headers"] == [(b"name", b"value")] + assert scope["bindings"] == {"pulsar": {"key": "key"}} + await send({"type": "message.ack"}) + + +@pytest.mark.integration +async def test_message_send( + app: MockApp, topic: str, pulsar_container: DockerContainer +) -> None: + send_topic = f"send-{uuid4()}" + + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + reader = await client.create_reader(send_topic, pulsar.MessageId.earliest) + + await producer.send(b"") + + async with app.call() as (scope, receive, send): + await send( + { + "type": "message.send", + "address": send_topic, + "headers": [(b"key", b"value")], + "payload": b"test", + } + ) + await send({"type": "message.ack"}) + + message = await reader.read_next() + assert message.data() == b"test" + assert message.properties() == {"key": "value"} + + await reader.close() + await client.close() + + +@pytest.mark.integration +async def test_message_send_uses_key_binding( + app: MockApp, topic: str, pulsar_container: DockerContainer +) -> None: + send_topic = f"send-{uuid4()}" + + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + reader = await client.create_reader(send_topic, pulsar.MessageId.earliest) + + await producer.send(b"") + + async with app.call() as (scope, receive, send): + await send( + { + "type": "message.send", + "address": send_topic, + "headers": [], + "payload": b"payload", + "bindings": {"pulsar": {"key": "ordering-key"}}, + } + ) + await send({"type": "message.ack"}) + + message = await reader.read_next() + assert message.data() == b"payload" + assert message.partition_key() == "ordering-key" + + await reader.close() + await client.close() + + +@pytest.mark.integration +async def test_message_send_defaults_empty_payload( + app: MockApp, topic: str, pulsar_container: DockerContainer +) -> None: + send_topic = f"send-{uuid4()}" + + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + reader = await client.create_reader(send_topic, pulsar.MessageId.earliest) + + await producer.send(b"") + + async with app.call() as (scope, receive, send): + await send( + { + "type": "message.send", + "address": send_topic, + "headers": [], + } + ) + await send({"type": "message.ack"}) + + message = await reader.read_next() + assert message.data() == b"" + assert message.properties() == {} + + await reader.close() + await client.close() + + +@pytest.mark.integration +async def test_send_ignores_non_message_send_events( + app: MockApp, topic: str, pulsar_container: DockerContainer +) -> None: + send_topic = f"send-{uuid4()}" + + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + reader = await client.create_reader(send_topic, pulsar.MessageId.earliest) + + await producer.send(b"") + + async with app.call() as (scope, receive, send): + await send({"type": "message.ack"}) + + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(reader.read_next(), timeout=1) + + await reader.close() + await client.close() + + +@pytest.mark.integration +async def test_message_nack( + pulsar_container: DockerContainer, topic: str, subscription_name: str +) -> None: + app = MockApp() + server = Server( + app, + topic, + service_url=service_url(pulsar_container), + subscription_name=subscription_name, + negative_ack_redelivery_delay_ms=100, + ) + async with app.lifespan(server=server): + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + await producer.send(b"value") + + async with app.call() as (scope, receive, send): + await send({"type": "message.nack", "message": "failure"}) + + async with app.call() as (scope, receive, send): + assert scope["type"] == "message" + assert scope["payload"] == b"value" + await send({"type": "message.ack"}) + + await client.close() + + +@pytest.mark.integration +async def test_message_send_closes_initialized_client( + pulsar_container: DockerContainer, +) -> None: + message_send = MessageSend(service_url(pulsar_container)) + send_topic = f"send-{uuid4()}" + + async with message_send: + await message_send( + {"type": "message.send", "address": send_topic, "headers": []} + ) + + with pytest.raises(RuntimeError, match="MessageSend not initialized"): + await message_send( + {"type": "message.send", "address": send_topic, "headers": []} + ) + + +@pytest.mark.integration +async def test_lifespan( + pulsar_container: DockerContainer, topic: str, subscription_name: str +) -> None: + app = MockApp() + server = Server( + app, + topic, + service_url=service_url(pulsar_container), + subscription_name=subscription_name, + ) + + state_item = uuid4() + + async with app.lifespan({"item": state_item}, server): + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + await producer.send(b"") + + async with app.call() as (scope, receive, send): + assert scope["type"] == "message" + assert scope["address"] == topic + assert scope["payload"] == b"" + assert scope["bindings"] == {"pulsar": {"key": None}} + assert scope["amgi"] == {"version": "2.0", "spec_version": "2.0"} + assert scope["state"] == {"item": state_item} + await send({"type": "message.ack"}) + + await client.close() + + +@pytest.mark.integration +def test_run( + pulsar_container: DockerContainer, topic: str, subscription_name: str +) -> None: + assert_run_can_terminate( + run, + topic, + service_url=service_url(pulsar_container), + subscription_name=subscription_name, + ) + + +@pytest.mark.integration +def test_run_cli( + pulsar_container: DockerContainer, topic: str, subscription_name: str +) -> None: + assert_run_can_terminate( + _run_cli, + [topic], + service_url=service_url(pulsar_container), + subscription_name=subscription_name, + ) + + +@pytest.mark.integration +async def test_message_receive_not_callable( + app: MockApp, topic: str, pulsar_container: DockerContainer +) -> None: + client = pulsar.asyncio.Client(service_url(pulsar_container)) + producer = await client.create_producer(topic) + await producer.send(b"test") + await client.close() + + async with app.call() as (scope, receive, send): + with pytest.raises(RuntimeError, match="Receive should not be called"): + await receive() + await send({"type": "message.ack"}) diff --git a/packages/asyncfast/src/asyncfast/_channel.py b/packages/asyncfast/src/asyncfast/_channel.py index 102b55f3..7f0719e5 100644 --- a/packages/asyncfast/src/asyncfast/_channel.py +++ b/packages/asyncfast/src/asyncfast/_channel.py @@ -205,6 +205,8 @@ async def send_message(send: MessageSendCallable, message: Mapping[str, Any]) -> "headers": message["headers"], "payload": message.get("payload"), } + if (bindings := message.get("bindings")) is not None: + message_send_event["bindings"] = bindings await send(message_send_event) diff --git a/packages/asyncfast/src/asyncfast/bindings.py b/packages/asyncfast/src/asyncfast/bindings.py index 94efb0a3..12e0ac9e 100644 --- a/packages/asyncfast/src/asyncfast/bindings.py +++ b/packages/asyncfast/src/asyncfast/bindings.py @@ -25,6 +25,19 @@ def dump_value(self, value: Any, type_adapter: TypeAdapter[Any]) -> Any: return type_adapter.dump_json(value) +class PulsarKey(Binding): # type: ignore[misc] + __protocol__ = "pulsar" + __field_name__ = "key" + + def dump_value(self, value: Any, type_adapter: TypeAdapter[Any]) -> Any: + python = type_adapter.dump_python(value, mode="json") + if python is None: + return None + if isinstance(python, str): + return python + return type_adapter.dump_json(value).decode() + + class SqsDelaySeconds(Binding): # type: ignore[misc] __protocol__ = "sqs" __field_name__ = "delay_seconds" diff --git a/packages/asyncfast/tests_asyncfast/test_channel.py b/packages/asyncfast/tests_asyncfast/test_channel.py index baff6d9d..f86f5ea5 100644 --- a/packages/asyncfast/tests_asyncfast/test_channel.py +++ b/packages/asyncfast/tests_asyncfast/test_channel.py @@ -1,3 +1,4 @@ +from dataclasses import dataclass from typing import Annotated from typing import Any from typing import AsyncGenerator @@ -7,11 +8,13 @@ from unittest.mock import call from unittest.mock import Mock +from asyncfast import Message from asyncfast._channel import Depends from asyncfast._channel import get_channel from asyncfast._channel import Header from asyncfast._channel import MessageSender from asyncfast.bindings import KafkaKey +from asyncfast.bindings import PulsarKey async def test_payload_basic() -> None: @@ -239,6 +242,40 @@ async def func() -> AsyncGenerator[Mapping[str, Any], None]: ) +async def test_async_generator_func_with_binding() -> None: + send_mock = AsyncMock() + + @dataclass + class SendMessage(Message, address="send_topic"): + key: Annotated[str, PulsarKey()] + + async def func() -> AsyncGenerator[SendMessage, None]: + yield SendMessage(key="KEY-001") + + await get_channel(func, "channel")( + { + "type": "message", + "amgi": {"version": "2.0", "spec_version": "2.0"}, + "address": "channel", + "headers": [], + "payload": b"1", + }, + Mock(), + send_mock, + {}, + ) + + send_mock.assert_awaited_once_with( + { + "type": "message.send", + "address": "send_topic", + "headers": [], + "payload": None, + "bindings": {"pulsar": {"key": "KEY-001"}}, + } + ) + + async def test_sync_generator_func() -> None: send_mock = AsyncMock() diff --git a/packages/asyncfast/tests_asyncfast/test_message_object_bindings.py b/packages/asyncfast/tests_asyncfast/test_message_object_bindings.py index fc8fdf04..f5000eee 100644 --- a/packages/asyncfast/tests_asyncfast/test_message_object_bindings.py +++ b/packages/asyncfast/tests_asyncfast/test_message_object_bindings.py @@ -4,6 +4,7 @@ from asyncfast import Message from asyncfast.bindings import KafkaKey +from asyncfast.bindings import PulsarKey from asyncfast.bindings import SqsDelaySeconds from asyncfast.bindings import SqsMessageDeduplicationId from asyncfast.bindings import SqsMessageGroupId @@ -38,6 +39,48 @@ class Response(Message, address="response_channel"): } +def test_message_binding_pulsar_key(message_benchmark: MessageBenchmark) -> None: + @dataclass + class Response(Message, address="response_channel"): + key: Annotated[UUID, PulsarKey()] + + response = Response(key=UUID("ec5e9f87-c896-4fb1-b028-8352ef654e05")) + + assert message_benchmark(response) == { + "address": "response_channel", + "bindings": {"pulsar": {"key": "ec5e9f87-c896-4fb1-b028-8352ef654e05"}}, + "headers": [], + } + + +def test_message_binding_pulsar_key_int(message_benchmark: MessageBenchmark) -> None: + @dataclass + class Response(Message, address="response_channel"): + key: Annotated[int, PulsarKey()] + + response = Response(key=10) + + assert message_benchmark(response) == { + "address": "response_channel", + "bindings": {"pulsar": {"key": "10"}}, + "headers": [], + } + + +def test_message_binding_pulsar_key_none(message_benchmark: MessageBenchmark) -> None: + @dataclass + class Response(Message, address="response_channel"): + key: Annotated[str | None, PulsarKey()] + + response = Response(key=None) + + assert message_benchmark(response) == { + "address": "response_channel", + "bindings": {"pulsar": {"key": None}}, + "headers": [], + } + + def test_message_binding_sqs_delay_seconds(message_benchmark: MessageBenchmark) -> None: @dataclass class Response(Message, address="response_channel"): diff --git a/tox.ini b/tox.ini index 39110a84..b52a7f38 100644 --- a/tox.ini +++ b/tox.ini @@ -5,9 +5,9 @@ env_list = py314-asyncfast-pydantic212 py313-asyncfast-pydantic2{8-12} clean - py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-redis, amgi-sqs-event-source-mapping, asyncfast-avro, asyncfast-cli, pytest-amgi} + py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-pulsar, amgi-redis, amgi-sqs-event-source-mapping, asyncfast-avro, asyncfast-cli, pytest-amgi} py3{10-12}-asyncfast-pydantic2{0-12} - py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-redis, amgi-sqs-event-source-mapping, amgi-types, asyncfast, asyncfast-avro, asyncfast-cli, pytest-amgi}-import + py3{10-14}-{amgi-aio-pika, amgi-aiobotocore, amgi-aiokafka, amgi-cloudevents, amgi-common, amgi-kafka-event-source-mapping, amgi-nats, amgi-paho-mqtt, amgi-pulsar, amgi-redis, amgi-sqs-event-source-mapping, amgi-types, asyncfast, asyncfast-avro, asyncfast-cli, pytest-amgi}-import [testenv] runner = uv-venv-lock-runner @@ -104,6 +104,12 @@ commands = labels = amgi-paho-mqtt uv_sync_flags = --package=amgi-paho-mqtt +[testenv:py3{10-14}-amgi-pulsar] +commands = + {[testenv]commands} packages/amgi-pulsar +labels = amgi-pulsar +uv_sync_flags = --package=amgi-pulsar + [testenv:py3{10-14}-amgi-redis] commands = {[testenv]commands} packages/amgi-redis @@ -187,6 +193,14 @@ uv_sync_flags = --package=amgi-paho-mqtt --no-dev +[testenv:py3{10-14}-amgi-pulsar-import] +commands = + python -c "import amgi_pulsar" +labels = amgi-pulsar +uv_sync_flags = + --package=amgi-pulsar + --no-dev + [testenv:py3{10-14}-amgi-redis-import] commands = python -c "import amgi_redis" diff --git a/uv.lock b/uv.lock index 5a541b64..ae91c2bc 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 5 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.15'", @@ -20,6 +20,7 @@ members = [ "amgi-kafka-event-source-mapping", "amgi-nats", "amgi-paho-mqtt", + "amgi-pulsar", "amgi-redis", "amgi-sqs-event-source-mapping", "amgi-types", @@ -711,6 +712,45 @@ dev = [ { name = "testcontainers", extras = ["mqtt"], specifier = ">=4.13.0" }, ] +[[package]] +name = "amgi-pulsar" +version = "0.49.0" +source = { editable = "packages/amgi-pulsar" } +dependencies = [ + { name = "amgi-common" }, + { name = "amgi-types" }, + { name = "pulsar-client" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, +] + +[package.dev-dependencies] +dev = [ + { name = "pytest" }, + { name = "pytest-asyncio" }, + { name = "pytest-cov" }, + { name = "pytest-timeout" }, + { name = "test-utils" }, + { name = "testcontainers" }, +] + +[package.metadata] +requires-dist = [ + { name = "amgi-common", editable = "packages/amgi-common" }, + { name = "amgi-types", editable = "packages/amgi-types" }, + { name = "pulsar-client", specifier = ">=3.10.0" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'", specifier = ">=4.15.0" }, +] + +[package.metadata.requires-dev] +dev = [ + { name = "pytest", specifier = ">=8.4.1" }, + { name = "pytest-asyncio", specifier = ">=1.3.0" }, + { name = "pytest-cov", specifier = ">=7.0.0" }, + { name = "pytest-timeout", specifier = ">=2.4.0" }, + { name = "test-utils", editable = "packages/test-utils" }, + { name = "testcontainers", specifier = ">=4.13.0" }, +] + [[package]] name = "amgi-redis" version = "0.49.0" @@ -1687,6 +1727,7 @@ dependencies = [ { name = "amgi-aiokafka" }, { name = "amgi-nats" }, { name = "amgi-paho-mqtt" }, + { name = "amgi-pulsar" }, { name = "asyncfast" }, ] @@ -1696,6 +1737,7 @@ requires-dist = [ { name = "amgi-aiokafka", editable = "packages/amgi-aiokafka" }, { name = "amgi-nats", editable = "packages/amgi-nats" }, { name = "amgi-paho-mqtt", editable = "packages/amgi-paho-mqtt" }, + { name = "amgi-pulsar", editable = "packages/amgi-pulsar" }, { name = "asyncfast", editable = "packages/asyncfast" }, ] @@ -3059,6 +3101,46 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/3a/ed/1cdcab6ba3d6ab7feca11fc14f0eeea80755bb53ef4e892079f31b10a25f/propcache-0.5.2-py3-none-any.whl", hash = "sha256:be1ddfcbb376e3de5d2e2db1d58d6d67463e6b4f9f040c000de8e300295465fe", size = 14036, upload-time = "2026-05-08T21:02:10.673Z" }, ] +[[package]] +name = "pulsar-client" +version = "3.13.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "certifi" }, +] +wheels = [ + { url = "https://files.pythonhosted.org/packages/58/28/872c9184a86219e10a0d2cf015586cfd2b2afaf96f2ab33f659d0a611f27/pulsar_client-3.13.0-cp310-cp310-macosx_13_0_universal2.whl", hash = "sha256:7b510a8be6ce5fc6dffb772b80baa00900a4b8a7342be38023fcd38cc7582ad0", size = 10726068, upload-time = "2026-07-02T09:19:52.79Z" }, + { url = "https://files.pythonhosted.org/packages/c0/de/8aa316e5e73ce0edd2bdc98bf42f990c23ce00aaef440a3d92dd188a23c5/pulsar_client-3.13.0-cp310-cp310-manylinux_2_28_aarch64.whl", hash = "sha256:61c4b8f478cbe843a9793f3e145709efb407ca6979a4b23b8ee7abd585770720", size = 7633567, upload-time = "2026-07-02T09:20:01.521Z" }, + { url = "https://files.pythonhosted.org/packages/2a/3f/0a333f57cce16cdcfb90d36ad6a89b05847748090f04adae203c17451a7e/pulsar_client-3.13.0-cp310-cp310-manylinux_2_28_x86_64.whl", hash = "sha256:8e4bd5a7237829b9fb670f477e54c16ff6dbfde363b40f252e957a50c1fba0d9", size = 7626302, upload-time = "2026-07-02T09:20:11.491Z" }, + { url = "https://files.pythonhosted.org/packages/78/51/5fa17080ad0c139456c7bdea1439d13b74f79d9f9e78b9cb8cc57a91240b/pulsar_client-3.13.0-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:bc0337146a525837c5bb945c50241427806f8200d96ad839c8b6b08cd631c987", size = 8154911, upload-time = "2026-07-02T09:20:15.435Z" }, + { url = "https://files.pythonhosted.org/packages/00/9c/b793faff94749545300fb3c7e8b1aad0ad3afe718d45351ab83e00ef1978/pulsar_client-3.13.0-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:6410dacd0c2cef0a0c5539acea448d487a1824ff1c69230ae8d366762938020e", size = 7990679, upload-time = "2026-07-02T09:20:19.506Z" }, + { url = "https://files.pythonhosted.org/packages/e0/59/38d9e19e1b3c2f28e5fb21cc4c97f438eaaa213f14f5f5cfe7dd99a7e923/pulsar_client-3.13.0-cp310-cp310-win_amd64.whl", hash = "sha256:7de4a2541059e54c1f6cdf5d961306a6fc7f50cd415425b185b653d7dfa5ec39", size = 3931880, upload-time = "2026-07-02T09:20:22.745Z" }, + { url = "https://files.pythonhosted.org/packages/43/d1/588228438017ceee45a7fcf83d0dd8cd5d80a628b8d0a8ff8dbe12baf06d/pulsar_client-3.13.0-cp311-cp311-macosx_13_0_universal2.whl", hash = "sha256:f2eb3afc2e1c002bc74eadd3f3ba49fb4c337d59e83f0861c79a901b62098ec0", size = 10730053, upload-time = "2026-07-02T09:20:26.754Z" }, + { url = "https://files.pythonhosted.org/packages/0d/c0/b35d12a40a5c53625b7b7a5fd4b237b8acf274e7485a5b47157c490b8e68/pulsar_client-3.13.0-cp311-cp311-manylinux_2_28_aarch64.whl", hash = "sha256:7c87a6944d5531b59b556b6078b9a0fb6d2dceff7a2edc8875bba43ed37c7f6a", size = 7635458, upload-time = "2026-07-02T09:20:30.81Z" }, + { url = "https://files.pythonhosted.org/packages/01/95/360f78e5b16c7f96825274483c9a3e6d37c8878c2727592b4e21ecdb6624/pulsar_client-3.13.0-cp311-cp311-manylinux_2_28_x86_64.whl", hash = "sha256:f958a6149b28cfe354d33f2af229e2257d0bc4e1f195eefbd4cbae5d7fd7961d", size = 7630025, upload-time = "2026-07-02T09:20:34.572Z" }, + { url = "https://files.pythonhosted.org/packages/c1/61/d10ad860a0d8f899ce450e0a462b6e83ee5839ed29030f920652989d828a/pulsar_client-3.13.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:f3f769e3128366df63296d7caa4934721f70083def652cb34fb0b55538ba325b", size = 8157148, upload-time = "2026-07-02T09:20:39.407Z" }, + { url = "https://files.pythonhosted.org/packages/b3/72/6111ebf31cc4d096343a84f5d636bea0dcc940a9e8a751a74dae7871b539/pulsar_client-3.13.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:8ce5c596a82b4b43d677d7ba82e6f8a553c6ea50ac7cb42bde986868dc851437", size = 7992879, upload-time = "2026-07-02T09:20:43.984Z" }, + { url = "https://files.pythonhosted.org/packages/af/9c/149ddd39dead27d9b6ac03c8b4ee588798d9b3cc297d11e44922b3c9c824/pulsar_client-3.13.0-cp311-cp311-win_amd64.whl", hash = "sha256:3111678f649afb78ec4d807d7fd352836912537a9cd1ca2ad43c1f992eb83301", size = 3932604, upload-time = "2026-07-02T09:20:47.737Z" }, + { url = "https://files.pythonhosted.org/packages/f6/62/1ab6eb1d1e8f9fab917e691655f9589279ec643aaaeb650ba4f134003fcb/pulsar_client-3.13.0-cp312-cp312-macosx_13_0_universal2.whl", hash = "sha256:f2f24bacf9ad7999a2c9813db68b438b776a2bc5440ed1bbb9cc62acbfc4f85d", size = 10753952, upload-time = "2026-07-02T09:21:00.035Z" }, + { url = "https://files.pythonhosted.org/packages/96/4a/0bf1648b69ab8a6b84079c42b38744b5e815dc80dcc5b24ef0159063fc9b/pulsar_client-3.13.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:03244a8af1886591ee56afb1228b85f8b678f980464a2847ebb9a152d8b5e36a", size = 7641537, upload-time = "2026-07-02T09:21:04.889Z" }, + { url = "https://files.pythonhosted.org/packages/ca/1c/c62cd2dc12c73383ae74e835c8057baaead77097bf897f1c802a4f8a4d91/pulsar_client-3.13.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:863fd7b8cdc162f3cc0ea7217c37604c66a7bc9bb5cd554566c5d07f89a50e5e", size = 7642737, upload-time = "2026-07-02T09:21:37.133Z" }, + { url = "https://files.pythonhosted.org/packages/6e/3b/c0049394b7caa42168dfce10e84bba219a3b1d9466ef9a3fdab65455e9c6/pulsar_client-3.13.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:da4fe1a58e81d1a3131aa8edb44eac91c9931ad6f8e21b2e3409cb8ee8e9173d", size = 8158732, upload-time = "2026-07-02T09:21:41.617Z" }, + { url = "https://files.pythonhosted.org/packages/26/74/b82320196cfde96304110c4ea69dff5b9e86e3f9d5daae6f33baa838e89d/pulsar_client-3.13.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:55a060c9a7b58e4daa80aa8126e340d378e0f55e5616523574d3144f5a649347", size = 8003510, upload-time = "2026-07-02T09:21:45.847Z" }, + { url = "https://files.pythonhosted.org/packages/cd/1c/471b4cdc329e01b42f35145c678e2ac1f375c9de3e718a8f15d931af48c1/pulsar_client-3.13.0-cp312-cp312-win_amd64.whl", hash = "sha256:f5d5dcfc45f73e05db754f2372d278c1183a674c6e2b91416a4f6c829911365b", size = 3933736, upload-time = "2026-07-02T09:21:51.941Z" }, + { url = "https://files.pythonhosted.org/packages/2b/f7/3cfe69993054e94e3e4a4de39d135aef70be12bfc9282d37ceda89b526d8/pulsar_client-3.13.0-cp313-cp313-macosx_13_0_universal2.whl", hash = "sha256:0e76acbc9ca5515650e940a0bb10c45860aa14707bb755dde01d0eb011606d77", size = 10755334, upload-time = "2026-07-02T09:21:56.719Z" }, + { url = "https://files.pythonhosted.org/packages/eb/5f/39091328c997125706518da1860d7410b5cc1a3bef5846a489ec26c185c7/pulsar_client-3.13.0-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:7cc1cb49a29a85b8469b42d3a088c0fde639887e0b06120d92c34a9412296528", size = 7643372, upload-time = "2026-07-02T09:22:00.417Z" }, + { url = "https://files.pythonhosted.org/packages/df/f8/e78b52cd028ac912a23b39cc0d820e6783cb4ed71df25cb770cb1ce1ffe4/pulsar_client-3.13.0-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:afa143cbadbe0e13c3afcc3925442ee67a3b611d28110462fbcd37fc95cdcd94", size = 7641926, upload-time = "2026-07-02T09:22:03.761Z" }, + { url = "https://files.pythonhosted.org/packages/93/19/6748ee7b20de8b74f23b61d7664849492c46af1fd269d4d4ac621c8ec8a7/pulsar_client-3.13.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:2cbcae67dc46c3b85288984a51ee0c194f8898cba54fc31b08eeffef8b113a30", size = 8161502, upload-time = "2026-07-02T09:22:07.776Z" }, + { url = "https://files.pythonhosted.org/packages/6e/d1/4d2c735a8a2988b710911d7b0ed371bac60da30747ad4827abe1ba73b9e6/pulsar_client-3.13.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:21c241e7a7f359928f677a97bf7e9b31723f575cc9adc0e09350f5e5a5c1be53", size = 8004562, upload-time = "2026-07-02T09:22:11.281Z" }, + { url = "https://files.pythonhosted.org/packages/e9/8d/75405f5fc2541f295fe74c999e5239332540a27590800d00173549e05bdb/pulsar_client-3.13.0-cp313-cp313-win_amd64.whl", hash = "sha256:56109a622fd5373f7971a0e052732dbbcdd0030b7e69c06377cd49763903df20", size = 3933831, upload-time = "2026-07-02T09:22:14.003Z" }, + { url = "https://files.pythonhosted.org/packages/fa/d7/d1fe1e8e6e37bbc0454e5eaddf354c69ecf5be00bfbfea6c6e4015c64109/pulsar_client-3.13.0-cp314-cp314-macosx_13_0_universal2.whl", hash = "sha256:cf62d0ea76b3dee86536115b9e5b5ecc8087920196afea23fd04d8b38ff0c275", size = 10742295, upload-time = "2026-07-02T09:22:21.381Z" }, + { url = "https://files.pythonhosted.org/packages/26/9d/29754da5168da73597071656f344efa91cde6edfba8e45426c70b90b83b2/pulsar_client-3.13.0-cp314-cp314-manylinux_2_28_aarch64.whl", hash = "sha256:6060e6856dad29231aba5464941c3154f5ef3175d7d8f98c9edf994df6cd8820", size = 7644768, upload-time = "2026-07-02T09:22:25.083Z" }, + { url = "https://files.pythonhosted.org/packages/12/97/9e878f005cdbaab43ae9d6ac1600fc139ed83825dfb55ce07c7cdb0a796a/pulsar_client-3.13.0-cp314-cp314-manylinux_2_28_x86_64.whl", hash = "sha256:edd8ffd12cf3dd4b72b080e815652f0c2f18045b5e9b779e63e43caad470e834", size = 7642379, upload-time = "2026-07-02T09:22:28.888Z" }, + { url = "https://files.pythonhosted.org/packages/80/88/ba0dee2c9e93188fcba50b19b879242cc7ba4620557a155d04d81682e67c/pulsar_client-3.13.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:25d56f51895a01ae885e71ae3f65369496462139bb506450fa11c1c28941177a", size = 8163723, upload-time = "2026-07-02T09:22:33.643Z" }, + { url = "https://files.pythonhosted.org/packages/17/26/1bfc34008bfb4007f12707b13fe0f1cb1c6ce2f493d6ca4d0b99c8a8d4b9/pulsar_client-3.13.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:d2f65f2e697d20dbef2a9f04763bada0e20ff34e13650204f95d9acbfed1ec1e", size = 8005834, upload-time = "2026-07-02T09:22:37.846Z" }, + { url = "https://files.pythonhosted.org/packages/4f/ae/b1e0ee8dd8a0bb14d4ef3e48474886fc4d77036b3c7fe0346805652f071a/pulsar_client-3.13.0-cp314-cp314-win_amd64.whl", hash = "sha256:43aa8b5532aa4ca9417342a4c5a0fcf9e03f9b1a64177be5a2420dea08547685", size = 4057946, upload-time = "2026-07-02T09:22:40.591Z" }, +] + [[package]] name = "py-cpuinfo" version = "9.0.0"