feat(indexers): send an indexer's grabs to a chosen download client

alembic/versions/i6c7d8e9f012_add_indexer_download_client.py

@@ -0,0 +1,35 @@
+"""Let an indexer send every grab to one chosen download client.
+
+Revision ID: i6c7d8e9f012
+Revises: h5b6c7d8e901
+Create Date: 2026-10-03
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "i6c7d8e9f012"
+down_revision = "h5b6c7d8e901"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Add the nullable client pin; existing indexers stay unpinned."""
+    with op.batch_alter_table("indexer_configs") as batch_op:
+        batch_op.add_column(sa.Column("download_client_id", sa.Integer(), nullable=True))
+        batch_op.create_foreign_key(
+            "fk_indexer_configs_download_client",
+            "download_client_configs",
+            ["download_client_id"],
+            ["id"],
+            ondelete="SET NULL",
+        )
+
+
+def downgrade() -> None:
+    """Drop the pin; those indexers fall back to priority-based selection."""
+    with op.batch_alter_table("indexer_configs") as batch_op:
+        batch_op.drop_constraint("fk_indexer_configs_download_client", type_="foreignkey")
+        batch_op.drop_column("download_client_id")

src/pullbox/api/v1/clients.py

@@ -243,16 +243,6 @@ async def add_client(
     session: DbSession,
 ) -> ClientResponse:
     """Add a new download client configuration."""
-    if body.client_type is not DownloadClientType.AIRDCPP:
-        existing = await session.execute(
-            select(DownloadClientConfig).where(DownloadClientConfig.client_type == body.client_type)
-        )
-        if existing.scalar_one_or_none():
-            raise ValidationError(
-                f"A {body.client_type.value} client is already configured. "
-                "Only one instance per client type is allowed."
-            )
-
     if body.client_type is DownloadClientType.AIRDCPP:
         if not get_settings().airdcpp_enabled:
             raise ValidationError("AirDC++ integration is disabled by the feature flag.")

src/pullbox/api/v1/indexers.py

@@ -9,14 +9,17 @@
 from sqlalchemy.exc import OperationalError
 
 from pullbox.api.deps import DbSession, InteractiveOperatorUser
+from pullbox.core.acquisition import AcquisitionProtocol
 from pullbox.core.encryption import decrypt_secret, encrypt_secret
 from pullbox.core.exceptions import NotFoundError, ValidationError
 from pullbox.core.sqlite_lock import (
     SQLITE_LOCK_RETRY_ATTEMPTS,
     is_sqlite_locked_error,
     sqlite_lock_retry_delay,
 )
-from pullbox.models.indexer import IndexerConfig
+from pullbox.models.client import DownloadClientConfig
+from pullbox.models.download import DownloadClientType
+from pullbox.models.indexer import IndexerConfig, IndexerType
 from pullbox.schemas.indexer import (
     IndexerCreate,
     IndexerResponse,
@@ -54,6 +57,7 @@ def _redact_indexer(indexer: IndexerConfig) -> dict[str, object]:
         "enable_automatic_search": indexer.enable_automatic_search,
         "enable_interactive_search": indexer.enable_interactive_search,
         "resolver_enabled": indexer.resolver_enabled,
+        "download_client_id": indexer.download_client_id,
         "last_success_at": indexer.last_success_at,
         "last_failure_at": indexer.last_failure_at,
         "last_error": indexer.last_error,
@@ -136,6 +140,7 @@ async def add_indexer(
         source="manual",
         resolver_enabled=body.resolver_enabled,
     )
+    await _validate_download_client(session, body.indexer_type, body.download_client_id)
     indexer = IndexerConfig(
         name=body.name,
         indexer_type=body.indexer_type,
@@ -149,6 +154,7 @@ async def add_indexer(
         enable_automatic_search=body.enable_automatic_search,
         enable_interactive_search=body.enable_interactive_search,
         resolver_enabled=body.resolver_enabled,
+        download_client_id=body.download_client_id,
     )
     session.add(indexer)
     await session.flush()
@@ -501,6 +507,11 @@ async def update_indexer(
             resolver_enabled=True,
         )
 
+    if update_data.get("download_client_id") is not None:
+        await _validate_download_client(
+            session, indexer.indexer_type, update_data["download_client_id"]
+        )
+
     # Encrypt api_key if provided; omitted → keep existing encrypted value
     if update_data.get("api_key"):
         update_data["api_key"] = encrypt_secret(update_data["api_key"])
@@ -530,6 +541,35 @@ def _validate_resolver_scope(
         )
 
 
+_PROTOCOL_BY_INDEXER_TYPE: dict[str, tuple[AcquisitionProtocol, ...]] = {
+    IndexerType.TORZNAB: (AcquisitionProtocol.TORRENT,),
+    IndexerType.NEWZNAB: (AcquisitionProtocol.USENET,),
+}
+
+
+async def _validate_download_client(
+    session: DbSession,
+    indexer_type: object,
+    download_client_id: int | None,
+) -> None:
+    """Check a pinned client exists and takes the protocol this indexer returns."""
+    if download_client_id is None:
+        return
+    client = await session.get(DownloadClientConfig, download_client_id)
+    if client is None:
+        raise ValidationError(f"Download client {download_client_id} does not exist.")
+    wanted = _PROTOCOL_BY_INDEXER_TYPE.get(
+        str(indexer_type),
+        (AcquisitionProtocol.TORRENT, AcquisitionProtocol.USENET),
+    )
+    if DownloadClientType(client.client_type).acquisition_protocol not in wanted:
+        kind = " or ".join(protocol.value for protocol in wanted)
+        raise ValidationError(
+            f'"{client.name}" is not a {kind} download client, so it cannot '
+            "take releases from this indexer."
+        )
+
+
 # ── Delete ───────────────────────────────────────────────────────────
 
 

src/pullbox/models/indexer.py

@@ -3,7 +3,7 @@
 import enum
 from datetime import datetime
 
-from sqlalchemy import Boolean, Integer, String, Text
+from sqlalchemy import Boolean, ForeignKey, Integer, String, Text
 from sqlalchemy import Enum as SQLAlchemyEnum
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -35,6 +35,13 @@ class IndexerConfig(Base, IdentityMixin, TimestampMixin):
     priority: Mapped[int] = mapped_column(Integer, default=50)
     categories: Mapped[str | None] = mapped_column(Text)
 
+    # Download client every grab from this indexer is sent to. Null means the
+    # highest-priority client for the release's protocol.
+    download_client_id: Mapped[int | None] = mapped_column(
+        ForeignKey("download_client_configs.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+
     # Manager sync tracking. The legacy Prowlarr integer remains during the
     # additive migration so existing databases and integrations stay readable.
     source: Mapped[str] = mapped_column(String(20), default="manual")

src/pullbox/schemas/indexer.py

@@ -25,6 +25,10 @@ class IndexerCreate(BaseModel):
         False,
         description="Allow a manual Torznab indexer to use the ranked browser resolver chain",
     )
+    download_client_id: int | None = Field(
+        None,
+        description="Send every grab from this indexer to this download client",
+    )
 
     @field_validator("url")
     @classmethod
@@ -51,6 +55,10 @@ class IndexerUpdate(BaseModel):
         None,
         description="Allow a manual Torznab indexer to use the ranked browser resolver chain",
     )
+    download_client_id: int | None = Field(
+        None,
+        description="Send every grab to this download client; null for the default client",
+    )
 
     @field_validator("url")
     @classmethod
@@ -82,6 +90,7 @@ class IndexerResponse(BaseModel):
     enable_automatic_search: bool = True
     enable_interactive_search: bool = True
     resolver_enabled: bool = False
+    download_client_id: int | None = None
     last_success_at: datetime | None = None
     last_failure_at: datetime | None = None
     last_error: str | None = None

src/pullbox/services/download_service.py

@@ -66,7 +66,7 @@ async def send_to_client(
             indexer_id = release.indexer_id
 
         # Select the appropriate client
-        client = self._select_client(release.protocol)
+        client = await self._client_for_release(session, release.protocol, indexer_id)
         if not client:
             raise ProviderError(
                 "download",
@@ -503,6 +503,54 @@ async def on_attempt(event: object) -> None:
             clear_download_progress(download_id)
             raise
 
+    async def _client_for_release(
+        self,
+        session: AsyncSession,
+        protocol: AcquisitionProtocol,
+        indexer_id: int | None,
+    ) -> DownloadClient | None:
+        """Use the indexer's pinned client when it has one, else the protocol default.
+
+        A pinned client is never swapped for another one: if it is disabled or
+        cannot take this protocol the grab fails, so releases from that indexer
+        never land in a client the user kept them out of.
+        """
+        if indexer_id is not None:
+            from pullbox.models.indexer import IndexerConfig
+
+            indexer = await session.get(IndexerConfig, indexer_id)
+            if indexer is not None and indexer.download_client_id is not None:
+                return await self._pinned_client(
+                    session, indexer.download_client_id, protocol, indexer.name
+                )
+        return self._select_client(protocol)
+
+    async def _pinned_client(
+        self,
+        session: AsyncSession,
+        client_config_id: int,
+        protocol: AcquisitionProtocol,
+        indexer_name: str,
+    ) -> DownloadClient:
+        client = self._registry.get_download_client(client_config_id)
+        if client is None:
+            from pullbox.models.client import DownloadClientConfig
+
+            config = await session.get(DownloadClientConfig, client_config_id)
+            label = config.name if config is not None else f"#{client_config_id}"
+            raise ProviderError(
+                "download",
+                f'Indexer "{indexer_name}" sends its grabs to download client "{label}", '
+                "which is disabled or unavailable.",
+            )
+        if DownloadClientType(client.client_type).acquisition_protocol is not protocol:
+            raise ProviderError(
+                "download",
+                f'Indexer "{indexer_name}" is set to download client "{client.name}", '
+                f"which cannot take {protocol.value} releases.",
+            )
+        return client
+
     def _select_client(self, protocol: AcquisitionProtocol) -> DownloadClient | None:
         """Select the highest-priority client for an acquisition protocol."""
         if protocol is AcquisitionProtocol.TORRENT:
@@ -541,9 +589,19 @@ async def _persisted_config_id_for_client(
 
     def get_client_for_download(self, download: DownloadHistory) -> DownloadClient | None:
         """Resolve the exact persisted client, with fallback for legacy null rows."""
-        if download.download_client_config_id is not None:
-            return self._registry.get_download_client(download.download_client_config_id)
-        return self._registry.get_client_for_type(str(download.download_client))
+        return self.get_client_for_identity(
+            download.download_client_config_id, download.download_client
+        )
+
+    def get_client_for_identity(
+        self,
+        client_config_id: int | None,
+        client_type: object,
+    ) -> DownloadClient | None:
+        """Resolve a client by its config ID, or by type for rows recorded without one."""
+        if client_config_id is not None:
+            return self._registry.get_download_client(client_config_id)
+        return self._registry.get_client_for_type(str(client_type))
 
     def get_client_for_type(self, client_type: object) -> DownloadClient | None:
         """Get the client for a given DownloadClientType value."""

src/pullbox/tasks/download_monitor_poll.py

@@ -30,7 +30,9 @@ async def poll_download_clients(
         client_type = item["download_client"]
         existing_path = item["downloaded_path"]
 
-        client = download_svc.get_client_for_type(client_type)
+        client = download_svc.get_client_for_identity(
+            item.get("download_client_config_id"), client_type
+        )
         if not client:
             continue
 

src/pullbox/tasks/download_monitor_read.py

@@ -37,6 +37,7 @@ def build_poll_item(download: Any) -> dict[str, object]:
         "external_id": download.external_id,
         "title": download.title,
         "download_client": download.download_client,
+        "download_client_config_id": download.download_client_config_id,
         "downloaded_path": download.downloaded_path,
         "issue_id": download.issue_id,
         "retry_count": download.retry_count,

src/pullbox/tasks/download_post_processing_sources.py

@@ -15,6 +15,7 @@
 
     from sqlalchemy.ext.asyncio import AsyncSession
 
+    from pullbox.models.client import DownloadClientConfig
     from pullbox.models.download import DownloadHistory
 
 logger = structlog.get_logger(__name__)
@@ -251,13 +252,7 @@ async def _resolve_local_path(
         )
         return str(mapped)
 
-    result = await session.execute(
-        select(DownloadClientConfig).where(
-            DownloadClientConfig.client_type == download.download_client,
-            DownloadClientConfig.enabled.is_(True),
-        )
-    )
-    client_cfg = result.scalars().first()
+    client_cfg = await _download_client_config(session, download)
 
     if client_cfg and client_cfg.remote_path and client_cfg.download_dir:
         windows_origin = _is_windows_origin_path(client_cfg.remote_path)
@@ -299,20 +294,38 @@ async def _resolve_local_path(
     return raw_path
 
 
-async def _resolve_local_download_root(
+async def _download_client_config(
     session: AsyncSession,
     download: DownloadHistory,
-) -> Path | None:
-    """Return the configured local root that bounds source cleanup."""
+) -> DownloadClientConfig | None:
+    """Return the config of the client that handled this download.
+
+    Several clients can share a type, so the exact config recorded on the
+    download wins; rows recorded before that identity existed fall back to the
+    first enabled client of the same type. The exact config is used even when
+    it has since been disabled: its paths still describe where this download is.
+    """
     from pullbox.models.client import DownloadClientConfig
 
+    if download.download_client_config_id is not None:
+        exact = await session.get(DownloadClientConfig, download.download_client_config_id)
+        if exact is not None:
+            return exact
     result = await session.execute(
         select(DownloadClientConfig).where(
             DownloadClientConfig.client_type == download.download_client,
             DownloadClientConfig.enabled.is_(True),
         )
     )
-    client_cfg = result.scalars().first()
+    return result.scalars().first()
+
+
+async def _resolve_local_download_root(
+    session: AsyncSession,
+    download: DownloadHistory,
+) -> Path | None:
+    """Return the configured local root that bounds source cleanup."""
+    client_cfg = await _download_client_config(session, download)
     if client_cfg is None or not client_cfg.download_dir:
         return None
     download_dir = client_cfg.download_dir.strip()