diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml new file mode 100644 index 0000000..2f87d94 --- /dev/null +++ b/.github/FUNDING.yml @@ -0,0 +1 @@ +github: encode diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..3ea8eef --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,29 @@ +name: Publish + +on: + push: + tags: + - "*" + +jobs: + publish: + name: "Publish release" + runs-on: "ubuntu-latest" + + steps: + - uses: "actions/checkout@v3" + - uses: "actions/setup-python@v4" + with: + python-version: "3.10" + + - name: "Install dependencies" + run: "scripts/install" + + - name: "Build package & docs" + run: "scripts/build" + + - name: "Publish to PyPI & deploy docs" + run: "scripts/publish" + env: + TWINE_USERNAME: __token__ + TWINE_PASSWORD: ${{ secrets.PYPI_TOKEN }} diff --git a/.github/workflows/test-suite.yml b/.github/workflows/test-suite.yml index 458bcad..81db761 100644 --- a/.github/workflows/test-suite.yml +++ b/.github/workflows/test-suite.yml @@ -14,7 +14,7 @@ jobs: strategy: matrix: - python-version: ["3.7", "3.8", "3.9"] + python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] services: zookeeper: @@ -66,4 +66,4 @@ jobs: - name: "Run tests" run: "scripts/test" - name: "Enforce coverage" - run: "scripts/coverage" \ No newline at end of file + run: "scripts/coverage" diff --git a/.gitignore b/.gitignore index 7b5d431..013870b 100644 --- a/.gitignore +++ b/.gitignore @@ -3,5 +3,7 @@ test.db .coverage .pytest_cache/ .mypy_cache/ -starlette.egg-info/ +*.egg-info/ venv/ +build/ +dist/ diff --git a/README.md b/README.md index 61709ac..959b34d 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Broadcaster helps you develop realtime streaming functionality by providing a simple broadcast API onto a number of different backend services. -It currently supports [Redis PUB/SUB](https://redis.io/topics/pubsub), [Apache Kafka](https://kafka.apache.org/), and [Postgres LISTEN/NOTIFY](https://www.postgresql.org/docs/current/sql-notify.html), plus a simple in-memory backend, that you can use for local development or during testing. +It currently supports [Redis PUB/SUB](https://redis.io/topics/pubsub), [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/), [Apache Kafka](https://kafka.apache.org/), and [Postgres LISTEN/NOTIFY](https://www.postgresql.org/docs/current/sql-notify.html), plus a simple in-memory backend, that you can use for local development or during testing. WebSockets Demo @@ -14,9 +14,9 @@ Here's a complete example of the backend code for a simple websocket chat app: ```python # Requires: `starlette`, `uvicorn`, `jinja2` # Run with `uvicorn example:app` +import anyio from broadcaster import Broadcast from starlette.applications import Starlette -from starlette.concurrency import run_until_first_complete from starlette.routing import Route, WebSocketRoute from starlette.templating import Jinja2Templates @@ -33,10 +33,15 @@ async def homepage(request): async def chatroom_ws(websocket): await websocket.accept() - await run_until_first_complete( - (chatroom_ws_receiver, {"websocket": websocket}), - (chatroom_ws_sender, {"websocket": websocket}), - ) + + async with anyio.create_task_group() as task_group: + # run until first is complete + async def run_chatroom_ws_receiver() -> None: + await chatroom_ws_receiver(websocket=websocket) + task_group.cancel_scope.cancel() + + task_group.start_soon(run_chatroom_ws_receiver) + await chatroom_ws_sender(websocket) async def chatroom_ws_receiver(websocket): @@ -65,7 +70,7 @@ The HTML template for the front end [is available here](https://github.com/encod ## Requirements -Python 3.7+ +Python 3.8+ ## Installation @@ -78,19 +83,42 @@ Python 3.7+ * `Broadcast('memory://')` * `Broadcast("redis://localhost:6379")` +* `Broadcast("redis-stream://localhost:6379")` * `Broadcast("postgres://localhost:5432/broadcaster")` * `Broadcast("kafka://localhost:9092")` + +### Using custom backends + +You can create your own backend and use it with `broadcaster`. +To do that you need to create a class which extends from `BroadcastBackend` +and pass it to the `broadcaster` via `backend` argument. + +```python +from broadcaster import Broadcaster, BroadcastBackend + +class MyBackend(BroadcastBackend): + +broadcaster = Broadcaster(backend=MyBackend()) +``` + ## Where next? At the moment `broadcaster` is in Alpha, and should be considered a working design document. The API should be considered subject to change. If you *do* want to use Broadcaster in its current -state, make sure to strictly pin your requirements to `broadcaster==0.2.0`. +state, make sure to strictly pin your requirements to `broadcaster==0.3.0`. To be more capable we'd really want to add some additional backends, provide API support for reading recent event history from persistent stores, and provide a serialization/deserialization API... * Serialization / deserialization to support broadcasting structured data. -* Backends for Redis Streams, Apache Kafka, and RabbitMQ. +* A backend for RabbitMQ. * Add support for `subscribe('chatroom', history=100)` for backends which provide persistence. (Redis Streams, Apache Kafka) This will allow applications to subscribe to channel updates, while also being given an initial window onto the most recent events. We *might* also want to support some basic paging operations, to allow applications to scan back in the event history. * Support for pattern subscribes in backends that support it. + +## Third Party Packages + +### MQTT backend +[Gist](https://gist.github.com/alex-oleshkevich/68411a0e7ad24d53afd28c3fa5da468c) + +Integrates MQTT with Broadcaster diff --git a/broadcaster/__init__.py b/broadcaster/__init__.py index edc56d6..3ee97d2 100644 --- a/broadcaster/__init__.py +++ b/broadcaster/__init__.py @@ -1,4 +1,5 @@ from ._base import Broadcast, Event +from .backends.base import BroadcastBackend -__version__ = "0.2.0" -__all__ = ["Broadcast", "Event"] +__version__ = "0.3.2" +__all__ = ["Broadcast", "Event", "BroadcastBackend"] diff --git a/broadcaster/_backends/kafka.py b/broadcaster/_backends/kafka.py deleted file mode 100644 index a3df086..0000000 --- a/broadcaster/_backends/kafka.py +++ /dev/null @@ -1,39 +0,0 @@ -import asyncio -import typing -from urllib.parse import urlparse - -from aiokafka import AIOKafkaConsumer, AIOKafkaProducer - -from .._base import Event -from .base import BroadcastBackend - - -class KafkaBackend(BroadcastBackend): - def __init__(self, url: str): - self._servers = [urlparse(url).netloc] - self._consumer_channels: typing.Set = set() - - async def connect(self) -> None: - loop = asyncio.get_event_loop() - self._producer = AIOKafkaProducer(loop=loop, bootstrap_servers=self._servers) - self._consumer = AIOKafkaConsumer(loop=loop, bootstrap_servers=self._servers) - await self._producer.start() - await self._consumer.start() - - async def disconnect(self) -> None: - await self._producer.stop() - await self._consumer.stop() - - async def subscribe(self, channel: str) -> None: - self._consumer_channels.add(channel) - self._consumer.subscribe(topics=self._consumer_channels) - - async def unsubscribe(self, channel: str) -> None: - await self._consumer.unsubscribe() - - async def publish(self, channel: str, message: typing.Any) -> None: - await self._producer.send_and_wait(channel, message.encode("utf8")) - - async def next_published(self) -> Event: - message = await self._consumer.getone() - return Event(channel=message.topic, message=message.value.decode("utf8")) diff --git a/broadcaster/_backends/redis.py b/broadcaster/_backends/redis.py deleted file mode 100644 index b545bad..0000000 --- a/broadcaster/_backends/redis.py +++ /dev/null @@ -1,36 +0,0 @@ -import typing -from urllib.parse import urlparse - -import asyncio_redis - -from .._base import Event -from .base import BroadcastBackend - - -class RedisBackend(BroadcastBackend): - def __init__(self, url: str): - parsed_url = urlparse(url) - self._host = parsed_url.hostname or "localhost" - self._port = parsed_url.port or 6379 - - async def connect(self) -> None: - self._pub_conn = await asyncio_redis.Connection.create(self._host, self._port) - self._sub_conn = await asyncio_redis.Connection.create(self._host, self._port) - self._subscriber = await self._sub_conn.start_subscribe() - - async def disconnect(self) -> None: - self._pub_conn.close() - self._sub_conn.close() - - async def subscribe(self, channel: str) -> None: - await self._subscriber.subscribe([channel]) - - async def unsubscribe(self, channel: str) -> None: - await self._subscriber.unsubscribe([channel]) - - async def publish(self, channel: str, message: typing.Any) -> None: - await self._pub_conn.publish(channel, message) - - async def next_published(self) -> Event: - message = await self._subscriber.next_published() - return Event(channel=message.channel, message=message.value) diff --git a/broadcaster/_base.py b/broadcaster/_base.py index c58cb1d..a63b22b 100644 --- a/broadcaster/_base.py +++ b/broadcaster/_base.py @@ -1,8 +1,13 @@ +from __future__ import annotations + import asyncio from contextlib import asynccontextmanager -from typing import Any, AsyncGenerator, AsyncIterator, Dict, Optional +from typing import TYPE_CHECKING, Any, AsyncGenerator, AsyncIterator, cast from urllib.parse import urlparse +if TYPE_CHECKING: # pragma: no cover + from broadcaster.backends.base import BroadcastBackend + class Event: def __init__(self, channel: str, message: str) -> None: @@ -10,11 +15,7 @@ def __init__(self, channel: str, message: str) -> None: self.message = message def __eq__(self, other: object) -> bool: - return ( - isinstance(other, Event) - and self.channel == other.channel - and self.message == other.message - ) + return isinstance(other, Event) and self.channel == other.channel and self.message == other.message def __repr__(self) -> str: return f"Event(channel={self.channel!r}, message={self.message!r})" @@ -25,33 +26,40 @@ class Unsubscribed(Exception): class Broadcast: - def __init__(self, url: str): - from broadcaster._backends.base import BroadcastBackend + def __init__(self, url: str | None = None, *, backend: BroadcastBackend | None = None) -> None: + assert url or backend, "Either `url` or `backend` must be provided." + self._backend = backend or self._create_backend(cast(str, url)) + self._subscribers: dict[str, set[asyncio.Queue[Event | None]]] = {} + def _create_backend(self, url: str) -> BroadcastBackend: parsed_url = urlparse(url) - self._backend: BroadcastBackend - self._subscribers: Dict[str, Any] = {} - if parsed_url.scheme == "redis": - from broadcaster._backends.redis import RedisBackend + if parsed_url.scheme in ("redis", "rediss"): + from broadcaster.backends.redis import RedisBackend + + return RedisBackend(url) + + elif parsed_url.scheme == "redis-stream": + from broadcaster.backends.redis import RedisStreamBackend - self._backend = RedisBackend(url) + return RedisStreamBackend(url) elif parsed_url.scheme in ("postgres", "postgresql"): - from broadcaster._backends.postgres import PostgresBackend + from broadcaster.backends.postgres import PostgresBackend - self._backend = PostgresBackend(url) + return PostgresBackend(url) if parsed_url.scheme == "kafka": - from broadcaster._backends.kafka import KafkaBackend + from broadcaster.backends.kafka import KafkaBackend - self._backend = KafkaBackend(url) + return KafkaBackend(url) elif parsed_url.scheme == "memory": - from broadcaster._backends.memory import MemoryBackend + from broadcaster.backends.memory import MemoryBackend - self._backend = MemoryBackend(url) + return MemoryBackend(url) + raise ValueError(f"Unsupported backend: {parsed_url.scheme}") - async def __aenter__(self) -> "Broadcast": + async def __aenter__(self) -> Broadcast: await self.connect() return self @@ -79,31 +87,30 @@ async def publish(self, channel: str, message: Any) -> None: await self._backend.publish(channel, message) @asynccontextmanager - async def subscribe(self, channel: str) -> AsyncIterator["Subscriber"]: - queue: asyncio.Queue = asyncio.Queue() + async def subscribe(self, channel: str) -> AsyncIterator[Subscriber]: + queue: asyncio.Queue[Event | None] = asyncio.Queue() try: if not self._subscribers.get(channel): await self._backend.subscribe(channel) - self._subscribers[channel] = set([queue]) + self._subscribers[channel] = {queue} else: self._subscribers[channel].add(queue) yield Subscriber(queue) - + finally: self._subscribers[channel].remove(queue) if not self._subscribers.get(channel): del self._subscribers[channel] await self._backend.unsubscribe(channel) - finally: await queue.put(None) class Subscriber: - def __init__(self, queue: asyncio.Queue) -> None: + def __init__(self, queue: asyncio.Queue[Event | None]) -> None: self._queue = queue - async def __aiter__(self) -> Optional[AsyncGenerator]: + async def __aiter__(self) -> AsyncGenerator[Event | None, None]: try: while True: yield await self.get() diff --git a/broadcaster/_backends/__init__.py b/broadcaster/backends/__init__.py similarity index 100% rename from broadcaster/_backends/__init__.py rename to broadcaster/backends/__init__.py diff --git a/broadcaster/_backends/base.py b/broadcaster/backends/base.py similarity index 83% rename from broadcaster/_backends/base.py rename to broadcaster/backends/base.py index 1c65b86..7017df3 100644 --- a/broadcaster/_backends/base.py +++ b/broadcaster/backends/base.py @@ -13,10 +13,10 @@ async def connect(self) -> None: async def disconnect(self) -> None: raise NotImplementedError() - async def subscribe(self, group: str) -> None: + async def subscribe(self, channel: str) -> None: raise NotImplementedError() - async def unsubscribe(self, group: str) -> None: + async def unsubscribe(self, channel: str) -> None: raise NotImplementedError() async def publish(self, channel: str, message: Any) -> None: diff --git a/broadcaster/backends/kafka.py b/broadcaster/backends/kafka.py new file mode 100644 index 0000000..f09dca1 --- /dev/null +++ b/broadcaster/backends/kafka.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import asyncio +import typing +from urllib.parse import urlparse + +from aiokafka import AIOKafkaConsumer, AIOKafkaProducer + +from .._base import Event +from .base import BroadcastBackend + + +class KafkaBackend(BroadcastBackend): + def __init__(self, urls: str | list[str]) -> None: + urls = [urls] if isinstance(urls, str) else urls + self._servers = [urlparse(url).netloc for url in urls] + self._consumer_channels: set[str] = set() + self._ready = asyncio.Event() + + async def connect(self) -> None: + self._producer = AIOKafkaProducer(bootstrap_servers=self._servers) # pyright: ignore + self._consumer = AIOKafkaConsumer(bootstrap_servers=self._servers) # pyright: ignore + await self._producer.start() + await self._consumer.start() + + async def disconnect(self) -> None: + await self._producer.stop() + await self._consumer.stop() + + async def subscribe(self, channel: str) -> None: + self._consumer_channels.add(channel) + self._consumer.subscribe(topics=self._consumer_channels) + await self._wait_for_assignment() + + async def unsubscribe(self, channel: str) -> None: + self._consumer.unsubscribe() + + async def publish(self, channel: str, message: typing.Any) -> None: + await self._producer.send_and_wait(channel, message.encode("utf8")) + + async def next_published(self) -> Event: + await self._ready.wait() + message = await self._consumer.getone() + value = message.value + + # for type compatibility: + # we declare Event.message as str, so convert None to empty string + if value is None: + value = b"" + return Event(channel=message.topic, message=value.decode("utf8")) + + async def _wait_for_assignment(self) -> None: + """Wait for the consumer to be assigned to the partition.""" + while not self._consumer.assignment(): + await asyncio.sleep(0.001) + + self._ready.set() diff --git a/broadcaster/_backends/memory.py b/broadcaster/backends/memory.py similarity index 84% rename from broadcaster/_backends/memory.py rename to broadcaster/backends/memory.py index 5a9fa53..bfd0c44 100644 --- a/broadcaster/_backends/memory.py +++ b/broadcaster/backends/memory.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import asyncio import typing @@ -7,10 +9,10 @@ class MemoryBackend(BroadcastBackend): def __init__(self, url: str): - self._subscribed: typing.Set = set() + self._subscribed: set[str] = set() async def connect(self) -> None: - self._published: asyncio.Queue = asyncio.Queue() + self._published: asyncio.Queue[Event] = asyncio.Queue() async def disconnect(self) -> None: pass diff --git a/broadcaster/_backends/postgres.py b/broadcaster/backends/postgres.py similarity index 94% rename from broadcaster/_backends/postgres.py rename to broadcaster/backends/postgres.py index 47ef4f6..7769962 100644 --- a/broadcaster/_backends/postgres.py +++ b/broadcaster/backends/postgres.py @@ -13,7 +13,7 @@ def __init__(self, url: str): async def connect(self) -> None: self._conn = await asyncpg.connect(self._url) - self._listen_queue: asyncio.Queue = asyncio.Queue() + self._listen_queue: asyncio.Queue[Event] = asyncio.Queue() async def disconnect(self) -> None: await self._conn.close() diff --git a/broadcaster/backends/redis.py b/broadcaster/backends/redis.py new file mode 100644 index 0000000..effb166 --- /dev/null +++ b/broadcaster/backends/redis.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +import asyncio +import typing + +from redis import asyncio as redis + +from .._base import Event +from .base import BroadcastBackend + + +class RedisBackend(BroadcastBackend): + _conn: redis.Redis + + def __init__(self, url: str | None = None, *, conn: redis.Redis | None = None): + if url is None: + assert conn is not None, "conn must be provided if url is not" + self._conn = conn + else: + self._conn = redis.Redis.from_url(url) + + self._pubsub = self._conn.pubsub() + self._ready = asyncio.Event() + self._queue: asyncio.Queue[Event] = asyncio.Queue() + self._listener: asyncio.Task[None] | None = None + + async def connect(self) -> None: + self._listener = asyncio.create_task(self._pubsub_listener()) + await self._pubsub.connect() # type: ignore[no-untyped-call] + + async def disconnect(self) -> None: + await self._pubsub.aclose() # type: ignore[no-untyped-call] + await self._conn.aclose() + if self._listener is not None: + self._listener.cancel() + + async def subscribe(self, channel: str) -> None: + self._ready.set() + await self._pubsub.subscribe(channel) + + async def unsubscribe(self, channel: str) -> None: + await self._pubsub.unsubscribe(channel) + + async def publish(self, channel: str, message: typing.Any) -> None: + await self._conn.publish(channel, message) + + async def next_published(self) -> Event: + return await self._queue.get() + + async def _pubsub_listener(self) -> None: + # redis-py does not listen to the pubsub connection if there are no channels subscribed + # so we need to wait until the first channel is subscribed to start listening + while True: + await self._ready.wait() + async for message in self._pubsub.listen(): + if message["type"] == "message": + event = Event( + channel=message["channel"].decode(), + message=message["data"].decode(), + ) + await self._queue.put(event) + + # when no channel subscribed, clear the event. + # And then in next loop, event will blocked again until + # the new channel subscribed.Now asyncio.Task will not exit again. + self._ready.clear() + + +StreamMessageType = typing.Tuple[bytes, typing.Tuple[typing.Tuple[bytes, typing.Dict[bytes, bytes]]]] + + +class RedisStreamBackend(BroadcastBackend): + def __init__(self, url: str): + url = url.replace("redis-stream", "redis", 1) + self.streams: dict[bytes | str | memoryview, int | bytes | str | memoryview] = {} + self._ready = asyncio.Event() + self._producer = redis.Redis.from_url(url) + self._consumer = redis.Redis.from_url(url) + + async def connect(self) -> None: + pass + + async def disconnect(self) -> None: + await self._producer.aclose() + await self._consumer.aclose() + + async def subscribe(self, channel: str) -> None: + try: + info = await self._consumer.xinfo_stream(channel) + last_id = info["last-generated-id"] + except redis.ResponseError: + last_id = "0" + self.streams[channel] = last_id + self._ready.set() + + async def unsubscribe(self, channel: str) -> None: + self.streams.pop(channel, None) + + async def publish(self, channel: str, message: typing.Any) -> None: + await self._producer.xadd(channel, {"message": message}) + + async def wait_for_messages(self) -> list[StreamMessageType]: + await self._ready.wait() + messages = None + while not messages: + messages = await self._consumer.xread(self.streams, count=1, block=100) + return messages + + async def next_published(self) -> Event: + messages = await self.wait_for_messages() + stream, events = messages[0] + _msg_id, message = events[0] + self.streams[stream.decode("utf-8")] = _msg_id.decode("utf-8") + return Event( + channel=stream.decode("utf-8"), + message=message.get(b"message", b"").decode("utf-8"), + ) diff --git a/broadcaster/py.typed b/broadcaster/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/example/app.py b/example/app.py index 9b57e44..a201221 100644 --- a/example/app.py +++ b/example/app.py @@ -1,8 +1,8 @@ import os -import uvicorn + +import anyio from starlette.applications import Starlette -from starlette.concurrency import run_until_first_complete from starlette.routing import Route, WebSocketRoute from starlette.templating import Jinja2Templates @@ -22,10 +22,15 @@ async def homepage(request): async def chatroom_ws(websocket): await websocket.accept() - await run_until_first_complete( - (chatroom_ws_receiver, {"websocket": websocket}), - (chatroom_ws_sender, {"websocket": websocket}), - ) + + async with anyio.create_task_group() as task_group: + # run until first is complete + async def run_chatroom_ws_receiver() -> None: + await chatroom_ws_receiver(websocket=websocket) + task_group.cancel_scope.cancel() + + task_group.start_soon(run_chatroom_ws_receiver) + await chatroom_ws_sender(websocket) async def chatroom_ws_receiver(websocket): diff --git a/example/requirements.txt b/example/requirements.txt index 2b7b4ad..44835e5 100644 --- a/example/requirements.txt +++ b/example/requirements.txt @@ -1,4 +1,5 @@ uvicorn +websockets starlette jinja2 broadcaster[redis,postgres,kafka] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c4e8036 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,81 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "broadcaster" +dynamic = ["version"] +description = "Simple broadcast channels." +readme = "README.md" +license = "BSD-3-Clause" +requires-python = ">=3.8" +authors = [ + { name = "Tom Christie", email = "tom@tomchristie.com" }, +] +classifiers = [ + "Development Status :: 3 - Alpha", + "Environment :: Web Environment", + "Intended Audience :: Developers", + "License :: OSI Approved :: BSD License", + "Operating System :: OS Independent", + "Topic :: Internet :: WWW/HTTP", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", +] +dependencies = [ + "anyio>=3.4.0,<5", + "typing_extensions>=3.10.0; python_version < '3.10'", +] + +[project.optional-dependencies] +redis = ["redis"] +postgres = ["asyncpg"] +kafka = ["aiokafka"] +test = ["pytest", "pytest-asyncio"] + +[project.urls] +Homepage = "https://github.com/encode/broadcaster" + +[tool.hatch.version] +path = "broadcaster/__init__.py" + +[tool.hatch.build.targets.sdist] +include = [ + "/broadcaster", +] + +[tool.ruff] +line-length = 120 + +[tool.ruff.lint] +select = ["E", "F", "I", "FA", "UP"] + +[tool.ruff.lint.isort] +combine-as-imports = true + +[tool.mypy] +strict = true +python_version = "3.8" +disallow_untyped_defs = true +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "tests.*" +disallow_untyped_defs = false +check_untyped_defs = true + +[tool.pytest.ini_options] +addopts = "-rxXs" +markers = "copied_from(source, changes=None): mark test as copied from somewhere else, along with a description of changes made to accodomate e.g. our test setup" + +[tool.coverage.run] +source_pkgs = ["broadcaster", "tests"] + +[tool.coverage.report] +fail_under = 78 +show_missing = true +skip_covered = true diff --git a/requirements.txt b/requirements.txt index 811910c..ed2926b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,25 +1,17 @@ -e .[redis,postgres,kafka] # Documentation -mkdocs -mkautodoc -mkdocs-material +mkdocs==1.5.3 +mkdocs-material==9.5.12 +mkautodoc==0.2.0 # Packaging -twine -wheel +build==1.1.1 +twine==5.0.0 # Tests & Linting -autoflake -black==20.8b1 -coverage==5.3 -flake8 -flake8-bugbear -flake8-pie==0.5.* -isort==5.* -mypy -pytest==5.* -pytest-asyncio -pytest-trio -trio -trio-typing +ruff==0.3.5 +coverage==7.4.3 +mypy==1.8.0 +pytest==8.0.2 +pytest-asyncio==0.23.6 \ No newline at end of file diff --git a/scripts/build b/scripts/build index 7d327b5..5d5e71a 100755 --- a/scripts/build +++ b/scripts/build @@ -8,6 +8,6 @@ fi set -x -${PREFIX}python setup.py sdist bdist_wheel +${PREFIX}python -m build ${PREFIX}twine check dist/* # ${PREFIX}mkdocs build diff --git a/scripts/check b/scripts/check index 77acf65..d8fb02b 100755 --- a/scripts/check +++ b/scripts/check @@ -1,14 +1,13 @@ #!/bin/sh -e export PREFIX="" -if [ -d 'venv' ] ; then +if [ -d 'venv' ]; then export PREFIX="venv/bin/" fi export SOURCE_FILES="broadcaster tests" set -x -${PREFIX}black --check --diff --target-version=py37 $SOURCE_FILES -${PREFIX}flake8 $SOURCE_FILES +${PREFIX}ruff format --check --diff $SOURCE_FILES ${PREFIX}mypy $SOURCE_FILES -${PREFIX}isort --check --diff --project=httpx $SOURCE_FILES +${PREFIX}ruff check $SOURCE_FILES diff --git a/scripts/coverage b/scripts/coverage index 73a2198..0015d39 100755 --- a/scripts/coverage +++ b/scripts/coverage @@ -4,8 +4,7 @@ export PREFIX="" if [ -d 'venv' ] ; then export PREFIX="venv/bin/" fi -export SOURCE_FILES="broadcaster tests" set -x -${PREFIX}coverage report --show-missing --skip-covered --fail-under=100 +${PREFIX}coverage report diff --git a/scripts/lint b/scripts/lint index 81851c6..cb718d0 100755 --- a/scripts/lint +++ b/scripts/lint @@ -4,10 +4,9 @@ export PREFIX="" if [ -d 'venv' ] ; then export PREFIX="venv/bin/" fi -export SOURCE_FILES="httpx tests" +export SOURCE_FILES="broadcaster tests" set -x -${PREFIX}autoflake --in-place --recursive $SOURCE_FILES -${PREFIX}isort --project=httpx $SOURCE_FILES -${PREFIX}black --target-version=py37 $SOURCE_FILES +${PREFIX}ruff format $SOURCE_FILES +${PREFIX}ruff check --fix $SOURCE_FILES \ No newline at end of file diff --git a/scripts/publish b/scripts/publish index 238d8cb..400f43f 100755 --- a/scripts/publish +++ b/scripts/publish @@ -1,34 +1,26 @@ #!/bin/sh -e -export VERSION=`cat broadcaster/__init__.py | grep __version__ | sed "s/__version__ = //" | sed "s/'//g"` -export PREFIX="" +VERSION_FILE="broadcaster/__init__.py" + if [ -d 'venv' ] ; then - export PREFIX="venv/bin/" + PREFIX="venv/bin/" +else + PREFIX="" fi -scripts/clean +if [ ! -z "$GITHUB_ACTIONS" ]; then + git config --local user.email "41898282+github-actions[bot]@users.noreply.github.com" + git config --local user.name "GitHub Action" -if ! command -v "${PREFIX}twine" &>/dev/null ; then - echo "Unable to find the 'twine' command." - echo "Install from PyPI, using '${PREFIX}pip install twine'." - exit 1 -fi + VERSION=`grep __version__ ${VERSION_FILE} | grep -o '[0-9][^"]*'` -if ! command -v "${PREFIX}wheel" &>/dev/null ; then - echo "Unable to find the 'wheel' command." - echo "Install from PyPI, using '${PREFIX}pip install wheel'." + if [ "refs/tags/${VERSION}" != "${GITHUB_REF}" ] ; then + echo "GitHub Ref '${GITHUB_REF}' did not match package version '${VERSION}'" exit 1 + fi fi -find broadcaster -type f -name "*.py[co]" -delete -find broadcaster -type d -name __pycache__ -delete +set -x -${PREFIX}python setup.py sdist bdist_wheel ${PREFIX}twine upload dist/* -#${PREFIX}mkdocs gh-deploy - -echo "You probably want to also tag the version now:" -echo "git tag -a ${VERSION} -m 'version ${VERSION}'" -echo "git push --tags" -scripts/clean diff --git a/setup.cfg b/setup.cfg deleted file mode 100644 index c860d81..0000000 --- a/setup.cfg +++ /dev/null @@ -1,24 +0,0 @@ -[flake8] -ignore = W503, E203, B305 -max-line-length = 120 - -[mypy] -disallow_untyped_defs = True -ignore_missing_imports = True - -[mypy-tests.*] -disallow_untyped_defs = False -check_untyped_defs = True - -[tool:isort] -profile = black -combine_as_imports = True - -[tool:pytest] -addopts = -rxXs -markers = - copied_from(source, changes=None): mark test as copied from somewhere else, along with a description of changes made to accodomate e.g. our test setup - -[coverage:run] -omit = venv/* -include = httpx/*, tests/* diff --git a/setup.py b/setup.py deleted file mode 100644 index 668b715..0000000 --- a/setup.py +++ /dev/null @@ -1,68 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -import os -import re - -from setuptools import setup - - -def get_version(package): - """ - Return package version as listed in `__version__` in `init.py`. - """ - with open(os.path.join(package, "__init__.py")) as f: - return re.search("__version__ = ['\"]([^'\"]+)['\"]", f.read()).group(1) - - -def get_long_description(): - """ - Return the README. - """ - with open("README.md", encoding="utf8") as f: - return f.read() - - -def get_packages(package): - """ - Return root package and all sub-packages. - """ - return [ - dirpath - for dirpath, dirnames, filenames in os.walk(package) - if os.path.exists(os.path.join(dirpath, "__init__.py")) - ] - - -setup( - name="broadcaster", - python_requires=">=3.7", - version=get_version("broadcaster"), - url="https://github.com/encode/broadcaster", - license="BSD", - description="Simple broadcast channels.", - long_description=get_long_description(), - long_description_content_type="text/markdown", - author="Tom Christie", - author_email="tom@tomchristie.com", - packages=get_packages("broadcaster"), - data_files=[("", ["LICENSE.md"])], - extras_require={ - "redis": ["asyncio-redis"], - "postgres": ["asyncpg"], - "kafka": ["aiokafka"], - "test": ["pytest", "pytest-asyncio"] - }, - classifiers=[ - "Development Status :: 3 - Alpha", - "Environment :: Web Environment", - "Intended Audience :: Developers", - "License :: OSI Approved :: BSD License", - "Operating System :: OS Independent", - "Topic :: Internet :: WWW/HTTP", - "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.7", - "Programming Language :: Python :: 3.8", - ], - # zip_safe=False, -) diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py index 61e7295..b88b317 100644 --- a/tests/test_broadcast.py +++ b/tests/test_broadcast.py @@ -1,6 +1,41 @@ +from __future__ import annotations + +import asyncio +import typing + import pytest +from redis import asyncio as redis + +from broadcaster import Broadcast, BroadcastBackend, Event +from broadcaster.backends.kafka import KafkaBackend +from broadcaster.backends.redis import RedisBackend + + +class CustomBackend(BroadcastBackend): + def __init__(self, url: str): + self._subscribed: set[str] = set() + + async def connect(self) -> None: + self._published: asyncio.Queue[Event] = asyncio.Queue() + + async def disconnect(self) -> None: + pass -from broadcaster import Broadcast + async def subscribe(self, channel: str) -> None: + self._subscribed.add(channel) + + async def unsubscribe(self, channel: str) -> None: + self._subscribed.remove(channel) + + async def publish(self, channel: str, message: typing.Any) -> None: + event = Event(channel=channel, message=message) + await self._published.put(event) + + async def next_published(self) -> Event: + while True: + event = await self._published.get() + if event.channel in self._subscribed: + return event @pytest.mark.asyncio @@ -23,11 +58,41 @@ async def test_redis(): assert event.message == "hello" +@pytest.mark.asyncio +async def test_redis_configured_client(): + backend = RedisBackend(conn=redis.Redis.from_url("redis://localhost:6379")) + async with Broadcast(backend=backend) as broadcast: + async with broadcast.subscribe("chatroom") as subscriber: + await broadcast.publish("chatroom", "hello") + event = await subscriber.get() + assert event.channel == "chatroom" + assert event.message == "hello" + + +@pytest.mark.asyncio +async def test_redis_requires_url_or_connection(): + with pytest.raises(AssertionError, match="conn must be provided if url is not"): + RedisBackend() + + +@pytest.mark.asyncio +async def test_redis_stream(): + async with Broadcast("redis-stream://localhost:6379") as broadcast: + async with broadcast.subscribe("chatroom") as subscriber: + await broadcast.publish("chatroom", "hello") + event = await subscriber.get() + assert event.channel == "chatroom" + assert event.message == "hello" + async with broadcast.subscribe("chatroom1") as subscriber: + await broadcast.publish("chatroom1", "hello") + event = await subscriber.get() + assert event.channel == "chatroom1" + assert event.message == "hello" + + @pytest.mark.asyncio async def test_postgres(): - async with Broadcast( - "postgres://postgres:postgres@localhost:5432/broadcaster" - ) as broadcast: + async with Broadcast("postgres://postgres:postgres@localhost:5432/broadcaster") as broadcast: async with broadcast.subscribe("chatroom") as subscriber: await broadcast.publish("chatroom", "hello") event = await subscriber.get() @@ -43,3 +108,37 @@ async def test_kafka(): event = await subscriber.get() assert event.channel == "chatroom" assert event.message == "hello" + + +@pytest.mark.asyncio +async def test_kafka_multiple_urls(): + async with Broadcast(backend=KafkaBackend(urls=["kafka://localhost:9092", "kafka://localhost:9092"])) as broadcast: + async with broadcast.subscribe("chatroom") as subscriber: + await broadcast.publish("chatroom", "hello") + event = await subscriber.get() + assert event.channel == "chatroom" + assert event.message == "hello" + + +@pytest.mark.asyncio +async def test_custom(): + backend = CustomBackend("") + async with Broadcast(backend=backend) as broadcast: + async with broadcast.subscribe("chatroom") as subscriber: + await broadcast.publish("chatroom", "hello") + event = await subscriber.get() + assert event.channel == "chatroom" + assert event.message == "hello" + + +@pytest.mark.asyncio +async def test_unknown_backend(): + with pytest.raises(ValueError, match="Unsupported backend"): + async with Broadcast(url="unknown://"): + pass + + +@pytest.mark.asyncio +async def test_needs_url_or_backend(): + with pytest.raises(AssertionError, match="Either `url` or `backend` must be provided."): + Broadcast() diff --git a/tests/test_unsubscribe.py b/tests/test_unsubscribe.py new file mode 100644 index 0000000..ae89401 --- /dev/null +++ b/tests/test_unsubscribe.py @@ -0,0 +1,26 @@ +import pytest + +from broadcaster import Broadcast + + +@pytest.mark.asyncio +async def test_unsubscribe(): + """The queue should be removed when the context manager is left.""" + async with Broadcast("memory://") as broadcast: + async with broadcast.subscribe("chatroom"): + pass + + assert "chatroom" not in broadcast._subscribers + + +@pytest.mark.asyncio +async def test_unsubscribe_w_exception(): + """In case an exception is raised inside the context manager, the queue should be removed.""" + async with Broadcast("memory://") as broadcast: + try: + async with broadcast.subscribe("chatroom"): + raise RuntimeError("MyException") + except RuntimeError: + pass + + assert "chatroom" not in broadcast._subscribers