refactor: rename work queue to competing consumers

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 selects the messaging topology the broker implements.
+
+### `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.
+
 ## 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.
 

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;
+}

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",
 )

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",
 )

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,

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",

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),
         ),
     )

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