Add a target

Create a direct package under backend/app/sync_v2/targets. The example below uses an example_target identifier and a simple Saved feature. Replace the fixture reads and outcomes with provider API calls; the public contract remains the same.

backend/app/sync_v2/targets/example_target/
├── __init__.py
└── plugin.py

targets/example_target/plugin.py

from dataclasses import dataclass
from decimal import Decimal
from typing import Any, Mapping, Sequence

from app.sync_v2.sdk import (
    ConfigSchema,
    DefaultTargetFeature,
    FeatureDefinition,
    FeaturePlanContext,
    HttpPolicy,
    OperationAction,
    SourceItem,
    TargetContext,
    TargetItemReference,
    TargetManifest,
    TargetMapping,
    TargetOperation,
    TargetRuntime,
    TargetSnapshot,
    WriteOutcome,
    WriteStatus,
)


class SavedFeature(DefaultTargetFeature):
    definition = FeatureDefinition(
        key="saved",
        label="Saved items",
        description="Keep source watchlist membership in Example Target.",
        required_source_capabilities=frozenset({"watchlist.observe"}),
        required_target_capabilities=frozenset({"saved.read", "saved.write"}),
        group_key="membership",
        group_label="Membership",
    )

    def plan(self, context: FeaturePlanContext) -> tuple[TargetOperation, ...]:
        desired = context.source.fields.get("watchlist") is True
        current = bool(context.snapshot and context.snapshot.fields.get("saved"))
        if desired == current:
            return ()
        return (self.operation(
            context,
            action=OperationAction.UPDATE,
            reason="Saved membership differs from the source observation.",
            before={"saved": current},
            after={"saved": desired},
            source_diagnostics=context.source.diagnostics,
            apply_payload={"saved": desired},
            revert_payload={"saved": current},
        ),)


class ExampleTargetRuntime(TargetRuntime):
    def __init__(self, context: TargetContext) -> None:
        self.context = context

    async def map_items(
        self, items: Sequence[SourceItem]
    ) -> Mapping[str, tuple[TargetMapping, ...]]:
        result: dict[str, tuple[TargetMapping, ...]] = {}
        for item in items:
            target_id = item.identifiers.get("example_target")
            if not target_id:
                continue
            reference = TargetItemReference(
                item_id=target_id,
                title=item.title,
                media_type=item.media_type,
                identifiers={"example_target": target_id},
                site_url=f"https://example.com/items/{target_id}",
            )
            result[item.item_key] = (TargetMapping(
                source_item_key=item.item_key,
                segment_key=item.item_key,
                source=item,
                target=reference,
                mapping_source="direct:example_target",
            ),)
        return result

    async def read_items(
        self,
        items: Sequence[TargetItemReference],
        feature_keys: frozenset[str],
    ) -> Mapping[str, TargetSnapshot]:
        return {
            item.item_id: TargetSnapshot(item=item, fields={"saved": False})
            for item in items
        }

    async def write_items(
        self, operations: Sequence[TargetOperation]
    ) -> Sequence[WriteOutcome]:
        return tuple(
            WriteOutcome(
                operation_id=operation.operation_id,
                status=WriteStatus.APPLIED,
                authoritative_after=operation.after,
            )
            for operation in operations
        )

    async def revert_items(
        self, operations: Sequence[TargetOperation]
    ) -> Sequence[WriteOutcome]:
        return tuple(
            WriteOutcome(
                operation_id=operation.operation_id,
                status=WriteStatus.APPLIED,
                authoritative_after=operation.before,
            )
            for operation in operations
        )


@dataclass(frozen=True, slots=True)
class ExampleTargetPlugin:
    manifest = TargetManifest(
        target_type="example_target",
        label="Example Target",
        plugin_version="1.0.0",
        supported_media_types=("movie",),
        auth_modes=("none",),
        capabilities=frozenset({"saved.read", "saved.write"}),
        resources={"website": "https://example.com/"},
    )
    instance_schema = ConfigSchema(title="Example Target account")
    http_policy = HttpPolicy(
        request_interval_seconds=Decimal("0.5"),
        retry_attempts=3,
        retry_delay_seconds=Decimal("1"),
    )
    features = (SavedFeature(),)
    auth = None
    actions = ()

    def create_runtime(self, context: TargetContext) -> TargetRuntime:
        return ExampleTargetRuntime(context)

    async def run_action(
        self, context: TargetContext, key: str, params: Mapping[str, Any]
    ) -> Mapping[str, Any]:
        raise ValueError(f"Unknown Example Target action: {key}")


def create_plugin() -> ExampleTargetPlugin:
    return ExampleTargetPlugin()

Mapping rules

Return zero mappings for an unsupported/unmatched item, one for ordinary media, or multiple stable segments when one source item maps to several target entries. Each mapping projection must use its segment_key as source.item_key.

Prefer, in order, profile manual overrides, direct provider identifiers, shared mapping-catalog candidates, and only then a provider search that is safe and unambiguous. Put the real provenance in mapping_source.

Override mapping_identity() when mapping depends on more canonical data than the default title, media type, identifiers, year, and mapping descriptors. Override side-effect-free refresh_mapping() when cached segmented projections must be rebuilt from fresh source observations.

Reads, writes, and Revert

• Batch reads by provider limits and return the complete state required by enabled features.
• Keep remote request construction and provider error interpretation inside the runtime.
• Return exactly one outcome for every submitted operation.
• Use retryable only when replay is safe; use ambiguous when acceptance cannot be proven.
• Reconcile non-idempotent timeouts before another write.
• Persist enough opaque revert_payload to restore exactly the changed fields.
• Do not change unrelated notes, tags, privacy, custom lists, or provider metadata.

OperationPresentation can add labels, safe badges, grouping, detail sections, artwork, and target URLs to Runs without provider branches in frontend code.

Optional target contracts

Targets may add hosted/custom auth, refresh, actions, rating scales, custom-list catalog/write/revert, provider-specific features, post-planning operation coordination, and reconciliation. Declare the matching capabilities and keep feature planning deterministic.

Verification

Test direct/manual/catalog mapping, batching, media filtering, dry-run/live, conflicts, partial and ambiguous outcomes, cancellation, pacing/retry, token refresh, Apply, Revert, and operation presentation. Discovery must expose the target in Targets without changes to core routes or frontend registries.