From 2efa8c93895fc3c3fc7b7ead53ed7797de862d46 Mon Sep 17 00:00:00 2001 From: Arseniy Popov Date: Sat, 20 Jun 2026 22:52:50 +0300 Subject: [PATCH 1/2] refactor: rename work queue to competing consumers --- docs/docs/sqlbroker/tutorial.md | 14 +++++++++++--- docs/docs/stylesheets/extra.css | 8 ++++++++ faststream_sqlbroker/__init__.py | 4 ++-- faststream_sqlbroker/sqlbroker/__init__.py | 4 ++-- faststream_sqlbroker/sqlbroker/schema.py | 18 ++++++++++-------- pyproject.toml | 2 +- tests/test_misconfigure.py | 2 +- tests/test_public_api.py | 6 +++--- uv.lock | 2 +- 9 files changed, 39 insertions(+), 21 deletions(-) diff --git a/docs/docs/sqlbroker/tutorial.md b/docs/docs/sqlbroker/tutorial.md index 16daec6..148ce10 100644 --- a/docs/docs/sqlbroker/tutorial.md +++ b/docs/docs/sqlbroker/tutorial.md @@ -27,9 +27,17 @@ PostgreSQL, MySQL, and SQLite are currently supported. pip install "faststream-sqlbroker" ``` +## Schema Variants + +A schema variant defines how messages are laid out in the database and how subscribers compete for work. You select it with `schema.variant` on `SqlBrokerSchemaConfig` (see [Broker](#broker){.internal-link}). + +### `COMPETING_CONSUMERS` (default) + +The [competing consumers](https://www.enterpriseintegrationpatterns.com/patterns/messaging/CompetingConsumers.html){.external-link target="_blank"} pattern: multiple processes share one queue, each message is handled by one concurrent worker, and processing order is not guaranteed. + ## Database Tables -Currently the only supported DB schema variant is `WORK_QUEUE`, at version `1`, which uses up to two tables — `message` (active messages) and `message_archive` (completed/failed messages). These settings are grouped under the broker's `schema` parameter via `SqlBrokerSchemaConfig`. Set `message_archive_table_name=None` there to omit using archiving on success and [DLQ](../sqlbroker/design.md#dead-letter-queue){.internal-link}. You can customize the tables to your liking (partition them, add indices, specify more specific data types like `JSONB`, etc.) as long as they generally conform to the selected schema definition. Schema check is done on startup if the broker's `validate_schema_on_start` is `True`. +The `COMPETING_CONSUMERS` variant (version `1`) uses up to two tables — `message` (active messages) and `message_archive` (completed/failed messages). These settings are grouped under the broker's `schema` parameter via `SqlBrokerSchemaConfig`. Set `message_archive_table_name=None` there to omit using archiving on success and [DLQ](../sqlbroker/design.md#dead-letter-queue){.internal-link}. You can customize the tables to your liking (partition them, add indices, specify more specific data types like `JSONB`, etc.) as long as they generally conform to the selected schema definition. Schema check is done on startup if the broker's `validate_schema_on_start` is `True`. ```python linenums="1" {!> docs_src/sqlbroker/tables.py !} @@ -47,8 +55,8 @@ Currently the only supported DB schema variant is `WORK_QUEUE`, at version `1`, - **`schema`** (default: `SqlBrokerSchemaConfig()`) — `SqlBrokerSchemaConfig` describing the broker tables and schema selection. - **`schema.message_table_name`** (default: `message`) — Name of the table containing active messages. - **`schema.message_archive_table_name`** (default: `message_archive`) — Name of the table containing completed/failed messages. Set to `None` to run without an archive table, in which case subscribers must set both `retain_in_archive_on_ack` and `retain_in_archive_on_reject` to `False`. -- **`schema.variant`** (default: `WORK_QUEUE`) — Schema variant to use. -- **`schema.version`** (default: `V1`) — Variant-specific schema version enum. For `WORK_QUEUE`, use `SqlBrokerWorkQueueSchemaVersion`. +- **`schema.variant`** (default: `COMPETING_CONSUMERS`) — Schema variant to use. +- **`schema.version`** (default: `V1`) — Variant-specific schema version enum. For `COMPETING_CONSUMERS`, use `SqlBrokerCompetingConsumersSchemaVersion`. - **`validate_schema_on_start`** (default: `True`) — Validates that the configured tables exist and conform to the expected schema. - **`graceful_timeout`** (default: `15.0`) — Seconds to wait for in-flight messages to finish processing during shutdown. diff --git a/docs/docs/stylesheets/extra.css b/docs/docs/stylesheets/extra.css index 0e08020..ad1559e 100644 --- a/docs/docs/stylesheets/extra.css +++ b/docs/docs/stylesheets/extra.css @@ -17,3 +17,11 @@ a.external-link::after { to keep the mark on the same line as the link */ content: "\00A0[↪]"; } + +article.md-content__inner.md-typeset > h2 { + background: var(--md-accent-fg-color--transparent); + border-left: 4px solid var(--md-accent-fg-color); + border-radius: 0 4px 4px 0; + margin-top: 2em; + padding: 0.35rem 0.75rem; +} diff --git a/faststream_sqlbroker/__init__.py b/faststream_sqlbroker/__init__.py index b2e6e07..2d5fee5 100644 --- a/faststream_sqlbroker/__init__.py +++ b/faststream_sqlbroker/__init__.py @@ -1,5 +1,6 @@ from .sqlbroker import ( SqlBroker, + SqlBrokerCompetingConsumersSchemaVersion, SqlBrokerMessage, SqlBrokerPublishCommand, SqlBrokerPublisher, @@ -8,7 +9,6 @@ SqlBrokerSchemaConfig, SqlBrokerSchemaType, SqlBrokerSchemaVariant, - SqlBrokerWorkQueueSchemaVersion, TestApp, ) from .sqlbroker.retry import ( @@ -32,6 +32,7 @@ "RetryStrategyProto", "RetryStrategyTemplate", "SqlBroker", + "SqlBrokerCompetingConsumersSchemaVersion", "SqlBrokerMessage", "SqlBrokerPublishCommand", "SqlBrokerPublisher", @@ -40,6 +41,5 @@ "SqlBrokerSchemaConfig", "SqlBrokerSchemaType", "SqlBrokerSchemaVariant", - "SqlBrokerWorkQueueSchemaVersion", "TestApp", ) diff --git a/faststream_sqlbroker/sqlbroker/__init__.py b/faststream_sqlbroker/sqlbroker/__init__.py index 9bd82fb..4ead2e5 100644 --- a/faststream_sqlbroker/sqlbroker/__init__.py +++ b/faststream_sqlbroker/sqlbroker/__init__.py @@ -5,10 +5,10 @@ from .broker import SqlBroker, SqlBrokerPublisher, SqlBrokerRoute, SqlBrokerRouter from .response import SqlBrokerPublishCommand from .schema import ( + SqlBrokerCompetingConsumersSchemaVersion, SqlBrokerSchemaConfig, SqlBrokerSchemaType, SqlBrokerSchemaVariant, - SqlBrokerWorkQueueSchemaVersion, ) except ImportError as e: @@ -21,6 +21,7 @@ __all__ = ( "SqlBroker", + "SqlBrokerCompetingConsumersSchemaVersion", "SqlBrokerMessage", "SqlBrokerPublishCommand", "SqlBrokerPublisher", @@ -29,6 +30,5 @@ "SqlBrokerSchemaConfig", "SqlBrokerSchemaType", "SqlBrokerSchemaVariant", - "SqlBrokerWorkQueueSchemaVersion", "TestApp", ) diff --git a/faststream_sqlbroker/sqlbroker/schema.py b/faststream_sqlbroker/sqlbroker/schema.py index f8e3443..60c39aa 100644 --- a/faststream_sqlbroker/sqlbroker/schema.py +++ b/faststream_sqlbroker/sqlbroker/schema.py @@ -21,10 +21,10 @@ class SqlBrokerSchemaVariant(str, Enum): - WORK_QUEUE = "work_queue" + COMPETING_CONSUMERS = "competing_consumers" -class SqlBrokerWorkQueueSchemaVersion(IntEnum): +class SqlBrokerCompetingConsumersSchemaVersion(IntEnum): V1 = 1 @@ -37,8 +37,10 @@ class SqlBrokerSchemaType(str, Enum): class SqlBrokerSchemaConfig: message_table_name: str = "message" message_archive_table_name: str | None = "message_archive" - variant: SqlBrokerSchemaVariant = SqlBrokerSchemaVariant.WORK_QUEUE - version: SqlBrokerWorkQueueSchemaVersion = SqlBrokerWorkQueueSchemaVersion.V1 + variant: SqlBrokerSchemaVariant = SqlBrokerSchemaVariant.COMPETING_CONSUMERS + version: SqlBrokerCompetingConsumersSchemaVersion = ( + SqlBrokerCompetingConsumersSchemaVersion.V1 + ) def _unsupported_schema_variant_error( @@ -55,14 +57,14 @@ def _unsupported_schema_version_error( return SetupError( "Unsupported SqlBroker schema version for " f"{variant.value}: {version}. Supported versions: " - f"{SqlBrokerWorkQueueSchemaVersion.V1.value}" + f"{SqlBrokerCompetingConsumersSchemaVersion.V1.value}" ) @dataclass(frozen=True) class SqlBrokerSchemaDefinition: variant: SqlBrokerSchemaVariant - version: SqlBrokerWorkQueueSchemaVersion + version: SqlBrokerCompetingConsumersSchemaVersion tables: dict[SqlBrokerSchemaType, Table] def get_table(self, schema_type: SqlBrokerSchemaType) -> Table | None: @@ -75,11 +77,11 @@ def define_sqlbroker_schema( ) -> SqlBrokerSchemaDefinition: variant = config.variant - if variant is not SqlBrokerSchemaVariant.WORK_QUEUE: + if variant is not SqlBrokerSchemaVariant.COMPETING_CONSUMERS: raise _unsupported_schema_variant_error(variant) version = config.version - if not isinstance(version, SqlBrokerWorkQueueSchemaVersion): + if not isinstance(version, SqlBrokerCompetingConsumersSchemaVersion): raise _unsupported_schema_version_error( variant=variant, version=version, diff --git a/pyproject.toml b/pyproject.toml index 21f0093..a76aade 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ authors = [ { name = "Arseniy Popov", email = "arseniypopov@gmail.com" }, ] requires-python = ">=3.10" -version = "0.1.0a5" +version = "0.1.0a6" dependencies = [ "faststream>=0.7.0rc1", diff --git a/tests/test_misconfigure.py b/tests/test_misconfigure.py index 4ca8acf..4c7cf1e 100644 --- a/tests/test_misconfigure.py +++ b/tests/test_misconfigure.py @@ -189,7 +189,7 @@ async def test_fail_on_unsupported_schema_version() -> None: broker = SqlBroker( engine=create_async_engine("sqlite+aiosqlite:///:memory:"), schema=SqlBrokerSchemaConfig( - variant=SqlBrokerSchemaVariant.WORK_QUEUE, + variant=SqlBrokerSchemaVariant.COMPETING_CONSUMERS, version=cast("Any", 2), ), ) diff --git a/tests/test_public_api.py b/tests/test_public_api.py index 8d061c8..1795c9f 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -1,16 +1,16 @@ from faststream_sqlbroker import ( ExponentialBackoffRetryStrategy, SqlBroker, + SqlBrokerCompetingConsumersSchemaVersion, SqlBrokerSchemaConfig, SqlBrokerSchemaType, SqlBrokerSchemaVariant, - SqlBrokerWorkQueueSchemaVersion, ) def test_top_level_exports() -> None: assert SqlBroker.__name__ == "SqlBroker" assert ExponentialBackoffRetryStrategy.__name__ == "ExponentialBackoffRetryStrategy" - assert SqlBrokerSchemaVariant.WORK_QUEUE.value == "work_queue" + assert SqlBrokerSchemaVariant.COMPETING_CONSUMERS.value == "competing_consumers" assert SqlBrokerSchemaType.MESSAGE.value == "message" - assert SqlBrokerSchemaConfig().version is SqlBrokerWorkQueueSchemaVersion.V1 + assert SqlBrokerSchemaConfig().version is SqlBrokerCompetingConsumersSchemaVersion.V1 diff --git a/uv.lock b/uv.lock index 09b9daf..91c53ef 100644 --- a/uv.lock +++ b/uv.lock @@ -708,7 +708,7 @@ wheels = [ [[package]] name = "faststream-sqlbroker" -version = "0.1.0a5" +version = "0.1.0a6" source = { editable = "." } dependencies = [ { name = "faststream" }, From 38f21b562c90e17b8cebf24004b99a9aa68d5f1d Mon Sep 17 00:00:00 2001 From: Arseniy Popov Date: Mon, 22 Jun 2026 20:44:50 +0300 Subject: [PATCH 2/2] refactor: rename work queue to competing consumers --- docs/docs/sqlbroker/tutorial.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/docs/sqlbroker/tutorial.md b/docs/docs/sqlbroker/tutorial.md index 148ce10..6dabcb0 100644 --- a/docs/docs/sqlbroker/tutorial.md +++ b/docs/docs/sqlbroker/tutorial.md @@ -29,9 +29,9 @@ pip install "faststream-sqlbroker" ## Schema Variants -A schema variant defines how messages are laid out in the database and how subscribers compete for work. You select it with `schema.variant` on `SqlBrokerSchemaConfig` (see [Broker](#broker){.internal-link}). +A schema variant selects the messaging topology the broker implements. -### `COMPETING_CONSUMERS` (default) +### `COMPETING_CONSUMERS` The [competing consumers](https://www.enterpriseintegrationpatterns.com/patterns/messaging/CompetingConsumers.html){.external-link target="_blank"} pattern: multiple processes share one queue, each message is handled by one concurrent worker, and processing order is not guaranteed.