refactor(core)!: Destination stores one partition at a time; PartitionedDestination folds into it

README.md

@@ -108,8 +108,9 @@ shop = Shop(
 )
 ```
 
-Your own destination is two methods, sync or async. `PartitionedDestination` and
-`DatabaseDestination` handle partition scoping for you.
+Your own destination is two methods, sync or async, each handling one partition, or the
+unpartitioned whole. Windows are split and gathered for you; `DatabaseDestination` adds the
+delete-then-insert dance for tables.
 
 ```python
 import json
@@ -120,14 +121,17 @@ from pathlib import Path
 class JSONLDestination(il.Destination):
     base_path: str = ""
 
-    def write(self, context: il.IOContext, data) -> None:
-        path = Path(self.base_path) / context.asset.dataset / f"{context.asset.table}.jsonl"
+    def _path(self, context: il.IOContext, partition: il.Partition | None) -> Path:
+        base = Path(self.base_path) / context.asset.dataset / context.asset.table
+        return base / ("data.jsonl" if partition is None else f"{partition.id}.jsonl")
+
+    def write_partition(self, context: il.IOContext, partition: il.Partition | None, data) -> None:
+        path = self._path(context, partition)
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text("\n".join(json.dumps(row, default=str) for row in data))
 
-    def read(self, context: il.IOContext):
-        path = Path(self.base_path) / context.asset.dataset / f"{context.asset.table}.jsonl"
-        return [json.loads(line) for line in path.read_text().splitlines()]
+    def read_partition(self, context: il.IOContext, partition: il.Partition | None):
+        return [json.loads(line) for line in self._path(context, partition).read_text().splitlines()]
 ```
 
 ### Running

docs/extending/representations.md

@@ -65,7 +65,7 @@ rows, whose record coercion rejects non-tabular data with a clear error.
 - Conform resolves the conformer through `Representation.of(result)`.
 - `Partition.slice()` and `TimePartition.slice()` filter through the representation, so window
   writes split correctly for any table type.
-- `PartitionedDestination` and `DatabaseDestination` convert through `to_records` and
+- `Destination` and `DatabaseDestination` convert through `to_records` and
   `from_records`.
 - `DatabaseDestination.read_representation` and `@destination(read_representation=...)` name
   the representation reads should materialize into.

docs/guide/destinations.md

@@ -88,14 +88,15 @@ Every `read()` and `write()` receives an immutable `IOContext`:
 | Field | Meaning |
 |-------|---------|
 | `asset` | The asset being read or written. `asset.table`, `asset.dataset`, `asset.partitioning` name the storage location. |
-| `partition_or_window` | The scope of this call, or `None` for an unpartitioned asset. |
+| `partition_or_window` | The partition or window of this call, or `None` for an unpartitioned asset. |
 | `schema` | The effective schema of the data: the declared one, or the one inferred during conform. `None` when none could be resolved. |
 | `metadata` | Run metadata (run id, backfill id). |
 
 ## Custom destinations
 
-Subclass `il.Destination`, or decorate a plain class with `@il.destination`, and implement
-`read()` and `write()`. Both may be sync or `async def`:
+A destination stores data one **partition** at a time, `None` standing for the whole of an
+unpartitioned asset. Subclass `il.Destination`, or decorate a plain class with `@il.destination`,
+and implement the two partition hooks. Both may be sync or `async def`:
 
 ```py
 import json
@@ -108,19 +109,28 @@ import interloper as il
 class JSONDestination(il.Destination):
     base_path: str = ""
 
-    def write(self, context: il.IOContext, data: Any) -> None:
-        path = Path(self.base_path) / context.asset.dataset / f"{context.asset.table}.json"
+    def _path(self, context: il.IOContext, partition: il.Partition | None) -> Path:
+        base = Path(self.base_path) / (context.asset.dataset or "") / context.asset.table
+        if partition is None:
+            return base / "data.json"
+        return base / f"{context.asset.partitioning.column}={partition.id}" / "data.json"
+
+    def write_partition(self, context: il.IOContext, partition: il.Partition | None, data: Any) -> None:
+        path = self._path(context, partition)
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(json.dumps(data, default=str))
 
-    def read(self, context: il.IOContext) -> Any:
-        path = Path(self.base_path) / context.asset.dataset / f"{context.asset.table}.json"
-        return json.loads(path.read_text())
+    def read_partition(self, context: il.IOContext, partition: il.Partition | None) -> Any:
+        return json.loads(self._path(context, partition).read_text())
 ```
 
-This form writes one file per asset and ignores `context.partition_or_window`: a partition
-rewrite replaces every partition and a window lands in one file. It suits unpartitioned assets;
-partitioned ones belong on `il.PartitionedDestination` below.
+`write()` and `read()` are the base class's: a window write is split into one `write_partition` call
+per partition, slicing the data through its [representation](../extending/representations.md)
+on the partition column; a window read returns one result per partition, newest first. A
+destination written this way is partition-correct by construction, and `CSVDestination`,
+`FileDestination`, `MemoryDestination` and `GCSDestination` are all built exactly like this. The
+partitions a context covers are on the context itself, `context.partitions` and `context.slices(data)`,
+for a backend that needs them.
 
 The decorator accepts the class's public ClassVars and field defaults, plus `relations=`; see the
 [decorators reference](../reference/decorators.md). A destination's own connection is a relation,
@@ -130,32 +140,15 @@ declared as an annotation or through `relations=`; see
 Override `partition_row_counts(context)` to report rows per partition; `asset.partition_row_counts()`
 and coverage tooling call it.
 
-### Partition-aware destinations
-
-`il.PartitionedDestination` implements the partition dispatch once. Subclasses implement two
-scope hooks and are partition-correct by construction:
-
-```py
-class JSONDestination(il.PartitionedDestination):
-    base_path: str = ""
-
-    def _write_scope(self, context, partition, data) -> None:
-        # partition is None for the unpartitioned whole
-        ...
-
-    def _read_scope(self, context, partition):
-        ...
-```
-
-A window write is split into one `_write_scope` call per partition, slicing the data through
-its [representation](../extending/representations.md) on the partition column. A window read
-returns one result per partition, newest first. `CSVDestination` and `MemoryDestination` are
-built this way.
+A backend whose storage is not per partition may override `write()` or `read()`
+wholesale. `DatabaseDestination` below does that for writes: rows carry the partition column, so
+a window clears every partition it covers and inserts the whole batch once.
 
 ### Database destinations
 
 `DatabaseDestination` (imported from `interloper.destination`, together with
-`WriteDisposition`) targets stores addressed by table and schema. Reads and writes reduce to a
+`WriteDisposition`) targets stores addressed by table and schema. Its `write_partition` clears one
+partition and inserts; its `write` batches a window into one insert. Reads and writes reduce to a
 small set of row operations:
 
 | Hook | Called for |
@@ -176,7 +169,7 @@ representation directly (a DataFrame into a Parquet load job) using `context.sch
 Behaviour the base class owns:
 
 - **Write disposition**: `write_disposition = WriteDisposition.REPLACE` (default) deletes the
-  matching scope before inserting; `APPEND` never deletes. A class attribute, not a field.
+  matching partition before inserting; `APPEND` never deletes. A class attribute, not a field.
 - **Time partitions are scoped by bounds**, not by equality, because rows of a monthly partition
   carry daily dates.
 - **Read representation**: rows are materialized into the representation named by

packages/interloper-core/src/interloper/__init__.py

@@ -26,7 +26,6 @@
     FileDestination,
     IOContext,
     MemoryDestination,
-    PartitionedDestination,
     destination,
 )
 from interloper.events import Event, EventBus, EventType
@@ -137,7 +136,6 @@
     "Partition",
     "PartitionConfig",
     "PartitionWindow",
-    "PartitionedDestination",
     "RESTClient",
     "RangePaginator",
     "RefreshTokenOAuthConnection",

packages/interloper-core/src/interloper/destination/__init__.py

@@ -7,7 +7,6 @@
 from interloper.destination.decorator import destination
 from interloper.destination.file import FileDestination
 from interloper.destination.memory import MemoryDestination
-from interloper.destination.partitioned import PartitionedDestination
 
 __all__ = [
     "CSVDestination",
@@ -17,7 +16,6 @@
     "FileDestination",
     "IOContext",
     "MemoryDestination",
-    "PartitionedDestination",
     "WriteDisposition",
     "destination",
 ]

packages/interloper-core/src/interloper/destination/base.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-from abc import abstractmethod
 from typing import Any, ClassVar
 
 from interloper.component import Component, ComponentDefinition
 from interloper.destination.context import IOContext
+from interloper.partitioning.base import Partition
 from interloper.utils.text import to_label
 
 
@@ -20,24 +20,33 @@ class DestinationDefinition(ComponentDefinition):
 
 
 class Destination(Component):
-    """A component that reads and writes asset data.
-
-    Subclass and implement ``read()`` and ``write()``. They may be written
-    as plain sync methods (the common case: most warehouse/file clients are
-    sync) or as ``async def`` for native async I/O (e.g. asyncpg, aiofiles).
-    The engine is async-native: it awaits async implementations directly and
-    offloads sync ones to a worker thread, so a destination never blocks the
-    event loop either way. An annotation naming a component class declares a
+    """A component that reads and writes asset data, one partition at a time.
+
+    A destination stores data per **partition**, ``None`` standing for the
+    whole of an unpartitioned asset. Subclass and implement
+    :meth:`write_partition` and :meth:`read_partition` for a single one;
+    :meth:`write` and :meth:`read` own the rest, splitting a window write into
+    one call per partition and gathering a window read into one result per
+    partition, so a destination is partition-correct by construction. A
+    backend that does not store per partition (a database that clears each
+    partition and inserts a window in one batch) overrides :meth:`write` or
+    :meth:`read` instead.
+
+    The hooks may be plain sync methods (the common case: most warehouse and
+    file clients are sync) or ``async def`` for native async I/O. The engine
+    is async-native: it awaits async implementations directly and offloads
+    sync ones to a worker thread, so a destination never blocks the event
+    loop either way. An annotation naming a component class declares a
     relation, which the destination resolves by name::
 
-        class PostgresDestination(Destination):
-            connection: PostgresConnection
+        class JSONDestination(Destination):
+            connection: BucketConnection
 
-            def read(self, context: IOContext) -> Any:
-                return query_table(self.connection.connection_string, context.table)
+            def write_partition(self, context: IOContext, partition: Partition | None, data: Any) -> None:
+                self.connection.put(self._path(context, partition), json.dumps(data, default=str))
 
-            def write(self, context: IOContext, data: Any) -> None:
-                insert_into(self.connection.connection_string, context.table, data)
+            def read_partition(self, context: IOContext, partition: Partition | None) -> Any:
+                return json.loads(self.connection.get(self._path(context, partition)))
     """
 
     tags: ClassVar[list[str]] = []
@@ -68,25 +77,63 @@ def definition(cls) -> DestinationDefinition:
             relations=dict(cls.relations),
         )
 
-    @abstractmethod
-    def read(self, context: IOContext) -> Any:
-        """Read data from this destination.
+    def write_partition(self, context: IOContext, partition: Partition | None, data: Any) -> None:
+        """Store *data* for one partition.
 
         Args:
-            context: Destination context with asset, partition, and metadata.
+            context: IO context carrying the target asset and the effective schema.
+            partition: The partition being stored, or ``None`` for the
+                unpartitioned whole.
+            data: The partition's slice of the data to store.
+
+        Raises:
+            NotImplementedError: Every destination implements this, unless it
+                overrides :meth:`write` for storage that is not per partition.
+        """
+        raise NotImplementedError(f"{type(self).__name__} must implement write_partition()")
 
-        Returns:
-            The data read from the destination.
+    def read_partition(self, context: IOContext, partition: Partition | None) -> Any:
+        """Load one partition.
+
+        Args:
+            context: IO context carrying the target asset and the effective schema.
+            partition: The partition to load, or ``None`` for the unpartitioned
+                whole.
+
+        Raises:
+            NotImplementedError: Every destination implements this, unless it
+                overrides :meth:`read` for storage that is not per partition.
         """
+        raise NotImplementedError(f"{type(self).__name__} must implement read_partition()")
 
-    @abstractmethod
     def write(self, context: IOContext, data: Any) -> None:
-        """Write data to this destination.
+        """Write data, one partition at a time.
+
+        A window is split into one :meth:`write_partition` call per partition,
+        each receiving its slice of the data; a single partition or the
+        unpartitioned whole is one call receiving the data as is.
 
         Args:
-            context: Destination context with asset, partition, and metadata.
-            data: The data to write.
+            context: IO context carrying the target asset, the partition or window,
+                and the effective schema.
+            data: The data to write, in its native representation.
+        """
+        for partition, chunk in context.slices(data):
+            self.write_partition(context, partition, chunk)
+
+    def read(self, context: IOContext) -> Any:
+        """Read data for the context's partition, or window.
+
+        Args:
+            context: IO context carrying the target asset, the partition or window,
+                and the effective schema.
+
+        Returns:
+            The partition's data; a window returns one result per partition, in
+            window order.
         """
+        results = [self.read_partition(context, partition) for partition in context.partitions]
+        return results if context.window else results[0]
 
     def partition_row_counts(self, context: IOContext) -> dict[str, int]:
         """Return row counts grouped by the asset's partition column.

packages/interloper-core/src/interloper/destination/context.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any
 
+from interloper.errors import ConfigError
 from interloper.partitioning.base import Partition, PartitionWindow
 
 if TYPE_CHECKING:
@@ -16,16 +17,69 @@
 class IOContext:
     """Immutable context passed to :meth:`Destination.read` and :meth:`Destination.write`.
 
-    Carries the target asset, optional partition scope, and arbitrary metadata
+    Carries the target asset, optional partition or window, and arbitrary metadata
     so that destination implementations can resolve the correct storage location.
 
-    ``schema`` is the *effective* schema of the data being written or read —
+    ``schema`` is the *effective* schema of the data being written or read:
     the asset's declared schema when set, otherwise the schema inferred during
-    conform.  Destinations use it for DDL, typed load jobs, and restoring
-    types on read.  ``None`` when no schema could be resolved.
+    conform. Destinations use it for DDL, typed load jobs, and restoring
+    types on read. ``None`` when no schema could be resolved.
+
+    A destination stores data per **partition**, ``None`` standing for the
+    unpartitioned whole; :attr:`partitions` and :meth:`slices` spell that out
+    for the three shapes ``partition_or_window`` can take, so no destination
+    branches on them itself.
     """
 
     asset: Asset
     partition_or_window: Partition | PartitionWindow | None = None
     metadata: dict[str, Any] = field(default_factory=dict)
     schema: type[Schema] | None = None
+
+    @property
+    def window(self) -> bool:
+        """Whether this context spans several partitions.
+
+        Returns:
+            True for a partition window, False for one partition or the
+            unpartitioned whole.
+        """
+        return isinstance(self.partition_or_window, PartitionWindow)
+
+    @property
+    def partitions(self) -> list[Partition | None]:
+        """The partitions this context covers, in window order.
+
+        Returns:
+            ``[None]`` for the unpartitioned whole, ``[partition]`` for one
+            partition, one entry per partition for a window.
+        """
+        target = self.partition_or_window
+        if isinstance(target, PartitionWindow):
+            return list(target)
+        return [target]
+
+    def slices(self, data: Any) -> list[tuple[Partition | None, Any]]:
+        """Pair each partition with its slice of *data*.
+
+        A window's data is split on the asset's partition column through the
+        data's representation; a single partition, or the whole, receives the
+        data as is. Data whose representation is not recognised cannot be
+        split and is handed to every partition as is.
+
+        Args:
+            data: The data being written, in its native representation.
+
+        Returns:
+            ``(partition, slice)`` pairs, one per partition, in :attr:`partitions` order.
+
+        Raises:
+            ConfigError: If a window is written for an asset that declares no
+                partitioning, since nothing says which column to slice on.
+        """
+        if not self.window:
+            return [(partition, data) for partition in self.partitions]
+        if self.asset.partitioning is None:
+            raise ConfigError(f"Asset '{self.asset.key}' is not partitioned and cannot be written over a window")
+        column = self.asset.partitioning.column
+        return [(partition, partition.slice(data, column)) for partition in self.partitions if partition is not None]