Repository: artigraph/artigraph (language: python)

Issue:
Override `arti.internal.models.Model.copy` to perform full validation
We call `Model.copy` in a few places in the code. It's mostly "trusted", but during field renames, the string keys can sometimes get out of line and don't error loud _early_. Or, we might assign the wrong type of data/bypass some conversions that we'd like to apply.

The pydantic GH issues have a few suggested patterns, but mostly boil down to converting the model to a dict, applying the updates, and then reinstantiating.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 78f7468c..e9b14a7c 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -6,7 +6,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [macos-latest, ubuntu-latest, windows-latest]
+        os: [macos-latest, ubuntu-latest]
         python-version: [3.9.8, 3.10.0]
     runs-on: ${{ matrix.os }}
     env:
diff --git a/poetry.lock b/poetry.lock
index 3d7d6d6c..0ef1c22e 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -802,7 +802,7 @@ test = ["shellingham (>=1.3.0,<2.0.0)", "pytest (>=4.4.0,<5.4.0)", "pytest-cov (
 name = "types-pyfarmhash"
 version = "0.2.2"
 description = "Typing stubs for pyfarmhash"
-category = "main"
+category = "dev"
 optional = false
 python-versions = "*"
 
@@ -852,7 +852,7 @@ python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,>=2.7"
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.9"
-content-hash = "9226998f690392cc8273887d5060b5a1f464c9c4ceaddaeb8cd235a6abf9cd04"
+content-hash = "2fa622a8e04055070949328ee4cb58d2c0a95948651e7d18c613a8ae181f3661"
 
 [metadata.files]
 appnope = [
@@ -1239,11 +1239,6 @@ regex = [
     {file = "regex-2021.11.10-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:30ab804ea73972049b7a2a5c62d97687d69b5a60a67adca07eb73a0ddbc9e29f"},
     {file = "regex-2021.11.10-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:68a067c11463de2a37157930d8b153005085e42bcb7ad9ca562d77ba7d1404e0"},
     {file = "regex-2021.11.10-cp310-cp310-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:162abfd74e88001d20cb73ceaffbfe601469923e875caf9118333b1a4aaafdc4"},
-    {file = "regex-2021.11.10-cp310-cp310-musllinux_1_1_aarch64.whl", hash = "sha256:b9ed0b1e5e0759d6b7f8e2f143894b2a7f3edd313f38cf44e1e15d360e11749b"},
-    {file = "regex-2021.11.10-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:473e67837f786404570eae33c3b64a4b9635ae9f00145250851a1292f484c063"},
-    {file = "regex-2021.11.10-cp310-cp310-musllinux_1_1_ppc64le.whl", hash = "sha256:2fee3ed82a011184807d2127f1733b4f6b2ff6ec7151d83ef3477f3b96a13d03"},
-    {file = "regex-2021.11.10-cp310-cp310-musllinux_1_1_s390x.whl", hash = "sha256:d5fd67df77bab0d3f4ea1d7afca9ef15c2ee35dfb348c7b57ffb9782a6e4db6e"},
-    {file = "regex-2021.11.10-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:5d408a642a5484b9b4d11dea15a489ea0928c7e410c7525cd892f4d04f2f617b"},
     {file = "regex-2021.11.10-cp310-cp310-win32.whl", hash = "sha256:98ba568e8ae26beb726aeea2273053c717641933836568c2a0278a84987b2a1a"},
     {file = "regex-2021.11.10-cp310-cp310-win_amd64.whl", hash = "sha256:780b48456a0f0ba4d390e8b5f7c661fdd218934388cde1a974010a965e200e12"},
     {file = "regex-2021.11.10-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:dba70f30fd81f8ce6d32ddeef37d91c8948e5d5a4c63242d16a2b2df8143aafc"},
@@ -1253,11 +1248,6 @@ regex = [
     {file = "regex-2021.11.10-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5537f71b6d646f7f5f340562ec4c77b6e1c915f8baae822ea0b7e46c1f09b733"},
     {file = "regex-2021.11.10-cp36-cp36m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:ed2e07c6a26ed4bea91b897ee2b0835c21716d9a469a96c3e878dc5f8c55bb23"},
     {file = "regex-2021.11.10-cp36-cp36m-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:ca5f18a75e1256ce07494e245cdb146f5a9267d3c702ebf9b65c7f8bd843431e"},
-    {file = "regex-2021.11.10-cp36-cp36m-musllinux_1_1_aarch64.whl", hash = "sha256:74cbeac0451f27d4f50e6e8a8f3a52ca074b5e2da9f7b505c4201a57a8ed6286"},
-    {file = "regex-2021.11.10-cp36-cp36m-musllinux_1_1_i686.whl", hash = "sha256:3598893bde43091ee5ca0a6ad20f08a0435e93a69255eeb5f81b85e81e329264"},
-    {file = "regex-2021.11.10-cp36-cp36m-musllinux_1_1_ppc64le.whl", hash = "sha256:50a7ddf3d131dc5633dccdb51417e2d1910d25cbcf842115a3a5893509140a3a"},
-    {file = "regex-2021.11.10-cp36-cp36m-musllinux_1_1_s390x.whl", hash = "sha256:61600a7ca4bcf78a96a68a27c2ae9389763b5b94b63943d5158f2a377e09d29a"},
-    {file = "regex-2021.11.10-cp36-cp36m-musllinux_1_1_x86_64.whl", hash = "sha256:563d5f9354e15e048465061509403f68424fef37d5add3064038c2511c8f5e00"},
     {file = "regex-2021.11.10-cp36-cp36m-win32.whl", hash = "sha256:93a5051fcf5fad72de73b96f07d30bc29665697fb8ecdfbc474f3452c78adcf4"},
     {file = "regex-2021.11.10-cp36-cp36m-win_amd64.whl", hash = "sha256:b483c9d00a565633c87abd0aaf27eb5016de23fed952e054ecc19ce32f6a9e7e"},
     {file = "regex-2021.11.10-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:fff55f3ce50a3ff63ec8e2a8d3dd924f1941b250b0aac3d3d42b687eeff07a8e"},
@@ -1267,11 +1257,6 @@ regex = [
     {file = "regex-2021.11.10-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d5ca078bb666c4a9d1287a379fe617a6dccd18c3e8a7e6c7e1eb8974330c626a"},
     {file = "regex-2021.11.10-cp37-cp37m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:dd33eb9bdcfbabab3459c9ee651d94c842bc8a05fabc95edf4ee0c15a072495e"},
     {file = "regex-2021.11.10-cp37-cp37m-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:05b7d6d7e64efe309972adab77fc2af8907bb93217ec60aa9fe12a0dad35874f"},
-    {file = "regex-2021.11.10-cp37-cp37m-musllinux_1_1_aarch64.whl", hash = "sha256:42b50fa6666b0d50c30a990527127334d6b96dd969011e843e726a64011485da"},
-    {file = "regex-2021.11.10-cp37-cp37m-musllinux_1_1_i686.whl", hash = "sha256:6e1d2cc79e8dae442b3fa4a26c5794428b98f81389af90623ffcc650ce9f6732"},
-    {file = "regex-2021.11.10-cp37-cp37m-musllinux_1_1_ppc64le.whl", hash = "sha256:0416f7399e918c4b0e074a0f66e5191077ee2ca32a0f99d4c187a62beb47aa05"},
-    {file = "regex-2021.11.10-cp37-cp37m-musllinux_1_1_s390x.whl", hash = "sha256:ce298e3d0c65bd03fa65ffcc6db0e2b578e8f626d468db64fdf8457731052942"},
-    {file = "regex-2021.11.10-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:dc07f021ee80510f3cd3af2cad5b6a3b3a10b057521d9e6aaeb621730d320c5a"},
     {file = "regex-2021.11.10-cp37-cp37m-win32.whl", hash = "sha256:e71255ba42567d34a13c03968736c5d39bb4a97ce98188fafb27ce981115beec"},
     {file = "regex-2021.11.10-cp37-cp37m-win_amd64.whl", hash = "sha256:07856afef5ffcc052e7eccf3213317fbb94e4a5cd8177a2caa69c980657b3cb4"},
     {file = "regex-2021.11.10-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:ba05430e819e58544e840a68b03b28b6d328aff2e41579037e8bab7653b37d83"},
@@ -1282,11 +1267,6 @@ regex = [
     {file = "regex-2021.11.10-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:85bfa6a5413be0ee6c5c4a663668a2cad2cbecdee367630d097d7823041bdeec"},
     {file = "regex-2021.11.10-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:f23222527b307970e383433daec128d769ff778d9b29343fb3496472dc20dabe"},
     {file = "regex-2021.11.10-cp38-cp38-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:da1a90c1ddb7531b1d5ff1e171b4ee61f6345119be7351104b67ff413843fe94"},
-    {file = "regex-2021.11.10-cp38-cp38-musllinux_1_1_aarch64.whl", hash = "sha256:f5be7805e53dafe94d295399cfbe5227f39995a997f4fd8539bf3cbdc8f47ca8"},
-    {file = "regex-2021.11.10-cp38-cp38-musllinux_1_1_i686.whl", hash = "sha256:a955b747d620a50408b7fdf948e04359d6e762ff8a85f5775d907ceced715129"},
-    {file = "regex-2021.11.10-cp38-cp38-musllinux_1_1_ppc64le.whl", hash = "sha256:139a23d1f5d30db2cc6c7fd9c6d6497872a672db22c4ae1910be22d4f4b2068a"},
-    {file = "regex-2021.11.10-cp38-cp38-musllinux_1_1_s390x.whl", hash = "sha256:ca49e1ab99593438b204e00f3970e7a5f70d045267051dfa6b5f4304fcfa1dbf"},
-    {file = "regex-2021.11.10-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:96fc32c16ea6d60d3ca7f63397bff5c75c5a562f7db6dec7d412f7c4d2e78ec0"},
     {file = "regex-2021.11.10-cp38-cp38-win32.whl", hash = "sha256:0617383e2fe465732af4509e61648b77cbe3aee68b6ac8c0b6fe934db90be5cc"},
     {file = "regex-2021.11.10-cp38-cp38-win_amd64.whl", hash = "sha256:a3feefd5e95871872673b08636f96b61ebef62971eab044f5124fb4dea39919d"},
     {file = "regex-2021.11.10-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:f7f325be2804246a75a4f45c72d4ce80d2443ab815063cdf70ee8fb2ca59ee1b"},
@@ -1297,11 +1277,6 @@ regex = [
     {file = "regex-2021.11.10-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:962b9a917dd7ceacbe5cd424556914cb0d636001e393b43dc886ba31d2a1e449"},
     {file = "regex-2021.11.10-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:fa8c626d6441e2d04b6ee703ef2d1e17608ad44c7cb75258c09dd42bacdfc64b"},
     {file = "regex-2021.11.10-cp39-cp39-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:3c5fb32cc6077abad3bbf0323067636d93307c9fa93e072771cf9a64d1c0f3ef"},
-    {file = "regex-2021.11.10-cp39-cp39-musllinux_1_1_aarch64.whl", hash = "sha256:cd410a1cbb2d297c67d8521759ab2ee3f1d66206d2e4328502a487589a2cb21b"},
-    {file = "regex-2021.11.10-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:e6096b0688e6e14af6a1b10eaad86b4ff17935c49aa774eac7c95a57a4e8c296"},
-    {file = "regex-2021.11.10-cp39-cp39-musllinux_1_1_ppc64le.whl", hash = "sha256:529801a0d58809b60b3531ee804d3e3be4b412c94b5d267daa3de7fadef00f49"},
-    {file = "regex-2021.11.10-cp39-cp39-musllinux_1_1_s390x.whl", hash = "sha256:0f594b96fe2e0821d026365f72ac7b4f0b487487fb3d4aaf10dd9d97d88a9737"},
-    {file = "regex-2021.11.10-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:2409b5c9cef7054dde93a9803156b411b677affc84fca69e908b1cb2c540025d"},
     {file = "regex-2021.11.10-cp39-cp39-win32.whl", hash = "sha256:3b5df18db1fccd66de15aa59c41e4f853b5df7550723d26aa6cb7f40e5d9da5a"},
     {file = "regex-2021.11.10-cp39-cp39-win_amd64.whl", hash = "sha256:83ee89483672b11f8952b158640d0c0ff02dc43d9cb1b70c1564b49abe92ce29"},
     {file = "regex-2021.11.10.tar.gz", hash = "sha256:f341ee2df0999bfdf7a95e448075effe0db212a59387de1a70690e4acb03d4c6"},
diff --git a/pyproject.toml b/pyproject.toml
index 20cde0bb..b877f1a6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -12,6 +12,7 @@ target-version = ['py38']
 exclude_lines = [
     "@overload",
     "if TYPE_CHECKING:",
+    "if not TYPE_CHECKING:",
     "pragma: no cover",
     "raise NotImplemented",
     "raise NotImplementedError",
@@ -20,8 +21,9 @@ fail_under = 100
 skip_covered = true
 
 [tool.coverage.run]
-source = ["arti"]
 branch = true
+omit = ["src/arti/internal/vendored/*" ]
+source = ["arti"]
 
 
 [tool.isort] # Black compatible config
@@ -57,7 +59,6 @@ pyfarmhash = "^0.2.2"
 python = "^3.9"
 python-box = "^5.4.1"
 sgqlc = "^14.1"
-types-pyfarmhash = "^0.2.1"
 
 [tool.poetry.dev-dependencies]
 black = "21.11b1"
@@ -79,6 +80,7 @@ pytest-mypy = "0.8.1"
 pytest-sugar = "0.9.4"
 sh = "^1.14.2"
 typer = "^0.4.0"
+types-pyfarmhash = "^0.2.1"
 
 
 [tool.pylint.format]
diff --git a/src/arti/__init__.py b/src/arti/__init__.py
new file mode 100644
index 00000000..1499f51c
--- /dev/null
+++ b/src/arti/__init__.py
@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
+
+import threading
+from typing import TYPE_CHECKING, Optional
+
+if TYPE_CHECKING:
+    from arti.graphs import Graph
+
+
+class _Context(threading.local):
+    def __init__(self) -> None:
+        super().__init__()
+        self.graph: Optional[Graph] = None
+
+
+context = _Context()
diff --git a/src/arti/artifacts/__init__.py b/src/arti/artifacts/__init__.py
index 67efb850..58bbd2e1 100644
--- a/src/arti/artifacts/__init__.py
+++ b/src/arti/artifacts/__init__.py
@@ -9,7 +9,9 @@
 from arti.annotations import Annotation
 from arti.formats import Format
 from arti.internal.models import Model
-from arti.storage import Storage
+from arti.internal.utils import classproperty
+from arti.partitions import CompositeKeyTypes, PartitionKey
+from arti.storage import InputFingerprints, Storage, StoragePartition
 from arti.types import Type
 
 
@@ -22,10 +24,11 @@ class BaseArtifact(Model):
     - storage: the data's persistent storage system, such as blob storage, database native, etc.
     """
 
-    # is_scalar denotes whether this Artifacts represents a *single* value of the specified type or
-    # a *collection*. Namely, even if the type is a Struct(...), but there is only one, it will be
-    # scalar for our purposes.
-    is_scalar: ClassVar[bool]
+    # NOTE: Narrow the fields that affect the fingerprint to minimize changes (which trigger
+    # recompute). Importantly, avoid fingerprinting the `.producer_output` (ie: the *upstream*
+    # producer) to prevent cascading fingerprint changes (Producer.fingerprint accesses the *input*
+    # Artifact.fingerprints). Even so, this may still be quite sensitive.
+    _fingerprint_includes_ = frozenset(["type", "format", "storage"])
 
     # Type *must* be set on the class and be rather static - small additions may be necessary at
     # Graph level (eg: dynamic column additions), but these should be minor. We might allow Struct
@@ -39,10 +42,35 @@ class BaseArtifact(Model):
     format: Format
     storage: Storage[Any]
 
-    # Hide the producer to prevent showing the entire upstream graph
-    producer: Optional["Producer"] = Field(None, repr=False)
+    # Hide in repr to prevent showing the entire upstream graph.
+    #
+    # ProducerOutput is a ForwardRef/cyclic import. Quote the entire hint to force full resolution
+    # during `.update_forward_refs`, rather than `Optional[ForwardRef("ProducerOutput")]`.
+    producer_output: "Optional[ProducerOutput]" = Field(None, repr=False)
+
+    # Class level alias for `type`, which must be set on (non-abstract) subclasses.
+    #
+    # Pydantic removes class defaults and stashes them in cls.__fields__. To ease access, we
+    # automatically populate this from `type` in `__init_subclass__`.
+    _type: ClassVar[Type]
 
-    # TODO: Allow smarter type/format/storage merging w/ the default?
+    @classmethod
+    def __init_subclass__(cls, **kwargs: Any) -> None:
+        super().__init_subclass__(**kwargs)
+        if not cls._abstract_ and cls.__fields__["type"].default is None:
+            raise ValueError(f"{cls.__name__} must set `type`")
+        cls._type = cls.__fields__["type"].default
+
+    @validator("type", always=True)
+    @classmethod
+    def _validate_type(cls, type_: Type) -> Type:
+        if type_ != cls._type:
+            # NOTE: We do a lot of class level validation (particularly in Producer) that relies on
+            # the *class* type, such as partition key validation. It's possible we could loosen this
+            # a bit by allowing *new* Struct fields, but still requiring an exact match for other
+            # Types (including existing Struct fields).
+            raise ValueError("overriding `type` is not supported")
+        return type_
 
     @validator("format", always=True)
     @classmethod
@@ -56,15 +84,34 @@ def _validate_format(cls, format: Format, values: dict[str, Any]) -> Format:
     def _validate_storage(cls, storage: Storage[Any], values: dict[str, Any]) -> Storage[Any]:
         if "type" in values and "format" in values:
             storage.supports(type_=values["type"], format=values["format"])
+            storage = storage.resolve_partition_key_spec(cls.partition_key_types).resolve_extension(
+                values["format"].extension
+            )
         return storage
 
+    @classproperty
+    @classmethod
+    def partition_key_types(cls) -> CompositeKeyTypes:
+        return PartitionKey.types_from(cls._type)
+
+    @classproperty
+    @classmethod
+    def is_partitioned(cls) -> bool:
+        return bool(cls.partition_key_types)
+
+    def discover_storage_partitions(
+        self, input_fingerprints: InputFingerprints = InputFingerprints()
+    ) -> tuple[StoragePartition, ...]:
+        # TODO: Should we support calculating the input fingerprints if not passed?
+        return self.storage.discover_partitions(self.partition_key_types, input_fingerprints)
+
 
 class Statistic(BaseArtifact):
     """A Statistic is a piece of data derived from an Artifact that can be tracked over time."""
 
-    # TODO: Set format/storage to some "system default" that can be used across backends.
+    # TODO: Set format/storage to some "system default" that can be used across backends?
 
-    is_scalar = True
+    _abstract_ = True
 
 
 class Artifact(BaseArtifact):
@@ -80,8 +127,7 @@ class Artifact(BaseArtifact):
     over time).
     """
 
-    # Artifacts are collections by default (think database tables, etc), but may be overridden.
-    is_scalar: ClassVar[bool] = False
+    _abstract_ = True
 
     annotations: tuple[Annotation, ...] = ()
     statistics: tuple[Statistic, ...] = ()
@@ -102,6 +148,8 @@ def cast(cls, value: Any) -> "Artifact":
         - a Producer instance with a multiple output Artifacts, an error is raised
         - other types, an error is raised
         """
+        from arti.producers import Producer
+
         # TODO: Leverage a TypeSystem("python") to cast to Artifact classes with "backend native"
         # storage to support builtin assignment and custom type registration.
         if isinstance(value, Artifact):
@@ -119,11 +167,9 @@ def cast(cls, value: Any) -> "Artifact":
                 f"{type(value).__name__} produces {len(output_artifacts)} Artifacts. Try assigning each to a new name in the Graph!"
             )
 
-        raise NotImplementedError("Casting python objects to Artifacts is not implemented yet!")
-
+        raise NotImplementedError(
+            f"Casting python objects ({value}) to Artifacts is not implemented yet!"
+        )
 
-from arti.producers import Producer  # noqa: E402 # # pylint: disable=wrong-import-position
 
-BaseArtifact.update_forward_refs()
-Statistic.update_forward_refs()
-Artifact.update_forward_refs()
+from arti.producers import ProducerOutput  # noqa: E402
diff --git a/src/arti/backends/__init__.py b/src/arti/backends/__init__.py
index 95b2a0e1..a888b952 100644
--- a/src/arti/backends/__init__.py
+++ b/src/arti/backends/__init__.py
@@ -1,7 +1,31 @@
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
 
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Protocol
 
-class Backend:
+from arti.fingerprints import Fingerprint
+from arti.internal.models import Model
+from arti.storage import StoragePartitions
+
+
+# TODO: Consider adding CRUD methods for "everything"?
+#
+# Likely worth making a distinction between lower level ("CRUD") methods vs higher level ("RPC" or
+# "composing") methods. Executors should operate on the high level methods, but those may have
+# defaults simply calling the lower level methods. If high level methods can be optimized (eg: not a
+# bunch of low level calls, each own network call), Backend can override.
+class BackendProtocol(Protocol):
+    def read_graph_partitions(self, graph_id: Fingerprint, name: str) -> StoragePartitions:
+        raise NotImplementedError()
+
+    def write_graph_partitions(
+        self, graph_id: Fingerprint, name: str, partitions: StoragePartitions
+    ) -> None:
+        raise NotImplementedError()
+
+
+class Backend(Model):
     """Backend represents a storage for internal Artigraph metadata.
 
     Backend storage is an addressable location (local path, database connection, etc) that
@@ -11,3 +35,7 @@ class Backend:
     - Artifact and Producer Fingerprints
     - etc
     """
+
+    @contextmanager
+    def connect(self) -> Iterator[BackendProtocol]:
+        raise NotImplementedError()
diff --git a/src/arti/backends/memory.py b/src/arti/backends/memory.py
new file mode 100644
index 00000000..60b76cb8
--- /dev/null
+++ b/src/arti/backends/memory.py
@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Iterator
+from contextlib import contextmanager
+from itertools import chain
+
+from pydantic import Field
+
+from arti.backends import Backend
+from arti.fingerprints import Fingerprint
+from arti.storage import StoragePartitions
+
+
+class MemoryBackend(Backend):
+    graph_partitions: dict[Fingerprint, dict[str, StoragePartitions]] = Field(
+        default_factory=lambda: defaultdict(lambda: defaultdict(StoragePartitions)), repr=False
+    )
+
+    @contextmanager
+    def connect(self) -> Iterator[MemoryBackend]:
+        yield self
+
+    def read_graph_partitions(self, graph_id: Fingerprint, name: str) -> StoragePartitions:
+        return self.graph_partitions[graph_id][name]
+
+    def write_graph_partitions(
+        self, graph_id: Fingerprint, name: str, partitions: StoragePartitions
+    ) -> None:
+        # NOTE: Be careful about deduping, otherwise we might have dup reads.
+        self.graph_partitions[graph_id][name] = StoragePartitions(
+            set(
+                chain(
+                    self.graph_partitions[graph_id][name],
+                    (
+                        partition.with_content_fingerprint(keep_existing=True)
+                        for partition in partitions
+                    ),
+                )
+            )
+        )
diff --git a/src/arti/executors/__init__.py b/src/arti/executors/__init__.py
new file mode 100644
index 00000000..c55712ba
--- /dev/null
+++ b/src/arti/executors/__init__.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
+
+import abc
+
+from arti.graphs import Graph
+from arti.internal.models import Model
+
+
+class Executor(Model):
+    @abc.abstractmethod
+    def build(self, graph: Graph) -> None:
+        raise NotImplementedError()
diff --git a/src/arti/executors/local.py b/src/arti/executors/local.py
new file mode 100644
index 00000000..2c4dc43e
--- /dev/null
+++ b/src/arti/executors/local.py
@@ -0,0 +1,146 @@
+import logging
+from graphlib import TopologicalSorter
+from itertools import chain
+
+from arti.artifacts import Artifact
+from arti.backends import BackendProtocol
+from arti.executors import Executor
+from arti.fingerprints import Fingerprint
+from arti.graphs import Graph
+from arti.producers import Producer
+from arti.storage import InputFingerprints
+
+# TODO: Factor this code out to reusable helpers/better homes.
+#   - Perhaps a lot of the _build_producer logic can live in a few Producer methods. We
+#   still want to expose each partition-to-build as a parallelizable unit of execution,
+#   so maybe one Producer method kicks out all to-be-built partitions+metadata, and then
+#   Executor can parallelize the `.build`.
+
+
+class LocalExecutor(Executor):
+    def _build_artifact(
+        self,
+        graph: Graph,
+        graph_id: Fingerprint,
+        backend: BackendProtocol,
+        artifact: Artifact,
+    ) -> None:
+        # NOTE: Should we do different things for "raw" vs produced artifacts?
+        #
+        # Eg: for raw we discover partitions, but for generated we... just
+        # lookup from the Backend somehow? For all, we should probably compute
+        # the statistics (or maybe compute should be part of the Producer step
+        # using the exiting Views?) and check thresholds. Threshold checking
+        # should apply for every run, but the statitics should no-op if already
+        # computed obviously.
+        #
+        # We can only discover raw Artifacts - all produced ones should be written   his will fail for generated Artifacts (missing input_fingerprints). So,
+        # maybe we only discover "if artifact.producer_output is None"
+        if artifact.producer_output is None:
+            backend.write_graph_partitions(
+                graph_id,
+                graph.artifact_to_key[artifact],
+                tuple(
+                    partition
+                    for partition in artifact.storage.discover_partitions(
+                        artifact.partition_key_types
+                    )
+                ),
+            )
+        # TODO: Check any assumptions about what/how many should be in the db?
+        # TODO: Calculate stats, run validation, etc...
+
+    # NOTE: This should probably be broken up to separate the .map and .build (ie: so we
+    # can map ahead of time and then parallelize build calls).
+    #
+    # TODO: Split into "see what partitions need built" and "build partition" so we can implement
+    # "sync" / "dry run" sort of things?
+    def _build_producer(
+        self,
+        graph: Graph,
+        graph_id: Fingerprint,
+        backend: BackendProtocol,
+        producer: Producer,
+    ) -> None:
+        input_partitions = {
+            name: backend.read_graph_partitions(graph_id, graph.artifact_to_key[artifact])
+            for name, artifact in producer.inputs.items()
+        }
+        partition_dependencies = producer.map(
+            **{
+                name: partitions
+                for name, partitions in input_partitions.items()
+                if name in producer._map_input_metadata_
+            }
+        )
+        # TODO: Need to validate the partition_dependencies against the Producer's
+        # partitioning scheme and such (basically, check user error). eg: if output is
+        # not partitioned, we expect only 1 entry in partition_dependencies
+        # (NotPartitioned).
+        partition_input_fingerprints = InputFingerprints(
+            {
+                composite_key: producer.compute_input_fingerprint(dependency_partitions)
+                for composite_key, dependency_partitions in partition_dependencies.items()
+            }
+        )
+        output_artifacts = graph.producer_outputs[producer]
+        existing_output_partitions = {
+            # NOTE: We'll append newly build partitions here as they are build.
+            output: output.discover_storage_partitions(
+                input_fingerprints=partition_input_fingerprints
+            )
+            for output in output_artifacts
+        }
+        for artifact, partitions in existing_output_partitions.items():
+            backend.write_graph_partitions(graph_id, graph.artifact_to_key[artifact], partitions)
+        # TODO: Guarantee all outputs have the same set of identified partitions
+        existing_output_keys = set(
+            partition.keys for partition in chain.from_iterable(existing_output_partitions.values())
+        )
+        for output_partition_key, dependencies in partition_dependencies.items():
+            if output_partition_key in existing_output_keys:
+                pk_str = f" for: {dict(output_partition_key)}" if output_partition_key else "."
+                logging.info(f"Skipping existing {type(producer).__name__} output{pk_str}")
+                continue
+            # TODO: Catch DispatchError and give a nicer error... maybe add this to our
+            # @dispatch wrapper (eg: msg arg, or even fn that returns the message to
+            # raise).
+            arguments = {
+                name: graph.read(
+                    artifact=producer.inputs[name],
+                    graph_id=graph_id,
+                    storage_partitions=partition_dependencies[output_partition_key][name],
+                    # TODO: We'll probably need to update Producer._build_input_views_
+                    # to hold *instances* (either objects set in Annotated or init
+                    # during Producer.__init_subclass__)
+                    view=view_class(),
+                )
+                for name, view_class in producer._build_input_views_.items()
+            }
+            outputs = producer.build(**arguments)
+            if len(producer._output_metadata_) == 1:
+                outputs = (outputs,)
+            for i, output in enumerate(outputs):
+                graph.write(
+                    output,
+                    artifact=output_artifacts[i],
+                    graph_id=graph_id,
+                    input_fingerprint=partition_input_fingerprints[output_partition_key],
+                    keys=output_partition_key,
+                    view=producer._output_metadata_[i][1](),
+                )
+
+    # TODO: Support "dry_run"?
+    def build(self, graph: Graph) -> None:
+        graph_id = graph.compute_id()
+        with graph.backend.connect() as backend:
+            for node in TopologicalSorter(graph.dependencies).static_order():
+                if isinstance(node, Artifact):
+                    logging.info(f"Syncing {type(node).__name__}")
+                    self._build_artifact(graph, graph_id, backend, node)
+                elif isinstance(node, Producer):
+                    logging.info(f"Building {type(node).__name__}")
+                    self._build_producer(graph, graph_id, backend, node)
+                else:
+                    raise NotImplementedError()
+        logging.info("Build finished.")
diff --git a/src/arti/fingerprints/__init__.py b/src/arti/fingerprints/__init__.py
index 260ceaca..5f64c033 100644
--- a/src/arti/fingerprints/__init__.py
+++ b/src/arti/fingerprints/__init__.py
@@ -1,8 +1,9 @@
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
 
+import operator
+from collections.abc import Callable
 from functools import reduce
-from operator import xor
-from typing import Optional
+from typing import Optional, Union
 
 import farmhash
 
@@ -10,6 +11,21 @@
 from arti.internal.utils import int64, uint64
 
 
+def _gen_fingerprint_binop(
+    op: Callable[[int, int], int]
+) -> "Callable[[Fingerprint, Union[int, Fingerprint]], Fingerprint]":
+    def _fingerprint_binop(self: "Fingerprint", other: "Union[int, Fingerprint]") -> "Fingerprint":
+        if isinstance(other, int):
+            other = Fingerprint.from_int(other)
+        if isinstance(other, Fingerprint):
+            if self.key is None or other.key is None:
+                return Fingerprint.empty()
+            return Fingerprint(key=op(self.key, other.key))
+        return NotImplemented
+
+    return _fingerprint_binop
+
+
 class Fingerprint(Model):
     """Fingerprint represents a unique identity as an int64 value.
 
@@ -28,7 +44,7 @@ class Fingerprint(Model):
     key: Optional[int64]
 
     def combine(self, *others: "Fingerprint") -> "Fingerprint":
-        return reduce(xor, others, self)
+        return reduce(operator.xor, others, self)
 
     @classmethod
     def empty(cls) -> "Fingerprint":
@@ -68,18 +84,15 @@ def is_empty(self) -> bool:
     def is_identity(self) -> bool:
         return self.key == 0
 
-    def __xor__(self, other: "Fingerprint") -> "Fingerprint":
-        if self.key is None:
-            return Fingerprint.empty()
-        if isinstance(other, Fingerprint):
-            if other.key is None:
-                return Fingerprint.empty()
-            return Fingerprint(key=self.key ^ other.key)
-        return NotImplemented
+    __and__ = _gen_fingerprint_binop(operator.__and__)
+    __lshift__ = _gen_fingerprint_binop(operator.__lshift__)
+    __or__ = _gen_fingerprint_binop(operator.__or__)
+    __rshift__ = _gen_fingerprint_binop(operator.__rshift__)
+    __xor__ = _gen_fingerprint_binop(operator.__xor__)
 
     def __eq__(self, other: object) -> bool:
         if isinstance(other, int):
-            return self.key == other
+            other = Fingerprint.from_int(other)
         if isinstance(other, Fingerprint):
             return self.key == other.key
         return NotImplemented
diff --git a/src/arti/formats/__init__.py b/src/arti/formats/__init__.py
index 1e3cb304..44c9d5ad 100644
--- a/src/arti/formats/__init__.py
+++ b/src/arti/formats/__init__.py
@@ -14,6 +14,7 @@ class Format(Model):
     """
 
     _abstract_ = True
+    extension: str = ""
     type_system: ClassVar[TypeSystem]
 
     def supports(self, type_: Type) -> None:
diff --git a/src/arti/formats/json.py b/src/arti/formats/json.py
new file mode 100644
index 00000000..5dea6660
--- /dev/null
+++ b/src/arti/formats/json.py
@@ -0,0 +1,9 @@
+from arti.formats import Format
+from arti.types.python import python_type_system
+
+
+class JSON(Format):
+    extension: str = "json"
+    # Perhaps we narrow down a json_type_system with the subset of supported types + a way to hook
+    # into json.JSON{De,En}coders.
+    type_system = python_type_system
diff --git a/src/arti/formats/pickle.py b/src/arti/formats/pickle.py
index 4a45eb8a..e82ec6b3 100644
--- a/src/arti/formats/pickle.py
+++ b/src/arti/formats/pickle.py
@@ -3,4 +3,5 @@
 
 
 class Pickle(Format):
+    extension = "pickle"
     type_system = python_type_system
diff --git a/src/arti/graphs/__init__.py b/src/arti/graphs/__init__.py
index 0e688d55..c758fa75 100644
--- a/src/arti/graphs/__init__.py
+++ b/src/arti/graphs/__init__.py
@@ -1,38 +1,133 @@
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
 
+from collections import defaultdict
+from collections.abc import Callable, Sequence
+from functools import cached_property, wraps
+from graphlib import TopologicalSorter
 from types import TracebackType
-from typing import Optional
+from typing import TYPE_CHECKING, Any, Literal, Optional, TypeVar, Union, cast
 
+from pydantic import Field, PrivateAttr
+
+import arti
+from arti import io
 from arti.artifacts import Artifact
 from arti.backends import Backend
-from arti.internal.utils import TypedBox
+from arti.backends.memory import MemoryBackend
+from arti.fingerprints import Fingerprint
+from arti.internal.models import Model
+from arti.internal.utils import TypedBox, frozendict
+from arti.partitions import CompositeKey
+from arti.producers import Producer
+from arti.storage import StoragePartition
+from arti.views import View
+
+if TYPE_CHECKING:
+    from arti.executors import Executor
+else:
+    from arti.internal.patches import patch_TopologicalSorter_class_getitem
+
+    patch_TopologicalSorter_class_getitem()
 
 # TODO: Add GraphMetadata model
 
 
-ArtifactBox = TypedBox[str, Artifact]
+SEALED: Literal[True] = True
+OPEN: Literal[False] = False
+BOX_KWARGS = {
+    status: {
+        "default_box": status is OPEN,
+        "frozen_box": status is SEALED,
+    }
+    for status in (OPEN, SEALED)
+}
+
+_Return = TypeVar("_Return")
+
+
+def requires_sealed(fn: Callable[..., _Return]) -> Callable[..., _Return]:
+    @wraps(fn)
+    def check_if_sealed(self: "Graph", *args: Any, **kwargs: Any) -> _Return:
+        if self._status is not SEALED:
+            raise ValueError(f"{fn.__name__} cannot be used while the Graph is still being defined")
+        return fn(self, *args, **kwargs)
+
+    return check_if_sealed
+
 
+class ArtifactBox(TypedBox[str, Artifact]):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        object.__setattr__(self, "_path", ())
+        super().__init__(*args, **kwargs)
 
-# TODO: When building out run logic, resolve and statically set all available fingerprints.
-# Specifically, we should pin the Producer.fingerprint, which may by dynamic (eg: version is a
-# Timestamp). Unbuilt Artifact (partitions) won't be fully resolved yet.
-class Graph:
+    def _TypedBox__cast_value(self, item: str, artifact: Any) -> Artifact:
+        artifact = super()._TypedBox__cast_value(item, artifact)  # type: ignore
+        if (graph := arti.context.graph) is not None:
+            artifact = artifact.copy(
+                update={
+                    "storage": (
+                        artifact.storage.resolve_graph_name(graph.name)
+                        .resolve_names(self._path + (item,))
+                        .resolve_path_tags(graph.path_tags)
+                    )
+                }
+            )
+        # Require an {input_fingerprint} template in the Storage if this Artifact is being generated
+        # by a Producer. Otherwise, strip the {input_fingerprint} template (if set) for "raw"
+        # Artifacts.
+        #
+        # We can't validate this at Artifact instantiation because the Producer is tracked later (by
+        # copying the instance and setting the `producer_output` attribute). We won't know the
+        # "final" instance until assignment here to the Graph.
+        if artifact.producer_output is None:
+            artifact = artifact.copy(
+                update={"storage": artifact.storage.resolve_input_fingerprint(Fingerprint.empty())}
+            )
+        elif not artifact.storage.includes_input_fingerprint_template:
+            raise ValueError(
+                "Produced Artifacts must have a '{input_fingerprint}' template in their Storage"
+            )
+        return cast(Artifact, artifact)
+
+    def _Box__convert_and_store(self, item: str, value: Artifact) -> None:
+        if isinstance(value, dict):
+            super()._Box__convert_and_store(item, value)  # pylint: disable=no-member
+            # TODO: Test if this works with `Box({"some": {"nested": "thing"}})`.
+            # Guessing not, may need to put in an empty dict/box first, set the path,
+            # and then update it.
+            object.__setattr__(self[item], "_path", self._path + (item,))
+        else:
+            super()._Box__convert_and_store(item, value)  # pylint: disable=no-member
+
+    def _Box__get_default(self, item: str, attr: bool = False) -> Any:
+        value = super()._Box__get_default(item, attr=attr)  # type: ignore
+        object.__setattr__(value, "_path", self._path + (item,))
+        return value
+
+
+Node = Union[Artifact, Producer]
+NodeDependencies = frozendict[Node, frozenset[Node]]
+
+
+class Graph(Model):
     """Graph stores a web of Artifacts connected by Producers."""
 
-    name: str
+    _fingerprint_excludes_ = frozenset(["backend"])
 
-    artifacts: ArtifactBox
+    name: str
+    backend: Backend = Field(default_factory=MemoryBackend)
+    path_tags: frozendict[str, str] = frozendict()
 
-    def __init__(self, name: str, *, backend: Optional[Backend] = None) -> None:
-        self.name = name
-        self.backend = backend or Backend()  # TODO: Fill in some default in-memory backend
-        self.artifacts = ArtifactBox()
-        self._sealed: bool = False
-        # Seal the class and convert the Boxes
-        self._toggle(sealed=True)
+    # Graph starts off sealed, but is opened within a `with Graph(...)` context
+    _status: Optional[bool] = PrivateAttr(None)
+    _artifacts: ArtifactBox = PrivateAttr(default_factory=lambda: ArtifactBox(**BOX_KWARGS[SEALED]))
+    _artifact_to_key: frozendict[Artifact, str] = PrivateAttr(frozendict())
 
     def __enter__(self) -> "Graph":
-        self._toggle(sealed=False)
+        if arti.context.graph is not None:
+            raise ValueError(f"Another graph is being defined: {arti.context.graph}")
+        arti.context.graph = self
+        self._toggle(OPEN)
         return self
 
     def __exit__(
@@ -41,12 +136,176 @@ def __exit__(
         exc_value: Optional[BaseException],
         exc_traceback: Optional[TracebackType],
     ) -> None:
-        self._toggle(sealed=True)
+        arti.context.graph = None
+        self._toggle(SEALED)
+        # Confirm the dependencies are acyclic
+        TopologicalSorter(self.dependencies).prepare()
 
-    def _toggle(self, sealed: bool) -> None:
-        self._sealed = sealed
-        box_kwargs = {
-            "default_box": not sealed,
-            "frozen_box": sealed,  # Unfreeze the box while building the graph
+    def _toggle(self, status: bool) -> None:
+        self._status = status
+        self._artifacts = ArtifactBox(self.artifacts, **BOX_KWARGS[status])
+        self._artifact_to_key = frozendict(
+            {artifact: key for key, artifact in self.artifacts.walk()}
+        )
+
+    @property
+    def artifacts(self) -> ArtifactBox:
+        return self._artifacts
+
+    @property
+    def artifact_to_key(self) -> frozendict[Artifact, str]:
+        return self._artifact_to_key
+
+    @requires_sealed
+    def build(self, executor: "Optional[Executor]" = None) -> None:
+        if executor is None:
+            from arti.executors.local import LocalExecutor
+
+            executor = LocalExecutor()
+        # TODO: Resolve and statically set all available fingerprints. Specifically, we
+        # should pin the Producer.fingerprint, which may by dynamic (eg: version is a
+        # Timestamp). Unbuilt Artifact (partitions) won't be fully resolved yet.
+        return executor.build(self)
+
+    @requires_sealed
+    def compute_id(self) -> Fingerprint:
+        """Identify a "unique" ID for this Graph at this point in time.
+
+        The ID aims to encode the structure of the Graph plus a _snapshot_ of the raw Artifact data
+        (partition kinds and contents). Any change that would affect data should prompt an ID
+        change, however changes to this ID don't directly cause data to be reproduced.
+
+        NOTE: There is currently a gap (and thus race condition) between when the Graph ID is
+        computed and when we read raw Artifacts data during Producer builds.
+        """
+        id = self.fingerprint
+        for node, _ in self.dependencies.items():
+            id = id.combine(node.fingerprint)
+            if isinstance(node, Artifact):
+                key = self.artifact_to_key[node]
+                id = id.combine(Fingerprint.from_string(key))
+                # Include fingerprints (including content_fingerprint!) for all raw Artifact
+                # partitions, triggering a graph ID change if these artifacts change out-of-band.
+                #
+                # TODO: Should we *also* inspect Producer.inputs for Artifacts _not_ inside this
+                # Graph and inspect their contents too? I guess we'll have to handle things a bit
+                # differently depending on if the external Artifacts are Produced (in an upstream
+                # Graph) or not.
+                if node.producer_output is None:
+                    partitions = [
+                        partition.with_content_fingerprint()
+                        for partition in node.discover_storage_partitions()
+                    ]
+                    if not partitions:
+                        raise ValueError(f"No data (partitions) found for `{key}`!")
+                    for partition in partitions:
+                        id = id.combine(partition.fingerprint)
+        if id.is_empty or id.is_identity:  # pragma: no cover
+            # NOTE: This shouldn't happen unless the logic above is faulty.
+            raise ValueError("Fingerprint is empty!")
+        return id
+
+    @cached_property  # type: ignore # python/mypy#1362
+    @requires_sealed
+    def dependencies(self) -> NodeDependencies:
+        artifact_deps = {
+            artifact: (
+                frozenset({artifact.producer_output.producer})
+                if artifact.producer_output is not None
+                else frozenset()
+            )
+            for _, artifact in self.artifacts.walk()
         }
-        self.artifacts = ArtifactBox(self.artifacts, **box_kwargs)
+        producer_deps = {
+            # NOTE: multi-output Producers will appear multiple times (but be deduped)
+            producer_output.producer: frozenset(producer_output.producer.inputs.values())
+            for artifact in artifact_deps
+            if (producer_output := artifact.producer_output) is not None
+        }
+        return NodeDependencies(artifact_deps | producer_deps)  # type: ignore
+
+    @cached_property  # type: ignore # python/mypy#1362
+    @requires_sealed
+    def producers(self) -> frozenset[Producer]:
+        return frozenset(self.producer_outputs)
+
+    @cached_property  # type: ignore # python/mypy#1362
+    @requires_sealed
+    def producer_outputs(self) -> frozendict[Producer, tuple[Artifact, ...]]:
+        d = defaultdict[Producer, dict[int, Artifact]](dict)
+        for _, artifact in self.artifacts.walk():
+            if artifact.producer_output is None:
+                continue
+            output = artifact.producer_output
+            d[output.producer][output.position] = artifact
+        return frozendict(
+            (producer, tuple(artifacts_by_position[i] for i in sorted(artifacts_by_position)))
+            for producer, artifacts_by_position in d.items()
+        )
+
+    # TODO: io.read/write probably need a bit of sanity checking (probably somewhere
+    # else), eg: type ~= view. Doing validation on the data, etc. Should some of this
+    # live on the View?
+
+    def read(
+        self,
+        artifact: Artifact,
+        *,
+        annotation: Optional[Any] = None,
+        graph_id: Optional[Fingerprint] = None,
+        storage_partitions: Optional[Sequence[StoragePartition]] = None,
+        view: Optional[View] = None,
+    ) -> Any:
+        key = self.artifact_to_key[artifact]
+        if annotation is None and view is None:
+            raise ValueError("Either `annotation` or `view` must be passed")
+        elif annotation is not None and view is not None:
+            raise ValueError("Only one of `annotation` or `view` may be passed")
+        elif annotation is not None:
+            view = View.get_class_for(annotation)()
+        if storage_partitions is None:
+            with self.backend.connect() as backend:
+                storage_partitions = backend.read_graph_partitions(
+                    graph_id or self.compute_id(), key
+                )
+        return io.read(
+            type=artifact.type,
+            format=artifact.format,
+            storage_partitions=storage_partitions,
+            view=view,
+        )
+
+    def write(
+        self,
+        data: Any,
+        *,
+        artifact: Artifact,
+        graph_id: Optional[Fingerprint] = None,
+        input_fingerprint: Fingerprint = Fingerprint.empty(),
+        keys: CompositeKey = CompositeKey(),
+        view: Optional[View] = None,
+    ) -> StoragePartition:
+        key = self.artifact_to_key[artifact]
+        if graph_id is not None and artifact.producer_output is None:
+            raise ValueError(
+                f"Writing to a raw Artifact (`{key}`) would cause a `graph_id` change."
+            )
+        if view is None:
+            view = View.get_class_for(type(data))()
+        storage_partition = artifact.storage.generate_partition(
+            input_fingerprint=input_fingerprint, keys=keys, with_content_fingerprint=False
+        )
+        io.write(
+            data,
+            type=artifact.type,
+            format=artifact.format,
+            storage_partition=storage_partition,
+            view=view,
+        )
+        storage_partition = storage_partition.with_content_fingerprint()
+        # TODO: Should we only do this in bulk? We might want the backends to
+        # transparently batch requests, but that's not so friendly with the transient
+        # ".connect".
+        with self.backend.connect() as backend:
+            backend.write_graph_partitions(graph_id or self.compute_id(), key, (storage_partition,))
+        return cast(StoragePartition, storage_partition)
diff --git a/src/arti/internal/__init__.py b/src/arti/internal/__init__.py
index 09dc39b8..73360e71 100644
--- a/src/arti/internal/__init__.py
+++ b/src/arti/internal/__init__.py
@@ -1,3 +1,27 @@
 import importlib.metadata
+from collections.abc import Iterator
+from contextlib import contextmanager
 
 version = importlib.metadata.version("arti")
+
+
+@contextmanager
+def wrap_exc(error_type: type[Exception], *, prefix: str) -> Iterator[None]:
+    """Wrap exceptions of `error_type` and add a message prefix.
+
+    `error_type` must be initializable with a single string message argument.
+
+    NOTE: When used inside a generator, any exceptions raised by the *caller of the generator* will **not** be wrapped.
+    """
+    try:
+        yield
+    except error_type as e:
+        msg = str(e)
+        if getattr(e, "wrapped", False) and e.__cause__ is not None:
+            src = e.__cause__  # Shorten exception chains to the root and last wrapped only
+        else:
+            msg = f" - {msg}"
+            src = e
+        error = error_type(f"{prefix}{msg}")
+        error.wrapped = True  # type: ignore
+        raise error from src
diff --git a/src/arti/internal/models.py b/src/arti/internal/models.py
index d8f5621b..eece8481 100644
--- a/src/arti/internal/models.py
+++ b/src/arti/internal/models.py
@@ -1,19 +1,34 @@
-from collections.abc import Mapping, Sequence
+from collections.abc import Generator, Mapping, Sequence
 from functools import cached_property
-from typing import Annotated, Any, ClassVar, Literal, Optional, get_args, get_origin
+from typing import (
+    TYPE_CHECKING,
+    Annotated,
+    Any,
+    ClassVar,
+    Literal,
+    Optional,
+    TypeVar,
+    get_args,
+    get_origin,
+)
 
 from pydantic import BaseModel, Extra, root_validator, validator
 from pydantic.fields import ModelField
+from pydantic.json import pydantic_encoder as pydantic_json_encoder
 
 from arti.internal.patches import patch_pydantic_ModelField__type_analysis
 from arti.internal.type_hints import is_union, lenient_issubclass
 from arti.internal.utils import class_name, classproperty, frozendict
 
+if TYPE_CHECKING:
+    from arti.fingerprints import Fingerprint
+    from arti.types import Type
+
 patch_pydantic_ModelField__type_analysis()
 
 
 def _check_types(value: Any, type_: type) -> Any:  # noqa: C901
-    mismatch_error = ValueError(f"Expected an instance of {type_}, got: {value}")
+    mismatch_error = ValueError(f"expected an instance of {type_}, got: {value}")
 
     if type_ is Any:
         return value
@@ -62,6 +77,9 @@ def _check_types(value: Any, type_: type) -> Any:  # noqa: C901
     return value
 
 
+_Model = TypeVar("_Model", bound="Model")
+
+
 class Model(BaseModel):
     # A model can be marked _abstract_ to prevent direct instantiation, such as when it is intended
     # as a base class for other models with arbitrary data. As the subclasses of an _abstract_ model
@@ -71,12 +89,23 @@ class Model(BaseModel):
     # 1: https://github.com/replicahq/artigraph/pull/60#discussion_r669089086
     _abstract_: ClassVar[bool] = True
     _class_key_: ClassVar[str] = class_name()
+    _fingerprint_excludes_: ClassVar[Optional[frozenset[str]]] = None
+    _fingerprint_includes_: ClassVar[Optional[frozenset[str]]] = None
 
     @classmethod
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)  # type: ignore # https://github.com/python/mypy/issues/4660
         # Default _abstract_ to False if not set explicitly on the class. __dict__ is read-only.
         setattr(cls, "_abstract_", cls.__dict__.get("_abstract_", False))
+        field_names = set(cls.__fields__)
+        if cls._fingerprint_excludes_ and (
+            unknown_excludes := cls._fingerprint_excludes_ - field_names
+        ):
+            raise ValueError(f"Unknown `_fingerprint_excludes_` field(s): {unknown_excludes}")
+        if cls._fingerprint_includes_ and (
+            unknown_includes := cls._fingerprint_includes_ - field_names
+        ):
+            raise ValueError(f"Unknown `_fingerprint_includes_` field(s): {unknown_includes}")
 
     @root_validator(pre=True)
     @classmethod
@@ -141,11 +170,55 @@ class Config:
         keep_untouched = (cached_property, classproperty)
         validate_assignment = True  # Unused with frozen, unless that is overridden in subclass.
 
+    def copy(self: _Model, *, validate: bool = True, **kwargs: Any) -> _Model:
+        copy = super().copy(**kwargs)
+        if validate:
+            return copy.validate(
+                dict(copy._iter(to_dict=False, by_alias=False, exclude_unset=True))
+            )
+        return copy
+
+    @staticmethod
+    def _fingerprint_json_encoder(obj: Any) -> Any:
+        from arti.fingerprints import Fingerprint
+
+        if isinstance(obj, Fingerprint):
+            return obj.key
+        if isinstance(obj, Model):
+            return obj.fingerprint
+        return pydantic_json_encoder(obj)
+
+    @property
+    def fingerprint(self) -> "Fingerprint":
+        from arti.fingerprints import Fingerprint
+
+        # `.json` cannot be used, even with a custom encoder, because it calls `.dict`, which
+        # converts the sub-models to dicts. Instead, we want to access `.fingerprint` (in the
+        # decoder).
+        data = dict(
+            sorted(  # Sort to ensure stability
+                self._iter(
+                    exclude=self._fingerprint_excludes_,
+                    include=self._fingerprint_includes_,
+                ),
+                key=lambda kv: kv[0],
+            )
+        )
+        json_repr = self.__config__.json_dumps(
+            data,
+            default=self._fingerprint_json_encoder,
+        )
+        return Fingerprint.from_string(f"{self._class_key_}:{json_repr}")
+
+    # Filter out non-fields from ._iter (and thus .dict, .json, etc), such as `@cached_property`
+    # after access (which just gets cached in .__dict__).
+    def _iter(self, *args: Any, **kwargs: Any) -> Generator[tuple[str, Any], None, None]:
+        for key, value in super()._iter(*args, **kwargs):
+            if key in self.__fields__:
+                yield key, value
+
     @classmethod
     def _pydantic_type_system_post_field_conversion_hook_(
-        cls, type_: "arti.types.Type", *, name: str, required: bool
-    ) -> "arti.types.Type":
+        cls, type_: "Type", *, name: str, required: bool
+    ) -> "Type":
         return type_
-
-
-import arti.types  # noqa: E402 # # pylint: disable=wrong-import-position
diff --git a/src/arti/internal/patches.py b/src/arti/internal/patches.py
index 479b94ed..0ebc31e5 100644
--- a/src/arti/internal/patches.py
+++ b/src/arti/internal/patches.py
@@ -1,8 +1,23 @@
 from typing import no_type_check
 
 
+def patch_TopologicalSorter_class_getitem() -> None:
+    """Patch adding TopologicalSorter.__class_getitem__ to support subscription.
+
+    TopologicalSorter is considered Generic in typeshed (hence mypy expects a type arg),
+    but is not at runtime.
+
+    Pending: https://github.com/python/cpython/pull/28714
+    """
+    from graphlib import TopologicalSorter
+    from types import GenericAlias
+
+    if not hasattr(TopologicalSorter, "__class_getitem__"):  # pragma: no cover
+        TopologicalSorter.__class_getitem__ = classmethod(GenericAlias)  # type: ignore
+
+
 @no_type_check
-def patch_pydantic_ModelField__type_analysis():  # noqa # pragma: no cover
+def patch_pydantic_ModelField__type_analysis() -> None:  # noqa # pragma: no cover
     """Patch to work around a bug causing dict subclasses to be converted to stock dicts[1].
 
     A PR to fix has been submitted[2] but not yet released - this patch pulls from that.
diff --git a/src/arti/internal/type_hints.py b/src/arti/internal/type_hints.py
index 6669a544..24fcd05b 100644
--- a/src/arti/internal/type_hints.py
+++ b/src/arti/internal/type_hints.py
@@ -2,14 +2,33 @@
 import sys
 import types
 from collections.abc import Callable
-from typing import Any, Union, cast, get_args, get_origin, get_type_hints, no_type_check
+from typing import _AnnotatedAlias  # type: ignore
+from typing import (
+    TYPE_CHECKING,
+    Annotated,
+    Any,
+    Union,
+    cast,
+    get_args,
+    get_origin,
+    get_type_hints,
+    no_type_check,
+)
 
 NoneType = cast(type, type(None))  # mypy otherwise treats type(None) as an object
 
 
 def _check_issubclass(klass: Any, check_type: type) -> bool:
-    klass_origin, check_type_origin = get_origin(klass), get_origin(check_type)
-    klass_args, check_type_args = get_args(klass), get_args(check_type)
+    # If a hint is Annotated, we want to unwrap the underlying type and discard the rest of the
+    # annotations.
+    klass_origin, klass_args = get_origin(klass), get_args(klass)
+    if klass_origin is Annotated:
+        klass = klass_args[0]
+        klass_origin, klass_args = get_origin(klass), get_args(klass)
+    check_type_origin, check_type_args = get_origin(check_type), get_args(check_type)
+    if check_type_origin is Annotated:
+        check_type = check_type_args[0]
+        check_type_origin, check_type_args = get_origin(check_type), get_args(check_type)
     # eg: issubclass(tuple, tuple)
     if klass_origin is None and check_type_origin is None:
         return issubclass(klass, check_type)
@@ -30,7 +49,8 @@ def _check_issubclass(klass: Any, check_type: type) -> bool:
             if not (
                 len(klass_args) == len(check_type_args)
                 and all(
-                    lenient_issubclass(klass_arg, check_type_arg)
+                    # check subclass OR things like "..."
+                    lenient_issubclass(klass_arg, check_type_arg) or klass_arg is check_type_arg
                     for (klass_arg, check_type_arg) in zip(klass_args, check_type_args)
                 )
             ):
@@ -50,7 +70,7 @@ def get_class_type_vars(klass: type) -> tuple[type, ...]:
 
 
 def lenient_issubclass(klass: Any, class_or_tuple: Union[type, tuple[type, ...]]) -> bool:
-    if not isinstance(klass, type):
+    if not (isinstance(klass, type) or is_Annotated(klass)):
         return False
     if isinstance(class_or_tuple, tuple):
         return any(lenient_issubclass(klass, subtype) for subtype in class_or_tuple)
@@ -76,7 +96,7 @@ def tidy_signature(
     force_tuple_return: bool = False,
     remove_owner: bool = False,
 ) -> inspect.Signature:
-    type_hints = get_type_hints(fn)
+    type_hints = get_type_hints(fn, include_extras=True)
     sig = sig.replace(return_annotation=type_hints.get("return", sig.return_annotation))
     return sig.replace(
         parameters=[
@@ -124,8 +144,19 @@ def signature(
     def is_union(type_: Any) -> bool:
         return type_ is Union
 
+    def is_typeddict(type_: Any) -> bool:
+        # mypy doesn't know of typing._TypedDictMeta, but `type: ignore` would be "unused" (and error)
+        # on other python versions.
+        if TYPE_CHECKING:
+            from typing import _TypedDict as _TypedDictMeta
+        else:
+            from typing import _TypedDictMeta
+
+        return isinstance(type_, _TypedDictMeta)
+
 
 else:  # pragma: no cover
+    from typing import is_typeddict as is_typeddict  # noqa: F401
 
     # mypy doesn't know of types.UnionType yet, but `type: ignore` would be "unused"
     # (and error) on other python versions.
@@ -135,6 +166,10 @@ def is_union(type_: Any) -> bool:
         return type_ is Union or type_ is types.UnionType  # noqa: E721
 
 
+def is_Annotated(type_: Any) -> bool:
+    return isinstance(type_, _AnnotatedAlias)
+
+
 def is_optional_hint(type_: Any) -> bool:
     # Optional[x] is represented as Union[x, NoneType]
     return is_union(get_origin(type_)) and NoneType in get_args(type_)
diff --git a/src/arti/internal/utils.py b/src/arti/internal/utils.py
index e0e6dace..0d77a2e9 100644
--- a/src/arti/internal/utils.py
+++ b/src/arti/internal/utils.py
@@ -1,9 +1,13 @@
+import functools
+import importlib
 import inspect
 import os.path
-from collections.abc import Callable, Generator, MutableMapping
+import pkgutil
+import threading
+from collections.abc import Callable, Generator, Iterator, MutableMapping
 from contextlib import contextmanager
 from tempfile import TemporaryDirectory
-from types import GenericAlias
+from types import GenericAlias, ModuleType
 from typing import IO, Any, ClassVar, Generic, Optional, TypeVar, Union, cast, overload
 
 from box import Box
@@ -11,6 +15,7 @@
 from multimethod import multidispatch
 
 from arti.internal.type_hints import lenient_issubclass, tidy_signature
+from arti.internal.vendored.setuptools import find_namespace_packages
 
 _K = TypeVar("_K")
 _V = TypeVar("_V")
@@ -48,18 +53,35 @@ def __get__(self, obj: Any, type_: Any) -> PropReturn:
 
 # This may be less useful once mypy supports ParamSpecs - after that, we might be able to define
 # multidispatch with a ParamSpec and have mypy check the handlers' arguments are covariant.
-class dispatch(multidispatch[RETURN]):
+class Dispatch(multidispatch[RETURN]):
     """Multiple dispatch for a set of functions based on parameter type.
 
     Usage is similar to `@functools.singledispatch`. The original definition defines the "spec" that
     subsequent handlers must follow, namely the name and (base)class of parameters.
     """
 
-    def __init__(self, func: Callable[..., RETURN]) -> None:
+    # multimethod/multidispatch __new__/__init__ work a bit specially - __new__ doesn't
+    # call __init__, but instead
+
+    # Modified from multidispatch to allow our extra kwargs
+    def __new__(
+        cls, func: Callable[..., RETURN], once_before: Callable[[], None]
+    ) -> "Dispatch[RETURN]":
+        return functools.update_wrapper(dict.__new__(cls), func)
+
+    def __init__(self, func: Callable[..., RETURN], once_before: Callable[[], None]) -> None:
         super().__init__(func)
         self.clean_signature = tidy_signature(func, self.signature)
         if self.clean_signature.return_annotation not in (Any, type(None)):
             raise NotImplementedError("Return type checking is not implemented yet!")
+        self.once_before = once_before
+        self.once_before_finished = False
+
+    def __call__(self, *args: Any, **kwargs: Any) -> RETURN:
+        if not self.once_before_finished:
+            self.once_before()
+            self.once_before_finished = True
+        return super().__call__(*args, **kwargs)
 
     @overload
     def register(self, __func: REGISTERED) -> REGISTERED:
@@ -93,6 +115,15 @@ def register(self, *args: Any) -> Callable[[REGISTERED], REGISTERED]:
         return super().register(*args)  # type: ignore
 
 
+def dispatch(
+    once_before: Callable[[], None] = lambda: None
+) -> Callable[[Callable[..., RETURN]], Dispatch[RETURN]]:
+    def decorate(func: Callable[..., RETURN]) -> Dispatch[RETURN]:
+        return Dispatch(func, once_before)
+
+    return decorate
+
+
 # frozendict is useful for models to preserve hashability (eg: key in a dict). Unfortunately, there
 # is 1 small pydantic issue[1] and an upstream issue with types.GenericAlias + deepcopy[2,3]. We can
 # work around the blocking deepcopy issue by replacing types.GenericAlias with typing.Generic (which
@@ -108,6 +139,43 @@ class frozendict(Generic[_K, _V], _frozendict[_K, _V]):
     pass
 
 
+def import_submodules(
+    path: list[str],  # module.__path__ is a list[str]
+    name: str,
+    *,
+    lock: threading.Lock = threading.Lock(),
+) -> dict[str, ModuleType]:
+    """Recursively import submodules.
+
+    This can be useful with registry patterns to automatically discover and import submodules
+    defining additional implementations.
+
+    `path` and `name` are usually provided from an existing module's `__path__` and `__name__`.
+
+    This function is thread-safe and supports namespace modules.
+
+    NOTE: This inherently triggers eager imports, which has performance impacts and may cause import
+    cycles. To reduce these issues, avoid calling during module definition.
+    """
+    # pkgutil.iter_modules is not recursive and pkgutil.walk_packages does not handle namespace
+    # packages... however we can leverage setuptools.find_namespace_packages, which was built for
+    # exactly this.
+    path_names = {p: name for p in path}
+    path_names.update(
+        {
+            os.sep.join([path, *name.split(".")]): f"{root_name}.{name}"
+            for path, root_name in path_names.items()
+            for name in find_namespace_packages(path)
+        }
+    )
+    with lock:
+        return {
+            name: importlib.import_module(name)
+            for path, name in path_names.items()
+            for _, name, _ in pkgutil.iter_modules([path], prefix=f"{name}.")
+        }
+
+
 _int_sub = TypeVar("_int_sub", bound="_int")
 
 
@@ -115,6 +183,9 @@ class _int(int):
     def __repr__(self) -> str:
         return f"{qname(self)}({int(self)})"
 
+    def __str__(self) -> str:
+        return str(int(self))
+
     # Stock magics.
     #
     # Using "self: TypeVar" so mypy will detect the returned subclass (rather than _int).
@@ -273,7 +344,10 @@ def qname(val: Union[object, type]) -> str:
     return type(val).__qualname__
 
 
-class TypedBox(Box, MutableMapping[_K, Union[_V, MutableMapping[_K, _V]]]):
+_K_str = TypeVar("_K_str")
+
+
+class TypedBox(Box, MutableMapping[_K_str, Union[_V, MutableMapping[_K_str, _V]]]):
     """TypedBox holds a collection of typed values.
 
     Subclasses must set the __target_type__ to a base class for the contained values.
@@ -282,7 +356,7 @@ class TypedBox(Box, MutableMapping[_K, Union[_V, MutableMapping[_K, _V]]]):
     __target_type__: ClassVar[type[_V]]
 
     @classmethod
-    def __class_getitem__(cls, item: tuple[type[_K], type[_V]]) -> GenericAlias:
+    def __class_getitem__(cls, item: tuple[type[_K_str], type[_V]]) -> GenericAlias:
         if not isinstance(item, tuple) or len(item) != 2:
             raise TypeError(f"{cls.__name__} expects a key and value type")
         key_type, value_type = item
@@ -307,7 +381,7 @@ def __setattr__(self, key: str, value: Any) -> None:
             return object.__setattr__(self, key, value)
         super().__setattr__(key, value)
 
-    def __cast_value(self, value: Any) -> _V:
+    def __cast_value(self, item: str, value: Any) -> _V:
         if isinstance(value, self.__target_type__):
             return value
         tgt_name = self.__target_type__.__name__
@@ -327,4 +401,12 @@ def _Box__convert_and_store(self, item: str, value: _V) -> None:
         elif item in self:
             raise ValueError(f"{item} is already set!")
         else:
-            super()._Box__convert_and_store(item, self.__cast_value(value))
+            super()._Box__convert_and_store(item, self.__cast_value(item, value))
+
+    def walk(self, root: tuple[_K_str, ...] = ()) -> Iterator[tuple[_K_str, _V]]:
+        for k, v in self.items():
+            subroot = root + (k,)
+            if isinstance(v, TypedBox):
+                yield from v.walk(root=subroot)
+            else:
+                yield ".".join(subroot), v  # type: ignore
diff --git a/src/arti/internal/vendored/setuptools.py b/src/arti/internal/vendored/setuptools.py
new file mode 100644
index 00000000..dbc494d5
--- /dev/null
+++ b/src/arti/internal/vendored/setuptools.py
@@ -0,0 +1,124 @@
+""" Vendored copy of find_namespace_packages from setuptools[1].
+
+Adding setuptools as a package dependency can be a bit messy, sometimes triggering an
+uninstall/reinstall on the _user_ side or other issues.
+
+Type hints have been added to pass `mypy` strict mode.
+
+1: https://github.com/pypa/setuptools/tree/c65e3380b0a18c92a0fc2d2b770b17cfaaec054b
+"""
+import os
+from collections.abc import Callable, Generator, Iterable
+from fnmatch import fnmatchcase
+
+
+# From: https://github.com/pypa/setuptools/blob/c65e3380b0a18c92a0fc2d2b770b17cfaaec054b/setuptools/_distutils/util.py#L166
+def convert_path(pathname: str) -> str:  # pragma: no cover
+    """Return 'pathname' as a name that will work on the native filesystem,
+    i.e. split it on '/' and put it back together again using the current
+    directory separator.  Needed because filenames in the setup script are
+    always supplied in Unix style, and have to be converted to the local
+    convention before we can actually use them in the filesystem.  Raises
+    ValueError on non-Unix-ish systems if 'pathname' either starts or
+    ends with a slash.
+    """
+    if os.sep == "/":
+        return pathname
+    if not pathname:
+        return pathname
+    if pathname[0] == "/":
+        raise ValueError("path '%s' cannot be absolute" % pathname)
+    if pathname[-1] == "/":
+        raise ValueError("path '%s' cannot end with '/'" % pathname)
+
+    paths = pathname.split("/")
+    while "." in paths:
+        paths.remove(".")
+    if not paths:
+        return os.curdir
+    return os.path.join(*paths)
+
+
+# From: https://github.com/pypa/setuptools/blob/c65e3380b0a18c92a0fc2d2b770b17cfaaec054b/setuptools/__init__.py#L39-L119
+class PackageFinder:
+    """
+    Generate a list of all Python packages found within a directory
+    """
+
+    @classmethod
+    def find(
+        cls, where: str = ".", exclude: Iterable[str] = (), include: Iterable[str] = ("*",)
+    ) -> list[str]:
+        """Return a list all Python packages found within directory 'where'
+        'where' is the root directory which will be searched for packages.  It
+        should be supplied as a "cross-platform" (i.e. URL-style) path; it will
+        be converted to the appropriate local path syntax.
+        'exclude' is a sequence of package names to exclude; '*' can be used
+        as a wildcard in the names, such that 'foo.*' will exclude all
+        subpackages of 'foo' (but not 'foo' itself).
+        'include' is a sequence of package names to include.  If it's
+        specified, only the named packages will be included.  If it's not
+        specified, all found packages will be included.  'include' can contain
+        shell style wildcard patterns just like 'exclude'.
+        """
+
+        return list(
+            cls._find_packages_iter(
+                convert_path(where),
+                cls._build_filter("ez_setup", "*__pycache__", *exclude),
+                cls._build_filter(*include),
+            )
+        )
+
+    @classmethod
+    def _find_packages_iter(
+        cls, where: str, exclude: Callable[[str], bool], include: Callable[[str], bool]
+    ) -> Generator[str, None, None]:
+        """
+        All the packages found in 'where' that pass the 'include' filter, but
+        not the 'exclude' filter.
+        """
+        for root, dirs, files in os.walk(where, followlinks=True):
+            # Copy dirs to iterate over it, then empty dirs.
+            all_dirs = dirs[:]
+            dirs[:] = []
+
+            for dir in all_dirs:
+                full_path = os.path.join(root, dir)
+                rel_path = os.path.relpath(full_path, where)
+                package = rel_path.replace(os.path.sep, ".")
+
+                # Skip directory trees that are not valid packages
+                if "." in dir or not cls._looks_like_package(full_path):
+                    continue
+
+                # Should this package be included?
+                if include(package) and not exclude(package):
+                    yield package
+
+                # Keep searching subdirectories, as there may be more packages
+                # down there, even if the parent was excluded.
+                dirs.append(dir)
+
+    @staticmethod
+    def _looks_like_package(path: str) -> bool:
+        """Does a directory look like a package?"""
+        return os.path.isfile(os.path.join(path, "__init__.py"))
+
+    @staticmethod
+    def _build_filter(*patterns: str) -> Callable[[str], bool]:
+        """
+        Given a list of patterns, return a callable that will be true only if
+        the input matches at least one of the patterns.
+        """
+        return lambda name: any(fnmatchcase(name, pat=pat) for pat in patterns)
+
+
+class PEP420PackageFinder(PackageFinder):
+    @staticmethod
+    def _looks_like_package(path: str) -> bool:
+        return True
+
+
+find_packages = PackageFinder.find
+find_namespace_packages = PEP420PackageFinder.find
diff --git a/src/arti/io/__init__.py b/src/arti/io/__init__.py
index 0b6462b0..d4102b15 100644
--- a/src/arti/io/__init__.py
+++ b/src/arti/io/__init__.py
@@ -1,24 +1,38 @@
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
 
-from typing import Any
+from collections.abc import Callable, Sequence
+from functools import partial
+from typing import Any, cast
 
 from arti.formats import Format
-from arti.internal.utils import dispatch
+from arti.internal.utils import dispatch, import_submodules
 from arti.storage import StoragePartition
+from arti.types import Type
 from arti.views import View
 
 # TODO write/read partitioned data, column subset
 
+# TODO: Auto import io/ submodules (like View) - perhaps add decorator on read/write to
+# lazily trigger the import on first use.
 
-@dispatch
-def read(*, format: Format, storage_partitions: list[StoragePartition], view: View) -> Any:
+# import_submodules(__path__, __name__)
+
+_import_submodules = cast(Callable[[], None], partial(import_submodules, __path__, __name__))
+
+
+@dispatch(once_before=_import_submodules)
+def read(
+    type: Type, format: Format, storage_partitions: Sequence[StoragePartition], view: View
+) -> Any:
     raise NotImplementedError(
         f"Read into {view} view from {format} format in {storage_partitions} storage not implemented."
     )
 
 
-@dispatch
-def write(data: Any, *, format: Format, storage_partition: StoragePartition, view: View) -> None:
+@dispatch(once_before=_import_submodules)
+def write(
+    data: Any, type: Type, format: Format, storage_partition: StoragePartition, view: View
+) -> None:
     raise NotImplementedError(
         f"Write from {view} view into {format} format in {storage_partition} storage not implemented."
     )
diff --git a/src/arti/io/json_localfile_python.py b/src/arti/io/json_localfile_python.py
new file mode 100644
index 00000000..0abf6fb6
--- /dev/null
+++ b/src/arti/io/json_localfile_python.py
@@ -0,0 +1,48 @@
+import json
+from collections.abc import Sequence
+from itertools import chain
+from pathlib import Path
+from typing import Any
+
+from arti.formats.json import JSON
+from arti.io import read, write
+from arti.storage.local import LocalFilePartition
+from arti.types import List, Type
+from arti.views.python import PythonBuiltin
+
+# TODO: Do I need to inject the partition keys into the returned data? Likely useful...
+# Maybe a View option?
+
+
+def _read_json_file(path: str) -> Any:
+    with open(path, "rb") as file:
+        return json.load(file)
+
+
+@read.register
+def _read_json_localfile_python(
+    type: Type,
+    format: JSON,
+    storage_partitions: Sequence[LocalFilePartition],
+    view: PythonBuiltin,
+) -> Any:
+    if isinstance(type, List) and type.is_partitioned:
+        return list(
+            chain.from_iterable(
+                _read_json_file(storage_partition.path) for storage_partition in storage_partitions
+            )
+        )
+    else:
+        if len(storage_partitions) != 1:
+            raise ValueError(f"Multiple partitions can only be read into a list, not {view}")
+        return _read_json_file(storage_partitions[0].path)
+
+
+@write.register
+def _write_json_localfile_python(
+    data: Any, type: Type, format: JSON, storage_partition: LocalFilePartition, view: PythonBuiltin
+) -> None:
+    path = Path(storage_partition.path)
+    path.parent.mkdir(exist_ok=True, parents=True)
+    with path.open("w") as file:
+        json.dump(data, file)
diff --git a/src/arti/io/pickle_localfile_python.py b/src/arti/io/pickle_localfile_python.py
new file mode 100644
index 00000000..189cba08
--- /dev/null
+++ b/src/arti/io/pickle_localfile_python.py
@@ -0,0 +1,53 @@
+import pickle
+from collections.abc import Sequence
+from itertools import chain
+from pathlib import Path
+from typing import Any
+
+from arti.formats.pickle import Pickle
+from arti.io import read, write
+from arti.storage.local import LocalFilePartition
+from arti.types import List, Type
+from arti.views.python import PythonBuiltin
+
+
+def _read_pickle_file(path: str) -> Any:
+    with open(path, "rb") as file:
+        return pickle.load(file)
+
+
+# TODO: Need to handle partitioned lists. Should I separate read and read_partitions?
+
+
+@read.register
+def _read_pickle_localfile_python(
+    type: Type,
+    format: Pickle,
+    storage_partitions: Sequence[LocalFilePartition],
+    view: PythonBuiltin,
+) -> Any:
+    if isinstance(type, List) and type.is_partitioned:
+        return list(
+            chain.from_iterable(
+                _read_pickle_file(storage_partition.path)
+                for storage_partition in storage_partitions
+            )
+        )
+    else:
+        if len(storage_partitions) != 1:
+            raise ValueError(f"Multiple partitions can only be read into a list, not {view}")
+        return _read_pickle_file(storage_partitions[0].path)
+
+
+@write.register
+def _write_pickle_localfile_python(
+    data: Any,
+    type: Type,
+    format: Pickle,
+    storage_partition: LocalFilePartition,
+    view: PythonBuiltin,
+) -> None:
+    path = Path(storage_partition.path)
+    path.parent.mkdir(exist_ok=True, parents=True)
+    with path.open("wb") as file:
+        pickle.dump(data, file)
diff --git a/src/arti/partitions/__init__.py b/src/arti/partitions/__init__.py
index 57862c21..91899a03 100644
--- a/src/arti/partitions/__init__.py
+++ b/src/arti/partitions/__init__.py
@@ -7,7 +7,7 @@
 
 from arti.internal.models import Model
 from arti.internal.utils import classproperty, frozendict, register
-from arti.types import Date, Int8, Int16, Int32, Int64, Null, Struct, Type
+from arti.types import Date, Int8, Int16, Int32, Int64, List, Null, Type
 
 
 class key_component(property):
@@ -16,8 +16,9 @@ class key_component(property):
 
 class PartitionKey(Model):
     _abstract_ = True
-    _by_type_: ClassVar[dict[type[Type], type["PartitionKey"]]] = {}
+    _by_type_: "ClassVar[dict[type[Type], type[PartitionKey]]]" = {}
 
+    default_key_components: ClassVar[frozendict[str, str]]
     matching_type: ClassVar[type[Type]]
 
     @classmethod
@@ -25,15 +26,20 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
         if cls._abstract_:
             return
-        if not hasattr(cls, "matching_type"):
-            raise TypeError(f"{cls.__name__} must set `matching_type`")
+        for attr in ("default_key_components", "matching_type"):
+            if not hasattr(cls, attr):
+                raise TypeError(f"{cls.__name__} must set `{attr}`")
+        if unknown := (set(cls.default_key_components) - cls.key_components):
+            raise TypeError(
+                f"Unknown key_components in {cls.__name__}.default_key_components: {unknown}"
+            )
         register(cls._by_type_, cls.matching_type, cls)
 
     @classproperty
     @classmethod
     def key_components(cls) -> frozenset[str]:
-        return frozenset(
-            {name for name in dir(cls) if isinstance(getattr_static(cls, name), key_component)}
+        return frozenset(cls.__fields__) | frozenset(
+            name for name in dir(cls) if isinstance(getattr_static(cls, name), key_component)
         )
 
     @classmethod
@@ -42,23 +48,27 @@ def from_key_components(cls, **key_components: str) -> "PartitionKey":
         raise NotImplementedError(f"Unable to parse '{cls.__name__}' from: {key_components}")
 
     @classmethod
-    def key_type_for(cls, type_: Type) -> type["PartitionKey"]:
+    def get_class_for(cls, type_: Type) -> type["PartitionKey"]:
         return cls._by_type_[type(type_)]
 
     @classmethod
-    def key_types_from(cls, type_: Type) -> frozendict[str, type["PartitionKey"]]:
-        if not isinstance(type_, Struct):
+    def types_from(cls, type_: Type) -> "CompositeKeyTypes":
+        if not isinstance(type_, List):
             return frozendict()
         return frozendict(
-            {name: cls.key_type_for(field) for name, field in type_.partition_fields.items()}
+            {name: cls.get_class_for(field) for name, field in type_.partition_fields.items()}
         )
 
 
 # CompositeKey is the set of named PartitionKeys that uniquely identify a single partition.
 CompositeKey = frozendict[str, PartitionKey]
+CompositeKeyTypes = frozendict[str, type[PartitionKey]]
+
+NotPartitioned = CompositeKey()
 
 
 class DateKey(PartitionKey):
+    default_key_components: ClassVar[frozendict[str, str]] = frozendict(Y="Y", m="m:02", d="d:02")
     matching_type = Date
 
     key: date
@@ -93,6 +103,7 @@ def from_key_components(cls, **key_components: str) -> PartitionKey:
 
 class _IntKey(PartitionKey):
     _abstract_ = True
+    default_key_components: ClassVar[frozendict[str, str]] = frozendict(key="key")
 
     key: int
 
@@ -127,6 +138,7 @@ class Int64Key(_IntKey):
 
 
 class NullKey(PartitionKey):
+    default_key_components: ClassVar[frozendict[str, str]] = frozendict(key="key")
     matching_type = Null
 
     key: None = None
diff --git a/src/arti/producers/__init__.py b/src/arti/producers/__init__.py
index 9631a080..80211c11 100644
--- a/src/arti/producers/__init__.py
+++ b/src/arti/producers/__init__.py
@@ -1,171 +1,253 @@
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
 
+from collections import defaultdict
 from collections.abc import Callable, Iterable, Iterator
-from inspect import Signature, getattr_static
-from typing import Any, ClassVar, Union, cast
-
-from pydantic.fields import ModelField
+from inspect import Parameter, Signature, getattr_static
+from typing import Annotated, Any, ClassVar, Optional, TypeVar, Union, cast, get_args, get_origin
 
 from arti.annotations import Annotation
-from arti.artifacts import Artifact
+from arti.artifacts import Artifact, BaseArtifact, Statistic
 from arti.fingerprints import Fingerprint
+from arti.internal import wrap_exc
 from arti.internal.models import Model
-from arti.internal.type_hints import signature
-from arti.internal.utils import ordinal
+from arti.internal.type_hints import NoneType, lenient_issubclass, signature
+from arti.internal.utils import frozendict, ordinal
+from arti.partitions import CompositeKey, CompositeKeyTypes, NotPartitioned
+from arti.storage import StoragePartitions
 from arti.versions import SemVer, Version
+from arti.views import View
 
 
 def _commas(vals: Iterable[Any]) -> str:
     return ", ".join([str(v) for v in vals])
 
 
+# TODO: Add @validator over all (build) fields checking for an io.{read,write} handler.
+
+
+ArtifactViewPair = tuple[type[Artifact], type[View]]
+BuildInputViews = frozendict[str, type[View]]
+MapInputMetadata = frozendict[str, type[Artifact]]
+# mypy doesn't yet support homogeneous tuple aliases: https://github.com/python/mypy/issues/9980
+OutputMetadata = tuple[ArtifactViewPair, ...]  # type: ignore
+
+PartitionDependencies = frozendict[CompositeKey, frozendict[str, StoragePartitions]]
+
+BuildSig = Callable[..., Any]
+MapSig = Callable[..., PartitionDependencies]
+
+_T = TypeVar("_T")
+
+
 class Producer(Model):
     """A Producer is a task that builds one or more Artifacts."""
 
     # User fields/methods
 
     annotations: tuple[Annotation, ...] = ()
-    version: ClassVar[Version] = SemVer(major=0, minor=0, patch=1)
+    version: Version = SemVer(major=0, minor=0, patch=1)
 
     # build and map must be @classmethods or @staticmethods.
-    build: ClassVar[Callable[..., Any]]
-    map: ClassVar[Callable[..., Any]]
+    build: ClassVar[BuildSig]
+    map: ClassVar[MapSig]
 
     # Internal fields/methods
 
     _abstract_: ClassVar[bool] = True
+    _fingerprint_excludes_ = frozenset(["annotations"])
+
     # NOTE: The following are set in __init_subclass__
-    _artifact_fields_: ClassVar[dict[str, ModelField]]
+    _input_artifact_types_: ClassVar[frozendict[str, type[Artifact]]]
     _build_sig_: ClassVar[Signature]
+    _build_input_views_: ClassVar[BuildInputViews]
+    _output_metadata_: ClassVar[OutputMetadata]
     _map_sig_: ClassVar[Signature]
+    _map_input_metadata_: ClassVar[MapInputMetadata]
 
     @classmethod
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
         if not cls._abstract_:
-            cls._artifact_fields_ = cls._validate_fields()
-            cls._build_sig_ = cls._validate_build_sig()
-            cls._map_sig_ = cls._validate_map_sig()
-            cls._validate_cohesive_params()
+            with wrap_exc(ValueError, prefix=cls.__name__):
+                cls._input_artifact_types_ = cls._validate_fields()
+                with wrap_exc(ValueError, prefix=".build"):
+                    (
+                        cls._build_sig_,
+                        cls._build_input_views_,
+                        cls._output_metadata_,
+                    ) = cls._validate_build_sig()
+                with wrap_exc(ValueError, prefix=".map"):
+                    cls._map_sig_, cls._map_input_metadata_ = cls._validate_map_sig()
+                cls._validate_no_unused_fields()
+
+    @classmethod
+    def _get_artifact_from_annotation(cls, annotation: Any) -> type[Artifact]:
+        hint_example = "eg: `Annotated[pd.DataFrame, MyArtifact]`"
+        origin, args = get_origin(annotation), get_args(annotation)
+        if origin is not Annotated:
+            raise ValueError(f"must be an Annotated hint ({hint_example})")
+        # We don't need the underlying type here - `_get_view_from_annotation` requires the entire
+        # (possibly Annotated) annotation in order to discover explicitly set Views.
+        _, *hints = args
+        artifacts = [hint for hint in hints if lenient_issubclass(hint, Artifact)]
+        if len(artifacts) == 0:
+            raise ValueError(f"Artifact is not set ({hint_example})")
+        if len(artifacts) > 1:
+            raise ValueError("multiple Artifacts set")
+        return cast(type[Artifact], artifacts[0])
 
     @classmethod
-    def _validate_fields(cls) -> dict[str, ModelField]:
+    def _get_view_from_annotation(cls, annotation: Any, artifact: type[Artifact]) -> type[View]:
+        view = View.get_class_for(annotation)
+        wrap_msg = f"{artifact.__name__}"
+        if artifact.is_partitioned:
+            wrap_msg = f"partitions of {artifact.__name__}"
+        with wrap_exc(ValueError, prefix=f" ({wrap_msg})"):
+            view.type_system.check_similarity(arti=artifact._type, python_type=annotation)
+        return view
+
+    @classmethod
+    def _validate_fields(cls) -> frozendict[str, type[Artifact]]:
         # NOTE: Aside from the base producer fields, all others should be Artifacts.
         #
         # Users can set additional class attributes, but they must be properly hinted as ClassVars.
         # These won't interact with the "framework" and can't be parameters to build/map.
         artifact_fields = {k: v for k, v in cls.__fields__.items() if k not in Producer.__fields__}
         for name, field in artifact_fields.items():
-            if not (field.default is None and field.default_factory is None and field.required):
-                raise ValueError(
-                    f"{cls.__name__}.{name} - field must not have a default nor be Optional."
-                )
-            if not issubclass(field.outer_type_, Artifact):
-                raise ValueError(
-                    f"{cls.__name__}.{name} - type hint must be an Artifact subclass, got: {field.outer_type_}"
-                )
-        return artifact_fields
+            with wrap_exc(ValueError, prefix=f".{name}"):
+                if not (field.default is None and field.default_factory is None and field.required):
+                    raise ValueError("field must not have a default nor be Optional.")
+                if not lenient_issubclass(field.outer_type_, Artifact):
+                    raise ValueError(
+                        f"type hint must be an Artifact subclass, got: {field.outer_type_}"
+                    )
+        return frozendict({name: field.outer_type_ for name, field in artifact_fields.items()})
 
     @classmethod
-    def _validate_parameters(cls, fn_name: str, sig: Signature) -> None:
+    def _validate_parameters(
+        cls, sig: Signature, *, validator: Callable[[str, Parameter, type[Artifact]], _T]
+    ) -> Iterator[_T]:
+        if undefined_params := set(sig.parameters) - set(cls._input_artifact_types_):
+            raise ValueError(
+                f"the following parameter(s) must be defined as a field: {undefined_params}"
+            )
         for name, param in sig.parameters.items():
-            if param.annotation is param.empty:
-                raise ValueError(
-                    f"{cls.__name__}.{fn_name} - `{name}` parameter must have a type hint."
-                )
-            if param.default is not param.empty:
-                raise ValueError(
-                    f"{cls.__name__}.{fn_name} - `{name}` parameter must not have a default."
-                )
-            if param.kind not in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY):
-                raise ValueError(
-                    "{cls.__name__}.{fn_name} - `{name}` parameter must be usable as a keyword argument."
-                )
+            with wrap_exc(ValueError, prefix=f" {name} param"):
+                if param.annotation is param.empty:
+                    raise ValueError("must have a type hint.")
+                if param.default is not param.empty:
+                    raise ValueError("must not have a default.")
+                if param.kind not in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY):
+                    raise ValueError("must be usable as a keyword argument.")
+                artifact = cls.__fields__[param.name].outer_type_
+                yield validator(name, param, artifact)
 
     @classmethod
-    def _validate_build_sig(cls) -> Signature:
+    def _validate_build_sig_return(cls, annotation: Any, *, i: int) -> ArtifactViewPair:
+        with wrap_exc(ValueError, prefix=f" {ordinal(i)} return"):
+            artifact = cls._get_artifact_from_annotation(annotation)
+            return artifact, cls._get_view_from_annotation(annotation, artifact)
+
+    @classmethod
+    def _validate_build_sig(cls) -> tuple[Signature, BuildInputViews, OutputMetadata]:
         """Validate the .build method"""
         if not hasattr(cls, "build"):
-            raise ValueError(f"{cls.__name__} - Producers must implement the `build` method.")
+            raise ValueError("must be implemented")
         if not isinstance(getattr_static(cls, "build"), (classmethod, staticmethod)):
-            raise ValueError(f"{cls.__name__}.build - must be a @classmethod or @staticmethod.")
+            raise ValueError("must be a @classmethod or @staticmethod")
         build_sig = signature(cls.build, force_tuple_return=True, remove_owner=True)
-        cls._validate_parameters("build", build_sig)
+        # Validate the parameters
+        build_input_metadata = BuildInputViews(
+            cls._validate_parameters(
+                build_sig,
+                validator=(
+                    lambda name, param, artifact: (
+                        name,
+                        cls._get_view_from_annotation(param.annotation, artifact),
+                    )
+                ),
+            )
+        )
         # Validate the return definition
         return_annotation = build_sig.return_annotation
         if return_annotation is build_sig.empty:
             # TODO: "side effect" Producers: https://github.com/replicahq/artigraph/issues/11
-            raise ValueError(
-                f"{cls.__name__}.build - A return value must be set with the output Artifact(s)."
-            )
-        for i, annotation in enumerate(return_annotation, 1):
-            if annotation is None or not issubclass(annotation, Artifact):
-                raise ValueError(
-                    f"{cls.__name__}.build - The {ordinal(i)} return value must be an Artifact, got: {annotation}"
-                )
-        return build_sig
+            raise ValueError("a return value must be set with the output Artifact(s).")
+        if return_annotation == (NoneType,):
+            raise ValueError("missing return signature")
+        output_metadata = OutputMetadata(
+            cls._validate_build_sig_return(annotation, i=i)
+            for i, annotation in enumerate(return_annotation, 1)
+        )
+        # Validate all output Artifacts have equivalent partitioning schemes.
+        #
+        # We currently require the partition key type *and* name to match, but in the
+        # future we might be able to extend the dependency metadata to support
+        # heterogeneous names if necessary.
+        artifacts_by_composite_key = defaultdict[CompositeKeyTypes, list[type[Artifact]]](list)
+        for (artifact, _) in output_metadata:
+            artifacts_by_composite_key[artifact.partition_key_types].append(artifact)
+        if len(artifacts_by_composite_key) != 1:
+            raise ValueError("all output Artifacts must have the same partitioning scheme")
+        # TODO: Save off output composite_key_types
+
+        return build_sig, build_input_metadata, output_metadata
 
     @classmethod
-    def _validate_map_sig(cls) -> Signature:
+    def _validate_map_sig(cls) -> tuple[Signature, MapInputMetadata]:
         """Validate partitioned Artifacts and the .map method"""
         if not hasattr(cls, "map"):
-            # TODO: if (input_is_partitioned, output_is_partitioned):
-            # - (input_is, output_is):
-            #       Error requiring override (even w/ equivalent partitioning, we can't know if the
-            #       columns actually align)
-            # - (input_is, output_is_not):
-            #       Default to all input partitions -> the one output, but may be overridden to
-            #       filter input partitions
-            # - (input_is_not, output_is):
-            #       Error requiring override (they must define the output partitions by inspecting
-            #       the input statistics or whatever)
-            # - (input_is_not, output_is_not):
-            #       Default 1<->1, any override must return 1<->1 too
-            def map() -> Any:
-                """Map dependencies between input and output Artifact partitions.
-
-                If there are multiple output Artifacts, they must have equivalent partitioning
-                schemes.
-                """
-                raise ValueError("Implement the default")
+            partitioned_outputs = [
+                artifact for (artifact, view) in cls._output_metadata_ if artifact.is_partitioned
+            ]
+            # TODO: Add runtime checking of `map` output (ie: output aligns w/ output
+            # artifacts and such).
+            if partitioned_outputs:
+                raise ValueError("must be implemented when the `build` outputs are partitioned")
+            else:
+
+                def map(**kwargs: StoragePartitions) -> PartitionDependencies:
+                    return PartitionDependencies(
+                        {NotPartitioned: {name: partitions for name, partitions in kwargs.items()}}
+                    )
 
+            # Narrow the map signature, which is validated below and used at graph build
+            # time (via cls._map_input_metadata_) to determine what arguments to pass to
+            # map.
+            map.__signature__ = Signature(  # type: ignore
+                [
+                    Parameter(name=name, annotation=StoragePartitions, kind=Parameter.KEYWORD_ONLY)
+                    for name, artifact in cls._input_artifact_types_.items()
+                    if name in cls._build_input_views_
+                ],
+                return_annotation=PartitionDependencies,
+            )
             cls.map = staticmethod(map)  # type: ignore
         if not isinstance(getattr_static(cls, "map"), (classmethod, staticmethod)):
-            raise ValueError(f"{cls.__name__}.map - must be a @classmethod or @staticmethod.")
+            raise ValueError("must be a @classmethod or @staticmethod")
         map_sig = signature(cls.map)
-        cls._validate_parameters("map", map_sig)
-        # TODO: Verify cls._build_sig_ output Artifacts (if multiple) have equivalent partitioning schemes
-        return map_sig  # TODO: Verify map output hint matches TBD spec
+
+        def validate_map_param(
+            name: str, param: Parameter, artifact: type[Artifact]
+        ) -> tuple[str, type[Artifact]]:
+            # TODO: Should we add some ArtifactPartition[MyArtifact] type?
+            if param.annotation != StoragePartitions:
+                raise ValueError("type hint must be `StoragePartitions`")
+            return name, artifact
+
+        map_input_metadata = MapInputMetadata(
+            cls._validate_parameters(map_sig, validator=validate_map_param)
+        )
+        return map_sig, map_input_metadata  # TODO: Verify map output hint matches TBD spec
 
     @classmethod
-    def _validate_cohesive_params(cls) -> None:
-        artifact_names = set(cls._artifact_fields_)
-        if unused_fields := artifact_names - (
+    def _validate_no_unused_fields(cls) -> None:
+        if unused_fields := set(cls._input_artifact_types_) - (
             set(cls._build_sig_.parameters) | set(cls._map_sig_.parameters)
         ):
             raise ValueError(
-                f"{cls.__name__} - the following fields aren't used in `.build` or `.map`: {unused_fields}"
+                f"the following fields aren't used in `.build` or `.map`: {unused_fields}"
             )
-        for (fn, sig) in (
-            ("build", cls._build_sig_),
-            ("map", cls._map_sig_),
-        ):
-            if undefined_params := set(sig.parameters) - artifact_names:
-                raise ValueError(
-                    f"{cls.__name__}.{fn} - the following parameter(s) must be defined as a field: {undefined_params}"
-                )
-            for param in sig.parameters.values():
-                field = cls.__fields__[param.name]
-                # TODO: Handle proper per-method annotations:
-                # - build: Replace with python type and match to a View + check if we can convert
-                #          from the Artifact's format/storage.
-                # - map:   Expect some ArtifactPartition[MyArtifact] type and match on wrapped type
-                #
-                # These checks should probably apply at class definition _and_ instantiation.
-                if field.type_ != param.annotation:
-                    raise ValueError(
-                        f"{cls.__name__}.{fn} - `{param.name}` parameter type hint must match the field: {field.type_}"
-                    )
 
     # NOTE: pydantic defines .__iter__ to return `self.__dict__.items()` to support `dict(model)`,
     # but we want to override to support easy expansion/assignment to a Graph  without `.out()` (eg:
@@ -176,16 +258,29 @@ def __iter__(self) -> Iterator[Artifact]:  # type: ignore
             ret = (ret,)
         return iter(ret)
 
-    @property
-    def fingerprint(self) -> Fingerprint:
-        """Return a Fingerprint of the Producer key + version.
+    def compute_input_fingerprint(
+        self, dependency_partitions: frozendict[str, StoragePartitions]
+    ) -> Fingerprint:
+        input_names = set(dependency_partitions)
+        expected_names = set(self._build_input_views_)
+        if input_names != expected_names:
+            raise ValueError(
+                f"Mismatched dependency inputs; expected {expected_names}, got {input_names}"
+            )
+        # We only care if the *code* or *input partition contents* changed, not if the input file
+        # paths changed (but have the same content as a prior run).
+        return Fingerprint.from_string(self._class_key_).combine(
+            self.version.fingerprint,
+            *(
+                partition.content_fingerprint
+                for name, partitions in dependency_partitions.items()
+                for partition in partitions
+            ),
+        )
 
-        The input and output Artifacts are ignored as the Producer instance may be used multiple
-        times to produce different output partitions. The "entropy mixing" will be performed for
-        *each* output with the static Producer.fingerprint + that output's specific input
-        partition dependencies.
-        """
-        return Fingerprint.from_string(self._class_key_).combine(self.version.fingerprint)
+    @property
+    def inputs(self) -> dict[str, Artifact]:
+        return {k: getattr(self, k) for k in self._input_artifact_types_}
 
     def out(self, *outputs: Artifact) -> Union[Artifact, tuple[Artifact, ...]]:
         """Configure the output Artifacts this Producer will build.
@@ -195,25 +290,81 @@ def out(self, *outputs: Artifact) -> Union[Artifact, tuple[Artifact, ...]]:
         if not outputs:
             # TODO: Raise a better error if the Artifacts don't have defaults set for
             # type/format/storage.
-            outputs = tuple(artifact() for artifact in self._build_sig_.return_annotation)
+            outputs = tuple(artifact() for (artifact, _) in self._output_metadata_)
         passed_n, expected_n = len(outputs), len(self._build_sig_.return_annotation)
         if passed_n != expected_n:
             ret_str = _commas(self._build_sig_.return_annotation)
             raise ValueError(
-                f"{self._class_key_}.out() - Expected {expected_n} arguments of ({ret_str}), but got: {outputs}"
+                f"{self._class_key_}.out() - expected {expected_n} arguments of ({ret_str}), but got: {outputs}"
             )
 
         def validate(artifact: Artifact, *, ord: int) -> Artifact:
-            expected_type = self._build_sig_.return_annotation[ord]
-            if not isinstance(artifact, expected_type):
-                raise ValueError(
-                    f"{self._class_key_}.out() - Expected the {ordinal(ord+1)} argument to be {expected_type}, got {type(artifact)}"
-                )
-            if artifact.producer is not None:
-                raise ValueError(f"{artifact} is produced by {artifact.producer}!")
-            return cast(Artifact, artifact.copy(update={"producer": self}))
+            (expected_type, _) = self._output_metadata_[ord]
+            with wrap_exc(ValueError, prefix=f"{self._class_key_}.out() {ordinal(ord+1)} argument"):
+                if not isinstance(artifact, expected_type):
+                    raise ValueError(f"expected instance of {expected_type}, got {type(artifact)}")
+                # TODO: Validate type/format/storage/view compatibility?
+                if artifact.producer_output is not None:
+                    raise ValueError(
+                        f"{artifact} is produced by {artifact.producer_output.producer}!"
+                    )
+            return artifact.copy(
+                update={"producer_output": ProducerOutput(producer=self, position=ord)}
+            )
 
         outputs = tuple(validate(artifact, ord=i) for i, artifact in enumerate(outputs))
         if len(outputs) == 1:
             return outputs[0]
         return outputs
+
+
+def producer(
+    *,
+    annotations: Optional[tuple[Annotation, ...]] = None,
+    map: Optional[MapSig] = None,
+    name: Optional[str] = None,
+    version: Optional[Version] = None,
+) -> Callable[[BuildSig], type[Producer]]:
+    def decorate(build: BuildSig) -> type[Producer]:
+        nonlocal name
+        name = build.__name__ if name is None else name
+        __annotations__ = {}
+        for param in signature(build).parameters.values():
+            with wrap_exc(ValueError, prefix=f"{name} {param.name} param"):
+                __annotations__[param.name] = Producer._get_artifact_from_annotation(
+                    param.annotation
+                )
+        # If overriding, set an explicit "annotations" hint until [1] is released.
+        #
+        # 1: https://github.com/samuelcolvin/pydantic/pull/3018
+        if annotations:
+            __annotations__["annotations"] = tuple[Annotation, ...]  # type: ignore
+        if version:
+            __annotations__["version"] = Version  # type: ignore
+        return type(
+            name,
+            (Producer,),
+            {
+                k: v
+                for k, v in {
+                    "__annotations__": __annotations__,
+                    "annotations": annotations,
+                    "build": staticmethod(build),
+                    "map": None if map is None else staticmethod(map),
+                    "version": version,
+                }.items()
+                if v is not None
+            },
+        )
+
+    return decorate
+
+
+class ProducerOutput(Model):
+    producer: Producer
+    position: int  # TODO: Support named output (defaulting to artifact classname)
+
+
+BaseArtifact.update_forward_refs(ProducerOutput=ProducerOutput)
+Statistic.update_forward_refs(ProducerOutput=ProducerOutput)
+Artifact.update_forward_refs(ProducerOutput=ProducerOutput)
diff --git a/src/arti/storage/__init__.py b/src/arti/storage/__init__.py
index 699ca595..058b8142 100644
--- a/src/arti/storage/__init__.py
+++ b/src/arti/storage/__init__.py
@@ -1,33 +1,67 @@
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
 
 import abc
+import os
 from typing import Any, ClassVar, Generic, Optional, TypeVar
 
 from arti.fingerprints import Fingerprint
 from arti.formats import Format
 from arti.internal.models import Model
-from arti.internal.type_hints import get_class_type_vars
-from arti.partitions import CompositeKey, PartitionKey
+from arti.internal.type_hints import get_class_type_vars, lenient_issubclass
+from arti.internal.utils import frozendict
+from arti.partitions import CompositeKey, CompositeKeyTypes
+from arti.storage._internal import InputFingerprints as InputFingerprints
+from arti.storage._internal import partial_format, strip_partition_indexes
 from arti.types import Type
 
+_StoragePartition = TypeVar("_StoragePartition", bound="StoragePartition")
+
 
 class StoragePartition(Model):
     keys: CompositeKey
-    fingerprint: Optional[Fingerprint] = None
+    input_fingerprint: Fingerprint = Fingerprint.empty()
+    content_fingerprint: Fingerprint = Fingerprint.empty()
 
-    def with_fingerprint(self) -> "StoragePartition":
-        return self.copy(update={"fingerprint": self.compute_fingerprint()})
+    def with_content_fingerprint(
+        self: _StoragePartition, keep_existing: bool = True
+    ) -> _StoragePartition:
+        if keep_existing and not self.content_fingerprint.is_empty:
+            return self
+        return self.copy(update={"content_fingerprint": self.compute_content_fingerprint()})
 
     @abc.abstractmethod
-    def compute_fingerprint(self) -> Fingerprint:
-        raise NotImplementedError("{type(self).__name__}.compute_fingerprint is not implemented!")
+    def compute_content_fingerprint(self) -> Fingerprint:
+        raise NotImplementedError(
+            "{type(self).__name__}.compute_content_fingerprint is not implemented!"
+        )
+
 
+StoragePartitions = tuple[StoragePartition, ...]  # type: ignore
 
-_StoragePartition = TypeVar("_StoragePartition", bound=StoragePartition)
+# mypy doesn't (yet?) support nested TypeVars[1], so mark internal as Any.
+#
+# 1: https://github.com/python/mypy/issues/2756
+_Storage = TypeVar("_Storage", bound="Storage[Any]")
 
 
 class Storage(Model, Generic[_StoragePartition]):
+    """Storage is a data reference identifying 1 or more partitions of data.
+
+    Storage fields should have defaults set with placeholders for tags and partition
+    keys. This allows automatic injection of the tags and partition keys for simple
+    cases.
+    """
+
     _abstract_ = True
+
+    # These separators are used in the default resolve_* helpers to format metadata into
+    # the storage fields.
+    #
+    # The defaults are tailored for "path"-like fields.
+    key_value_sep: ClassVar[str] = "="
+    partition_name_component_sep: ClassVar[str] = "_"
+    segment_sep: ClassVar[str] = os.sep
+
     storage_partition_type: ClassVar[type[_StoragePartition]]
 
     @classmethod
@@ -51,11 +85,118 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
                 f"{cls.__name__} fields must match {cls.storage_partition_type.__name__} ({expected_field_types}), got: {fields}"
             )
 
-    def supports(self, type_: Type, format: Format) -> None:
-        # TODO: Ensure the storage supports all of the specified types and partitioning on the
-        # specified field(s).
-        pass
+    @property
+    def _format_fields(self) -> frozendict[str, str]:
+        return frozendict(
+            {
+                name: value
+                for name in self.__fields__
+                if lenient_issubclass(type(value := getattr(self, name)), str)
+            }
+        )
 
     @abc.abstractmethod
-    def discover_partitions(self, **key_types: type[PartitionKey]) -> tuple[_StoragePartition, ...]:
+    def discover_partitions(
+        self,
+        key_types: CompositeKeyTypes,
+        input_fingerprints: InputFingerprints = InputFingerprints(),
+    ) -> tuple[_StoragePartition, ...]:
         raise NotImplementedError()
+
+    def generate_partition(
+        self,
+        keys: CompositeKey,
+        input_fingerprint: Fingerprint,
+        with_content_fingerprint: bool = True,
+    ) -> _StoragePartition:
+        format_kwargs = dict[Any, Any](keys)
+        if input_fingerprint.is_empty:
+            if self.includes_input_fingerprint_template:
+                raise ValueError(f"{self} requires an input_fingerprint, but none was provided")
+        else:
+            if not self.includes_input_fingerprint_template:
+                raise ValueError(f"{self} does not specify a {{input_fingerprint}} template")
+            format_kwargs["input_fingerprint"] = str(input_fingerprint.key)
+        field_values = {
+            name: (
+                strip_partition_indexes(original).format(**format_kwargs)
+                if lenient_issubclass(type(original := getattr(self, name)), str)
+                else original
+            )
+            for name in self.__fields__
+            if name in self.storage_partition_type.__fields__
+        }
+        partition = self.storage_partition_type(
+            input_fingerprint=input_fingerprint, keys=keys, **field_values
+        )
+        if with_content_fingerprint:
+            partition = partition.with_content_fingerprint()
+        return partition
+
+    @property
+    def includes_input_fingerprint_template(self) -> bool:
+        return any("{input_fingerprint}" in val for val in self._format_fields.values())
+
+    def _resolve_field(self, spec: str, placeholder_values: dict[str, str]) -> str:
+        for placeholder, value in placeholder_values.items():
+            if not value:
+                # Strip placeholder *and* any trailing self.segment_sep.
+                trim = "{" + placeholder + "}"
+                if f"{trim}{self.segment_sep}" in spec:
+                    trim = f"{trim}{self.segment_sep}"
+                spec = spec.replace(trim, "")
+        return partial_format(spec, **placeholder_values)
+
+    # TODO: Reconsider the `resolve_*` interfaces to reduce the number of copies if possible.
+    def resolve(self: _Storage, **placeholder_values: str) -> _Storage:
+        return self.copy(
+            update={
+                name: updated_value
+                for name, original in self._format_fields.items()
+                # Avoid "setting" the value if not updated to reduce pydantic repr verbosity (which
+                # only shows "set" fields by default).
+                if (updated_value := self._resolve_field(original, placeholder_values)) != original
+            }
+        )
+
+    def resolve_extension(self: _Storage, extension: Optional[str]) -> _Storage:
+        if extension is None:
+            return self.resolve(extension="")
+        return self.resolve(extension=extension)
+
+    def resolve_graph_name(self: _Storage, graph_name: str) -> _Storage:
+        return self.resolve(graph_name=graph_name)
+
+    def resolve_input_fingerprint(self: _Storage, input_fingerprint: Fingerprint) -> _Storage:
+        val = str(input_fingerprint.key)
+        if input_fingerprint.is_empty:
+            val = ""
+        return self.resolve(input_fingerprint=val)
+
+    def resolve_names(self: _Storage, names: tuple[str, ...]) -> _Storage:
+        name = names[-1] if names else ""
+        return self.resolve(names=self.segment_sep.join(names), name=name)
+
+    def resolve_partition_key_spec(self: _Storage, key_types: CompositeKeyTypes) -> _Storage:
+        key_component_specs = {
+            f"{name}{self.partition_name_component_sep}{component_name}": f"{{{name}.{component_spec}}}"
+            for name, pk in key_types.items()
+            for component_name, component_spec in pk.default_key_components.items()
+        }
+        return self.resolve(
+            partition_key_spec=self.segment_sep.join(
+                f"{name}{self.key_value_sep}{spec}" for name, spec in key_component_specs.items()
+            )
+        )
+
+    def resolve_path_tags(self: _Storage, path_tags: frozendict[str, str]) -> _Storage:
+        return self.resolve(
+            path_tags=self.segment_sep.join(
+                f"{tag}{self.key_value_sep}{value}" for tag, value in path_tags.items()
+            )
+        )
+
+    def supports(self: _Storage, type_: Type, format: Format) -> None:
+        # TODO: Ensure the storage supports all of the specified types and partitioning on the
+        # specified field(s).
+        pass
diff --git a/src/arti/storage/_internal.py b/src/arti/storage/_internal.py
index 67ef49eb..b5612252 100644
--- a/src/arti/storage/_internal.py
+++ b/src/arti/storage/_internal.py
@@ -1,34 +1,73 @@
 import string
 from collections import defaultdict
-from collections.abc import Mapping
-from typing import Any, Iterable, Literal, Optional, overload
+from collections.abc import Callable, Mapping
+from functools import partial
+from typing import Any, Iterable, Optional
 
 import parse
 
+from arti.fingerprints import Fingerprint
 from arti.internal.utils import frozendict
-from arti.partitions import CompositeKey, PartitionKey
+from arti.partitions import CompositeKey, CompositeKeyTypes, PartitionKey
 
+InputFingerprints = frozendict[CompositeKey, Fingerprint]
 
-class WildcardPlaceholder:
-    def __init__(self, name: str, key_type: type[PartitionKey]):
-        self._name = name
-        self._key_type = key_type
-        self._valid_attributes = set(key_type.__fields__) | key_type.key_components
+
+class Placeholder:
+    def __init__(self, key: str) -> None:
+        self._key = key
+
+
+class FormatPlaceholder(Placeholder):
+    def __format__(self, spec: str) -> str:
+        result = self._key
+        if spec:
+            result += ":" + spec
+        return "{" + result + "}"
+
+    def __getitem__(self, key: Any) -> "FormatPlaceholder":
+        self._key = f"{self._key}[{key}]"
+        return self
+
+    def __getattr__(self, attr: str) -> "FormatPlaceholder":
+        self._key = f"{self._key}.{attr}"
+        return self
+
+
+# Used to convert things like `/{date_key.Y[1970]` to `/{date_key.Y` so we can format in
+# *real* partition key values.
+class StripIndexPlaceholder(FormatPlaceholder):
+    def __getitem__(self, key: Any) -> "StripIndexPlaceholder":
+        return self
+
+
+class WildcardPlaceholder(Placeholder):
+    def __init__(self, key: str, key_types: CompositeKeyTypes):
+        super().__init__(key)
+        if key not in key_types:
+            raise ValueError(f"No '{key}' partition key found, expected one of {tuple(key_types)}")
+        self._key_type = key_types[key]
         self._attribute: Optional[str] = None  # What field/key_component is being accessed
 
+    @classmethod
+    def with_key_types(
+        cls, key_types: Mapping[str, type[PartitionKey]]
+    ) -> Callable[[str], "WildcardPlaceholder"]:
+        return partial(cls, key_types=key_types)
+
     def _err_if_no_attribute(self) -> None:
         if self._attribute is None:
             example = sorted(self._key_type.key_components)[0]
             raise ValueError(
-                f"'{self._name}' cannot be used directly in a partition path; access one of the key components (eg: '{self._name}.{example}')."
+                f"'{self._key}' cannot be used directly in a partition path; access one of the key components (eg: '{self._key}.{example}')."
             )
 
     def __getattr__(self, name: str) -> "WildcardPlaceholder":
         if self._attribute is not None:
             raise ValueError(
-                f"'{self._name}.{self._attribute}.{name}' cannot be used in a partition path; only immediate '{self._name}' attributes (such as '{self._attribute}') can be used."
+                f"'{self._key}.{self._attribute}.{name}' cannot be used in a partition path; only immediate '{self._key}' attributes (such as '{self._attribute}') can be used."
             )
-        if name not in self._valid_attributes:
+        if name not in self._key_type.key_components:
             raise AttributeError(
                 f"'{self._key_type.__name__}' has no field or key component '{name}'"
             )
@@ -49,68 +88,58 @@ def __str__(self) -> str:
 
 # NOTE: We might pre-populate this with the Graph tags so the hard coded templates are already available (rather than
 # being filled in with a WildcardPlaceholder).
-class WildcardFormatDict(dict[str, Any]):
+class FormatDict(dict[str, Any]):
     def __init__(
         self,
-        key_types: Mapping[str, type[PartitionKey]],
+        placeholder_type: Callable[[str], Placeholder],
         /,
         *args: Iterable[tuple[str, Any]],
         **kwargs: Any,
     ) -> None:
         super().__init__(*args, **kwargs)
-        self.key_types = key_types
+        self.placeholder_type = placeholder_type
 
-    def __missing__(self, key: str) -> WildcardPlaceholder:
-        if key not in self.key_types:
-            raise ValueError(
-                f"No '{key}' partition key found, expected one of {tuple(self.key_types)}"
-            )
-        return WildcardPlaceholder(name=key, key_type=self.key_types[key])
+    def __missing__(self, key: str) -> Placeholder:
+        return self.placeholder_type(key)
 
 
-def spec_to_wildcard(spec: str, key_types: Mapping[str, type[PartitionKey]]) -> str:
-    return string.Formatter().vformat(spec, (), WildcardFormatDict(key_types))
+def partial_format(spec: str, **kwargs: Any) -> str:
+    return string.Formatter().vformat(spec, (), FormatDict(FormatPlaceholder, **kwargs))
 
 
-@overload
-def extract_partition_keys(
-    *,
-    error_on_no_match: Literal[False],
-    key_types: Mapping[str, type[PartitionKey]],
-    parser: parse.Parser,
-    path: str,
-    spec: str,
-) -> Optional[CompositeKey]:
-    ...
+# This is hacky...
+def strip_partition_indexes(spec: str) -> str:
+    return string.Formatter().vformat(spec, (), FormatDict(StripIndexPlaceholder))
 
 
-@overload
-def extract_partition_keys(
-    *,
-    error_on_no_match: Literal[True] = True,
-    key_types: Mapping[str, type[PartitionKey]],
-    parser: parse.Parser,
-    path: str,
-    spec: str,
-) -> CompositeKey:
-    ...
+def spec_to_wildcard(spec: str, key_types: Mapping[str, type[PartitionKey]]) -> str:
+    return string.Formatter().vformat(
+        spec.replace("{input_fingerprint}", "*"),
+        (),
+        FormatDict(WildcardPlaceholder.with_key_types(key_types)),
+    )
 
 
-def extract_partition_keys(
+def extract_placeholders(
     *,
     error_on_no_match: bool = True,
     key_types: Mapping[str, type[PartitionKey]],
     parser: parse.Parser,
     path: str,
     spec: str,
-) -> Optional[CompositeKey]:
+) -> tuple[Fingerprint, CompositeKey]:
     parsed_value = parser.parse(path)
     if parsed_value is None:
         if error_on_no_match:
             raise ValueError(f"Unable to parse '{path}' with '{spec}'.")
-        return None
+        return Fingerprint.empty(), CompositeKey()
+    input_fingerprint = Fingerprint.empty()
     key_components = defaultdict[str, dict[str, str]](dict)
     for k, v in parsed_value.named.items():
+        if k == "input_fingerprint":
+            assert isinstance(v, str)
+            input_fingerprint = Fingerprint.from_int(int(v))
+            continue
         key, component = k.split(".")
         # parsing a string like "{date.Y[1970]}" will return a dict like {'1970': '1970'}.
         if isinstance(v, dict):
@@ -123,16 +152,24 @@ def extract_partition_keys(
     }
     if set(keys) != set(key_types):
         raise ValueError(
-            f"Expected to find partition keys for {sorted(key_types)}, only found {sorted(key)}. Is the partitioning spec ('{spec}') complete?"
+            f"Expected to find partition keys for {sorted(key_types)}, only found {sorted(keys)}. Is the partitioning spec ('{spec}') complete?"
         )
-    return frozendict(keys)
+    return input_fingerprint, frozendict(keys)
 
 
-def parse_partition_keys(
-    paths: set[str], *, spec: str, key_types: Mapping[str, type[PartitionKey]]
-) -> Mapping[str, CompositeKey]:
+def parse_spec(
+    paths: set[str],
+    *,
+    input_fingerprints: InputFingerprints = InputFingerprints(),
+    key_types: Mapping[str, type[PartitionKey]],
+    spec: str,
+) -> Mapping[str, tuple[Fingerprint, CompositeKey]]:
     parser = parse.compile(spec, case_sensitive=True)
     return {
-        path: extract_partition_keys(parser=parser, path=path, spec=spec, key_types=key_types)
-        for path in paths
+        path: (input_fingerprint, keys)
+        for (path, (input_fingerprint, keys)) in (
+            (path, extract_placeholders(parser=parser, path=path, spec=spec, key_types=key_types))
+            for path in paths
+        )
+        if input_fingerprints.get(keys, Fingerprint.empty()) == input_fingerprint
     }
diff --git a/src/arti/storage/local.py b/src/arti/storage/local.py
index c14650c5..bc579874 100644
--- a/src/arti/storage/local.py
+++ b/src/arti/storage/local.py
@@ -1,16 +1,20 @@
+from __future__ import annotations
+
 import hashlib
+import tempfile
 from glob import glob
+from pathlib import Path
 
 from arti.fingerprints import Fingerprint
-from arti.partitions import PartitionKey
-from arti.storage import Storage, StoragePartition
-from arti.storage._internal import parse_partition_keys, spec_to_wildcard
+from arti.partitions import CompositeKeyTypes
+from arti.storage import InputFingerprints, Storage, StoragePartition
+from arti.storage._internal import parse_spec, spec_to_wildcard
 
 
 class LocalFilePartition(StoragePartition):
     path: str
 
-    def compute_fingerprint(self, buffer_size: int = 1024 * 1024) -> Fingerprint:
+    def compute_content_fingerprint(self, buffer_size: int = 1024 * 1024) -> Fingerprint:
         with open(self.path, mode="rb") as f:
             sha = hashlib.sha1()
             data = f.read(buffer_size)
@@ -21,18 +25,31 @@ def compute_fingerprint(self, buffer_size: int = 1024 * 1024) -> Fingerprint:
 
 
 class LocalFile(Storage[LocalFilePartition]):
-    path: str
+    path: str = str(
+        Path(tempfile.gettempdir())
+        / "{graph_name}"
+        / "{path_tags}"
+        / "{names}"
+        / "{partition_key_spec}"
+        / "{input_fingerprint}"
+        / "{name}.{extension}"
+    )
 
     def discover_partitions(
-        self, **key_types: type[PartitionKey]
+        self,
+        key_types: CompositeKeyTypes,
+        input_fingerprints: InputFingerprints = InputFingerprints(),
     ) -> tuple[LocalFilePartition, ...]:
         wildcard = spec_to_wildcard(self.path, key_types)
         paths = set(glob(wildcard))
-        path_keys = parse_partition_keys(paths, spec=self.path, key_types=key_types)
+        path_metadata = parse_spec(
+            paths, spec=self.path, key_types=key_types, input_fingerprints=input_fingerprints
+        )
         return tuple(
             self.storage_partition_type(
-                path=path,
+                input_fingerprint=input_fingerprint,
                 keys=keys,
+                path=path,
             )
-            for path, keys in path_keys.items()
+            for path, (input_fingerprint, keys) in path_metadata.items()
         )
diff --git a/src/arti/types/__init__.py b/src/arti/types/__init__.py
index 9d544f9d..de98881d 100644
--- a/src/arti/types/__init__.py
+++ b/src/arti/types/__init__.py
@@ -4,7 +4,8 @@
 from operator import attrgetter
 from typing import Any, ClassVar, Literal, Optional
 
-from pydantic import PrivateAttr, root_validator, validator
+from pydantic import PrivateAttr, validator
+from pydantic.fields import ModelField
 
 from arti.internal.models import Model
 from arti.internal.type_hints import lenient_issubclass
@@ -15,6 +16,9 @@ class Type(Model):
     """Type represents a data type."""
 
     _abstract_ = True
+    # NOTE: Exclude the description to minimize fingerprint changes (and thus rebuilds).
+    _fingerprint_excludes_ = frozenset(["description"])
+
     description: Optional[str]
     nullable: bool = False
 
@@ -120,42 +124,22 @@ class Int64(_Int):
 
 
 class List(Type):
-    value_type: Type
-
-
-class Map(Type):
-    key_type: Type
-    value_type: Type
-
-
-class Null(Type):
-    pass
-
-
-class Set(Type):
-    value_type: Type
-
-
-class String(Type):
-    pass
-
-
-class Struct(Type, _NamedMixin):
-    fields: frozendict[str, Type]
+    element: Type
     partition_by: tuple[str, ...] = ()
     cluster_by: tuple[str, ...] = ()
 
     @validator("partition_by", "cluster_by")
     @classmethod
     def _validate_field_ref(
-        cls, references: tuple[str, ...], values: dict[str, Any]
+        cls, references: tuple[str, ...], values: dict[str, Any], field: ModelField
     ) -> tuple[str, ...]:
-        # Fields failed validation, so skip validating references
-        if (fields := values.get("fields")) is None:
+        if (element := values.get("element")) is None:
             return references
-        known, requested = set(fields), set(references)
+        if references and not isinstance(element, Struct):
+            raise ValueError("requires element to be a Struct")
+        known, requested = set(element.fields), set(references)
         if unknown := requested - known:
-            raise ValueError(f"Unknown field(s): {unknown}")
+            raise ValueError(f"unknown field(s): {unknown}")
         return references
 
     @validator("cluster_by")
@@ -166,26 +150,39 @@ def _validate_cluster_by(
         if (partition_by := values.get("partition_by")) is None:
             return cluster_by
         if overlapping := set(cluster_by) & set(partition_by):
-            raise ValueError(f"Clustering fields overlap with partition fields: {overlapping}")
+            raise ValueError(f"clustering fields overlap with partition fields: {overlapping}")
         return cluster_by
 
-    @root_validator
-    @classmethod
-    def _validate_nested_struct(cls, values: dict[str, Any]) -> dict[str, Any]:
-        subfields_with_root_settings = {
-            name
-            for name, type_ in values.get("fields", {}).items()
-            if isinstance(type_, Struct) and (type_.partition_by or type_.cluster_by)
-        }
-        if subfields_with_root_settings:
-            raise ValueError(
-                f"Nested Structs ({subfields_with_root_settings}) cannot set `partition_by` or `cluster_by`"
-            )
-        return values
+    @property
+    def is_partitioned(self) -> bool:
+        return bool(self.partition_fields)
 
     @property
     def partition_fields(self) -> frozendict[str, Type]:
-        return frozendict({name: self.fields[name] for name in self.partition_by})
+        if not isinstance(self.element, Struct):
+            return frozendict()
+        return frozendict({name: self.element.fields[name] for name in self.partition_by})
+
+
+class Map(Type):
+    key: Type
+    value: Type
+
+
+class Null(Type):
+    pass
+
+
+class Set(Type):
+    element: Type
+
+
+class String(Type):
+    pass
+
+
+class Struct(Type, _NamedMixin):
+    fields: frozendict[str, Type]
 
 
 class Timestamp(Type):
@@ -301,3 +298,8 @@ def to_system(self, type_: Type, *, hints: dict[str, Any]) -> Any:
             if adapter.matches_artigraph(type_, hints=hints):
                 return adapter.to_system(type_, hints=hints)
         raise NotImplementedError(f"No {self} adapter for Artigraph type: {type_}.")
+
+    def check_similarity(self, *, arti: Type, python_type: type) -> None:
+        system_type = self.to_system(arti, hints={})
+        if not lenient_issubclass(system_type, python_type):
+            raise ValueError(f"{python_type} cannot be used to represent {arti}")
diff --git a/src/arti/types/python.py b/src/arti/types/python.py
index 03fb6162..3baf22ee 100644
--- a/src/arti/types/python.py
+++ b/src/arti/types/python.py
@@ -2,11 +2,16 @@
 from collections.abc import Mapping
 from functools import partial
 from itertools import chain
-from typing import _TypedDictMeta  # type: ignore
 from typing import Any, Literal, Optional, TypedDict, Union, get_args, get_origin, get_type_hints
 
 import arti.types
-from arti.internal.type_hints import NoneType, is_optional_hint, is_union, lenient_issubclass
+from arti.internal.type_hints import (
+    NoneType,
+    is_optional_hint,
+    is_typeddict,
+    is_union,
+    lenient_issubclass,
+)
 from arti.types import Type, TypeAdapter, TypeSystem, _ScalarClassTypeAdapter
 
 python_type_system = TypeSystem(key="python")
@@ -44,9 +49,9 @@ def to_artigraph(cls, type_: Any, *, hints: dict[str, Any]) -> Type:
 class PyValueContainer(TypeAdapter):
     @classmethod
     def to_artigraph(cls, type_: Any, *, hints: dict[str, Any]) -> Type:
-        (value_type,) = get_args(type_)
+        (element,) = get_args(type_)
         return cls.artigraph(
-            value_type=python_type_system.to_artigraph(value_type, hints=hints),
+            element=python_type_system.to_artigraph(element, hints=hints),
         )
 
     @classmethod
@@ -57,7 +62,7 @@ def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:
     def to_system(cls, type_: Type, *, hints: dict[str, Any]) -> Any:
         assert isinstance(type_, cls.artigraph)
         return cls.system[
-            python_type_system.to_system(type_.value_type, hints=hints),  # type: ignore
+            python_type_system.to_system(type_.element, hints=hints),  # type: ignore
         ]
 
 
@@ -163,10 +168,10 @@ class PyMap(TypeAdapter):
 
     @classmethod
     def to_artigraph(cls, type_: Any, *, hints: dict[str, Any]) -> Type:
-        key_type, value_type = get_args(type_)
+        key, value = get_args(type_)
         return cls.artigraph(
-            key_type=python_type_system.to_artigraph(key_type, hints=hints),
-            value_type=python_type_system.to_artigraph(value_type, hints=hints),
+            key=python_type_system.to_artigraph(key, hints=hints),
+            value=python_type_system.to_artigraph(value, hints=hints),
         )
 
     @classmethod
@@ -177,8 +182,8 @@ def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:
     def to_system(cls, type_: Type, *, hints: dict[str, Any]) -> Any:
         assert isinstance(type_, cls.artigraph)
         return cls.system[
-            python_type_system.to_system(type_.key_type, hints=hints),
-            python_type_system.to_system(type_.value_type, hints=hints),
+            python_type_system.to_system(type_.key, hints=hints),
+            python_type_system.to_system(type_.value, hints=hints),
         ]  # type: ignore
 
 
@@ -231,7 +236,7 @@ def to_artigraph(cls, type_: Any, *, hints: dict[str, Any]) -> Type:
     def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:
         # NOTE: This check is probably a little shaky, particularly across python versions. Consider
         # using the typing_inspect package.
-        return isinstance(type_, _TypedDictMeta)
+        return is_typeddict(type_)
 
     @classmethod
     def to_system(cls, type_: Type, *, hints: dict[str, Any]) -> Any:
diff --git a/src/arti/types/sgqlc.py b/src/arti/types/sgqlc.py
index c7f5e5ac..bae5fc80 100644
--- a/src/arti/types/sgqlc.py
+++ b/src/arti/types/sgqlc.py
@@ -138,7 +138,7 @@ class SgqlcList(TypeAdapter):
     def to_artigraph(cls, type_: Any, *, hints: dict[str, Any]) -> at.Type:
         assert issubclass(type_, cls.system)
         # list_of subclasses the input type (hence, we can fetch it with .mro)
-        return cls.artigraph(value_type=sgqlc_type_system.to_artigraph(type_.mro()[1], hints=hints))
+        return cls.artigraph(element=sgqlc_type_system.to_artigraph(type_.mro()[1], hints=hints))
 
     @classmethod
     def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:
@@ -147,7 +147,7 @@ def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:
     @classmethod
     def to_system(cls, type_: at.Type, *, hints: dict[str, Any]) -> Any:
         assert isinstance(type_, cls.artigraph)
-        return st.list_of(sgqlc_type_system.to_system(type_.value_type, hints=hints))
+        return st.list_of(sgqlc_type_system.to_system(type_.element, hints=hints))
 
 
 class _StructAdapter(TypeAdapter):
diff --git a/src/arti/versions/__init__.py b/src/arti/versions/__init__.py
index 5aeed00f..1afb0313 100644
--- a/src/arti/versions/__init__.py
+++ b/src/arti/versions/__init__.py
@@ -2,7 +2,6 @@
 
 import inspect
 import subprocess
-from abc import abstractmethod
 from collections.abc import Callable
 from datetime import datetime, timezone
 from typing import Any, cast
@@ -11,17 +10,11 @@
 
 from arti.fingerprints import Fingerprint
 from arti.internal.models import Model
-from arti.internal.utils import qname
 
 
 class Version(Model):
     _abstract_ = True
 
-    @property
-    @abstractmethod
-    def fingerprint(self) -> Fingerprint:
-        raise NotImplementedError(f"{qname(self)}.fingerprint is not implemented!")
-
 
 class GitCommit(Version):
     sha: str = Field(
@@ -30,10 +23,6 @@ class GitCommit(Version):
         )
     )
 
-    @property
-    def fingerprint(self) -> Fingerprint:
-        return Fingerprint.from_string(self.sha)
-
 
 class SemVer(Version):
     """SemVer fingerprinting only considers the major component, unless it is less than 0.
@@ -57,10 +46,6 @@ def fingerprint(self) -> Fingerprint:
 class String(Version):
     value: str
 
-    @property
-    def fingerprint(self) -> Fingerprint:
-        return Fingerprint.from_string(self.value)
-
 
 class _SourceDescriptor:  # Experimental :)
     # Using AST rather than literal source will likely be less "noisy":
diff --git a/src/arti/views/__init__.py b/src/arti/views/__init__.py
index aa90ba5c..d2de9fa4 100644
--- a/src/arti/views/__init__.py
+++ b/src/arti/views/__init__.py
@@ -1,9 +1,10 @@
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # type: ignore
 
-from typing import Any, ClassVar
+from typing import Annotated, Any, ClassVar, get_args, get_origin
 
 from arti.internal.models import Model
-from arti.internal.utils import register
+from arti.internal.type_hints import lenient_issubclass
+from arti.internal.utils import import_submodules, register
 from arti.types import TypeSystem
 
 
@@ -14,7 +15,7 @@ class View(Model):
     """
 
     _abstract_ = True
-    _registry_: ClassVar[dict[type, type["View"]]] = {}
+    _by_python_type_: "ClassVar[dict[type, type[View]]]" = {}
 
     priority: ClassVar[int] = 0  # Set priority of this view for its python_type. Higher is better.
     python_type: ClassVar[type]
@@ -24,4 +25,25 @@ class View(Model):
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
         if not cls._abstract_:
-            register(cls._registry_, cls.python_type, cls, lambda x: x.priority)
+            register(cls._by_python_type_, cls.python_type, cls, lambda x: x.priority)
+
+    @classmethod
+    def get_class_for(cls, annotation: Any) -> type["View"]:
+        import_submodules(__path__, __name__)
+
+        origin, args = get_origin(annotation), get_args(annotation)
+        if origin is Annotated:
+            annotation, *hints = args
+            views: list[type[View]] = [hint for hint in hints if lenient_issubclass(hint, View)]
+            if len(views) == 0:
+                return cls.get_class_for(annotation)
+            if len(views) == 1:
+                return views[0]
+            raise ValueError("multiple Views set")
+        if origin is None:
+            origin = annotation
+        if origin not in cls._by_python_type_:
+            raise ValueError(
+                f"{annotation} cannot be matched to a View, try setting one explicitly (eg: `Annotated[pd.DataFrame, MyArtifact, PandasDataFrame]`)"
+            )
+        return cls._by_python_type_[origin]
diff --git a/src/arti/views/python.py b/src/arti/views/python.py
index 41216e82..6a26a286 100644
--- a/src/arti/views/python.py
+++ b/src/arti/views/python.py
@@ -1,65 +1,43 @@
-import pickle
 from datetime import date, datetime
-from typing import Any, ClassVar
 
-from arti.formats.pickle import Pickle
 from arti.internal.type_hints import NoneType
-from arti.io import read, write
-from arti.storage.local import LocalFilePartition
-from arti.types import TypeSystem
 from arti.types.python import python_type_system
 from arti.views import View
 
 
-class _PythonBuiltin(View):
+class PythonBuiltin(View):
     _abstract_ = True
 
-    type_system: ClassVar[TypeSystem] = python_type_system
+    type_system = python_type_system
 
 
-class Date(_PythonBuiltin):
+class Date(PythonBuiltin):
     python_type = date
 
 
-class Datetime(_PythonBuiltin):
+class Datetime(PythonBuiltin):
     python_type = datetime
 
 
-class Dict(_PythonBuiltin):
+class Dict(PythonBuiltin):
     python_type = dict
 
 
-class Float(_PythonBuiltin):
+class Float(PythonBuiltin):
     python_type = float
 
 
-class Int(_PythonBuiltin):
+class Int(PythonBuiltin):
     python_type = int
 
 
-class Null(_PythonBuiltin):
-    python_type = NoneType
-
-
-class Str(_PythonBuiltin):
-    python_type = str
+class List(PythonBuiltin):
+    python_type = list
 
 
-def _read_pickle_file(path: str) -> Any:
-    with open(path, "rb") as file:
-        return pickle.load(file)
-
-
-@read.register
-def _read_pickle_localfile_python(
-    *, format: Pickle, storage_partitions: list[LocalFilePartition], view: _PythonBuiltin
-) -> Any:
-    return [_read_pickle_file(storage_partition.path) for storage_partition in storage_partitions]
+class Null(PythonBuiltin):
+    python_type = NoneType
 
 
-@write.register
-def _write_pickle_localfile_python(
-    data: Any, *, format: Pickle, storage_partition: LocalFilePartition, view: _PythonBuiltin
-) -> None:
-    with open(storage_partition.path, "wb") as file:
-        pickle.dump(data, file)
+class Str(PythonBuiltin):
+    python_type = str


Test patch:
diff --git a/tests/arti/artifacts/test_artifact.py b/tests/arti/artifacts/test_artifact.py
index 5c14261b..1cc32347 100644
--- a/tests/arti/artifacts/test_artifact.py
+++ b/tests/arti/artifacts/test_artifact.py
@@ -1,10 +1,14 @@
+from typing import Any
+
 import pytest
 
 from arti.annotations import Annotation
 from arti.artifacts import Artifact
 from arti.formats import Format
+from arti.partitions import CompositeKeyTypes, Int64Key
 from arti.statistics import Statistic
-from arti.types import Int64, Type
+from arti.storage import Storage
+from arti.types import Int64, List, Struct, Type
 from tests.arti.dummies import A1, A2, P1, P2, DummyFormat, DummyStatistic, DummyStorage
 
 
@@ -31,6 +35,12 @@ class BadStorage(DummyStorage):
         def supports(self, type_: Type, format: Format) -> None:
             raise ValueError("Storage - Boo!")
 
+    with pytest.raises(ValueError, match="MissingTypeArtifact must set `type`") as exc:
+
+        class MissingTypeArtifact(Artifact):
+            format: DummyFormat = DummyFormat()
+            storage: DummyStorage = DummyStorage()
+
     with pytest.raises(ValueError, match="Format - Boo!"):
 
         class BadFormatArtifact(Artifact):
@@ -49,20 +59,26 @@ class BadStorageArtifact(Artifact):
 
         BadStorageArtifact()
 
-    with pytest.raises(ValueError, match="Expected an instance of") as exc:
+    with pytest.raises(ValueError, match="overriding `type` is not supported") as exc:
 
-        class BadTypeArtifact(Artifact):
-            format: BadFormat = BadFormat()
-            storage: BadStorage = BadStorage()
+        class OverrideTypeArtifact(Artifact):
+            type: Int64 = Int64()
+            format: DummyFormat = DummyFormat()
+            storage: DummyStorage = DummyStorage()
 
-        BadTypeArtifact(type="junk")
+        OverrideTypeArtifact(type=Int64(nullable=True))
     # Confirm {Format,Storage}.support are not called
     with pytest.raises(AssertionError):
         exc.match("Format - Boo!")
     with pytest.raises(AssertionError):
         exc.match("Storage - Boo!")
 
-    Artifact(type=Int64(), format=DummyFormat(), storage=DummyStorage())
+    class GoodArtifact(Artifact):
+        type: Type = Int64()
+        format: Format = DummyFormat()
+        storage: Storage[Any] = DummyStorage()
+
+    GoodArtifact()
 
 
 def test_instance_attr_merging() -> None:
@@ -85,3 +101,29 @@ class MyArtifact(A1):
     artifact = MyArtifact(annotations=[Ann2(y=10)], statistics=[Stat2()])
     assert tuple(type(a) for a in artifact.annotations) == (Ann1, Ann2)
     assert tuple(type(s) for s in artifact.statistics) == (Stat1, Stat2)
+
+
+def test_Artifact_partition_key_types() -> None:
+    class NonPartitioned(Artifact):
+        type: List = List(element=Struct(fields={"a": Int64()}))
+
+    assert NonPartitioned.partition_key_types == CompositeKeyTypes()
+    assert not NonPartitioned.is_partitioned
+
+    class Partitioned(Artifact):
+        type: List = List(element=Struct(fields={"a": Int64()}), partition_by=("a",))
+
+    assert Partitioned.partition_key_types == CompositeKeyTypes({"a": Int64Key})
+    assert Partitioned.is_partitioned
+
+
+def test_Artifact_storage_path_resolution() -> None:
+    class S(DummyStorage):
+        key = "test-{partition_key_spec}"
+
+    class A(Artifact):
+        type: Type = List(element=Struct(fields={"a": Int64()}), partition_by=("a",))
+        format: Format = DummyFormat()
+        storage: S
+
+    assert A(storage=S()).storage.key == "test-a_key={a.key}"
diff --git a/tests/arti/dummies.py b/tests/arti/dummies.py
index 25d89718..28411cb1 100644
--- a/tests/arti/dummies.py
+++ b/tests/arti/dummies.py
@@ -1,11 +1,15 @@
+from __future__ import annotations
+
+from typing import Annotated, Optional
+
 from arti.annotations import Annotation
 from arti.artifacts import Artifact
 from arti.fingerprints import Fingerprint
 from arti.formats import Format
-from arti.partitions import PartitionKey
+from arti.partitions import CompositeKey, CompositeKeyTypes
 from arti.producers import Producer
 from arti.statistics import Statistic
-from arti.storage import Storage, StoragePartition
+from arti.storage import InputFingerprints, Storage, StoragePartition
 from arti.types import Int32, Struct, TypeSystem
 
 dummy_type_system = TypeSystem(key="dummy")
@@ -20,13 +24,23 @@ class DummyFormat(Format):
 
 
 class DummyPartition(StoragePartition):
-    def compute_fingerprint(self) -> Fingerprint:
-        return Fingerprint(key=5)
+    key: str = "test"
+
+    def compute_content_fingerprint(self) -> Fingerprint:
+        return Fingerprint.from_string(self.key)
 
 
 class DummyStorage(Storage[DummyPartition]):
-    def discover_partitions(self, **key_types: type[PartitionKey]) -> tuple[DummyPartition, ...]:
-        return ()
+    key: str = "test-{graph_name}-{path_tags}-{names}-{partition_key_spec}-{input_fingerprint}-{name}.{extension}"
+
+    def discover_partitions(
+        self, key_types: CompositeKeyTypes, input_fingerprints: Optional[InputFingerprints] = None
+    ) -> tuple[DummyPartition, ...]:
+        if key_types != CompositeKeyTypes():
+            raise NotImplementedError()
+        if input_fingerprints is not None and input_fingerprints != InputFingerprints():
+            raise NotImplementedError()
+        return (self.storage_partition_type(keys=CompositeKey(), key=self.key),)
 
 
 # NOTE: when using a subclass of the original type hint, we must override[1].
@@ -68,16 +82,16 @@ class P1(Producer):
     a1: A1
 
     @staticmethod
-    def build(a1: A1) -> A2:
-        return A2()
+    def build(a1: dict) -> Annotated[dict, A2]:  # type: ignore
+        return {}
 
 
 class P2(Producer):
     a2: A2
 
     @staticmethod
-    def build(a2: A2) -> tuple[A3, A4]:
-        return A3(), A4()
+    def build(a2: dict) -> tuple[Annotated[dict, A3], Annotated[dict, A4]]:  # type: ignore
+        return {}, {}
 
 
 class P3(Producer):
@@ -85,5 +99,5 @@ class P3(Producer):
     a2: A2
 
     @staticmethod
-    def build(a1: A1, a2: A2) -> tuple[A3, A4]:
-        return A3(), A4()
+    def build(a1: dict, a2: dict) -> tuple[Annotated[dict, A3], Annotated[dict, A4]]:  # type: ignore
+        return {}, {}
diff --git a/tests/arti/graphs/test_graph.py b/tests/arti/graphs/test_graph.py
index 1dcd9001..cb705055 100644
--- a/tests/arti/graphs/test_graph.py
+++ b/tests/arti/graphs/test_graph.py
@@ -1,33 +1,188 @@
+import re
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Annotated, cast
+
 import pytest
 from box import BoxError
 
+from arti.artifacts import Artifact
+from arti.executors.local import LocalExecutor
+from arti.fingerprints import Fingerprint
+from arti.formats.json import JSON
 from arti.graphs import Graph
-from tests.arti.dummies import A1, A2, A3, A4, P1, P2, P3
+from arti.internal.utils import frozendict
+from arti.partitions import CompositeKey
+from arti.producers import producer
+from arti.storage.local import LocalFile, LocalFilePartition
+from arti.types import Int64
+from arti.views import python as python_views
+from tests.arti.dummies import A1, A2, A3, A4, P1, P2
+
+
+# TODO: Add a test-scoped fixture to generate these (may need to be a func that can generate multiple).
+class Num(Artifact):
+    type: Int64 = Int64()
+    format: JSON = JSON()
+    # Require callers to set the storage instance in a separate tempdir.
+    storage: LocalFile
+
+
+@producer()
+def div(a: Annotated[int, Num], b: Annotated[int, Num]) -> Annotated[int, Num]:
+    return a // b
 
 
-def test_Graph() -> None:
+@pytest.fixture
+def graph() -> Graph:
     # NOTE: .out() supports strict Artifact subclass mypy typing with the mypy_plugin, but Producers
     # also support simple iteration (eg: `a, b = MyProducer(...)`).
-    with Graph("test") as g1:
-        g1.artifacts.a = A1()
-        g1.artifacts.b.c = P1(a1=g1.artifacts.a).out()
-        g1.artifacts.d, g1.artifacts.e = P2(a2=g1.artifacts.b.c).out()
-        g1.artifacts.f, g1.artifacts.g = P3(a1=g1.artifacts.a, a2=g1.artifacts.b.c).out()
+    with Graph(name="test") as g:
+        g.artifacts.a = A1()
+        g.artifacts.b = P1(a1=g.artifacts.a).out()
+        g.artifacts.c.a, g.artifacts.c.b = P2(a2=g.artifacts.b).out()
+    return g
+
+
+def test_Graph(graph: Graph) -> None:
+    assert isinstance(graph.artifacts.a, A1)
+    assert isinstance(graph.artifacts.b, A2)
+    assert isinstance(graph.artifacts.c.a, A3)
+    assert isinstance(graph.artifacts.c.b, A4)
+    assert not graph.artifacts.a.storage.includes_input_fingerprint_template
+    assert graph.artifacts.b.storage.includes_input_fingerprint_template
+    assert graph.artifacts.c.a.storage.includes_input_fingerprint_template
+    assert graph.artifacts.c.b.storage.includes_input_fingerprint_template
+
+
+def test_Graph_compute_id() -> None:
+    with Graph(name="test") as g:
+        g.artifacts.a = A1()
+        p1 = P1(a1=g.artifacts.a)
+        g.artifacts.b = cast(A2, p1.out())
+
+    id_components = [
+        g.fingerprint,
+        Fingerprint.from_string("a"),
+        Fingerprint.from_string("b"),
+        g.artifacts.a.fingerprint,
+        g.artifacts.b.fingerprint,
+        p1.fingerprint,
+        *(
+            storage_partition.with_content_fingerprint().fingerprint
+            for storage_partition in g.artifacts.a.discover_storage_partitions()
+        ),
+    ]
+
+    assert g.compute_id() == Fingerprint.combine(*id_components)
+    # Ensure order independence
+    assert g.compute_id() == Fingerprint.combine(*reversed(id_components))
+
+
+def test_Graph_compute_id_missing_input_artifact() -> None:
+    with TemporaryDirectory() as _dir:
+        dir = Path(_dir)
+        with Graph(name="test") as g:
+            g.artifacts.a = Num(storage=LocalFile(path=str(dir / "a.json")))
+
+        with pytest.raises(ValueError, match=re.escape("No data (partitions) found for `a`")):
+            assert g.compute_id()
+
+
+def test_Graph_compute_id_producer_arg_order() -> None:
+    with TemporaryDirectory() as _dir:
+        dir = Path(_dir)
+        a = Num(storage=LocalFile(path=str(dir / "a.json")))
+        with open(a.storage.path, "w") as f:
+            f.write("10")
+        b = Num(storage=LocalFile(path=str(dir / "b.json")))
+        with open(b.storage.path, "w") as f:
+            f.write("5")
+        c = Num(storage=LocalFile(path=str(dir / "{input_fingerprint}/c.json")))
+
+        # Create two Graphs, varying only by the arg order to the Producer.
+        with Graph(name="test") as g_ab:
+            g_ab.artifacts.c = div(a=a, b=b).out(c)
+        with Graph(name="test") as g_ba:
+            g_ba.artifacts.c = div(a=b, b=a).out(c)
+
+        assert g_ab.compute_id() != g_ba.compute_id()
+
 
-    with Graph("test-2") as g2:
-        g2.artifacts.upstream.a = g1.artifacts.a
+def test_Graph_build() -> None:
+    side_effect = 0
 
-    assert isinstance(g1.artifacts.a, A1)
-    assert isinstance(g1.artifacts.b.c, A2)
-    assert isinstance(g1.artifacts.d, A3)
-    assert isinstance(g1.artifacts.e, A4)
-    assert isinstance(g1.artifacts.f, A3)
-    assert isinstance(g1.artifacts.g, A4)
-    assert g1.artifacts.a is g2.artifacts.upstream.a
+    @producer()
+    def increment(i: Annotated[int, Num]) -> Annotated[int, Num]:
+        nonlocal side_effect
+        side_effect += 1
+        return i + 1
+
+    @producer()
+    def dup(i: Annotated[int, Num]) -> tuple[Annotated[int, Num], Annotated[int, Num]]:
+        return i, i
+
+    with TemporaryDirectory() as _dir:
+        dir = Path(_dir)
+        with Graph(name="test") as g:
+            g.artifacts.a = Num(storage=LocalFile(path=str(dir / "a.json")))
+            g.artifacts.b = increment(i=g.artifacts.a).out(
+                Num(storage=LocalFile(path=str(dir / "b/{input_fingerprint}.json")))
+            )
+            # Test multiple return values
+            g.artifacts.c, g.artifacts.d = dup(i=g.artifacts.a).out(
+                Num(storage=LocalFile(path=str(dir / "c/{input_fingerprint}.json"))),
+                Num(storage=LocalFile(path=str(dir / "d/{input_fingerprint}.json"))),
+            )
+
+        a, b, c, d = (
+            g.artifacts.a,
+            cast(Num, g.artifacts.b),
+            cast(Num, g.artifacts.c),
+            cast(Num, g.artifacts.d),
+        )
+        # Bootstrap the initial artifact and build
+        g.write(0, artifact=a)
+        g.build()
+        assert side_effect == 1
+        assert g.read(b, annotation=int) == 1
+        assert g.read(c, annotation=int) == g.read(d, annotation=int) == 0
+        # A second build should no-op
+        g.build(executor=LocalExecutor())
+        assert side_effect == 1
+        assert g.read(b, annotation=int) == 1
+        assert g.read(c, annotation=int) == g.read(d, annotation=int) == 0
+        # Changing the raw Artifact data should trigger a rerun
+        g.write(1, artifact=a)
+        g.build()
+        assert side_effect == 2
+        assert g.read(b, annotation=int) == 2
+        assert g.read(c, annotation=int) == g.read(d, annotation=int) == 1
+        # Changing back to the original data should no-op
+        g.write(0, artifact=a)
+        g.build()
+        assert side_effect == 2
+        assert g.read(b, annotation=int) == 1
+        assert g.read(c, annotation=int) == g.read(d, annotation=int) == 0
+
+
+def test_Graph_dependencies(graph: Graph) -> None:
+    p1 = graph.artifacts.b.producer_output.producer
+    p2 = graph.artifacts.c.a.producer_output.producer
+    assert graph.dependencies == frozendict(
+        {
+            graph.artifacts.a: frozenset(),
+            p1: frozenset({graph.artifacts.a}),
+            graph.artifacts.b: frozenset({p1}),
+            p2: frozenset({graph.artifacts.b}),
+            graph.artifacts.c.a: frozenset({p2}),
+            graph.artifacts.c.b: frozenset({p2}),
+        }
+    )
 
 
 def test_Graph_errors() -> None:
-    with Graph("test") as graph:
+    with Graph(name="test") as graph:
         graph.artifacts.a = A1()
         graph.artifacts.b = P1(a1=graph.artifacts.a).out()
 
@@ -35,3 +190,89 @@ def test_Graph_errors() -> None:
         graph.artifacts.a = A1()
     with pytest.raises(AttributeError, match="has no attribute"):
         graph.artifacts.z
+
+    with Graph(name="outer"):
+        with pytest.raises(ValueError, match="Another graph is being defined"):
+            with Graph(name="inner"):
+                pass
+
+
+def test_Graph_producers(graph: Graph) -> None:
+    p1 = graph.artifacts.b.producer_output.producer
+    p2 = graph.artifacts.c.a.producer_output.producer
+    assert graph.producers == frozenset({p1, p2})
+
+
+def test_Graph_producer_output(graph: Graph) -> None:
+    p1 = graph.artifacts.b.producer_output.producer
+    p2 = graph.artifacts.c.a.producer_output.producer
+    assert graph.producer_outputs == frozendict(
+        {
+            p1: (graph.artifacts.b,),
+            p2: (graph.artifacts.c.a, graph.artifacts.c.b),
+        }
+    )
+
+    with Graph(name="test") as g:
+        with pytest.raises(
+            ValueError,
+            match="producer_outputs cannot be used while the Graph is still being defined",
+        ):
+            g.producer_outputs
+
+
+def test_Graph_read_write() -> None:
+    with TemporaryDirectory() as _dir:
+        dir = Path(_dir)
+        with Graph(name="test") as g:
+            g.artifacts.i = Num(storage=LocalFile(path=str(dir / "i.json")))
+
+        i = g.artifacts.i
+        # Test write
+        storage_partition = g.write(5, artifact=i)
+        assert isinstance(storage_partition, LocalFilePartition)
+        assert storage_partition.content_fingerprint != Fingerprint.empty()
+        assert storage_partition.input_fingerprint == Fingerprint.empty()
+        assert storage_partition.keys == CompositeKey()
+        assert storage_partition.path.endswith(i.format.extension)
+        with pytest.raises(
+            ValueError,
+            match=re.escape("Writing to a raw Artifact (`i`) would cause a `graph_id` change."),
+        ):
+            g.write(10, artifact=i, graph_id=Fingerprint.from_string("junk"))
+        # Test read
+        assert g.read(i, annotation=int) == 5
+        assert g.read(i, view=python_views.Int()) == 5
+        assert g.read(i, annotation=int, storage_partitions=[storage_partition]) == 5
+        with pytest.raises(ValueError, match="Either `annotation` or `view` must be passed"):
+            g.read(i)
+        with pytest.raises(ValueError, match="Only one of `annotation` or `view` may be passed"):
+            g.read(i, annotation=int, view=python_views.Int())
+
+
+def test_Graph_references(graph: Graph) -> None:
+    with Graph(name="test-2") as g2:
+        g2.artifacts.upstream.a = graph.artifacts.a
+    assert graph.artifacts.a == g2.artifacts.upstream.a
+
+
+def test_Graph_storage_resolution() -> None:
+    with Graph(name="test", path_tags={"tag": "value"}) as g:
+        g.artifacts.root.a = Num(storage=LocalFile())
+        g.artifacts.root.b = Num(storage=LocalFile())
+        g.artifacts.c = cast(
+            Num, div(a=g.artifacts.root.a, b=g.artifacts.root.b).out(Num(storage=LocalFile()))
+        )
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Produced Artifacts must have a '{input_fingerprint}' template in their Storage"
+            ),
+        ):
+            g.artifacts.d = div(a=g.artifacts.root.a, b=g.artifacts.root.b).out(
+                Num(storage=LocalFile(path="junk"))
+            )
+
+    assert g.artifacts.root.a.storage.path.endswith("/test/tag=value/root/a/a.json")
+    assert g.artifacts.root.b.storage.path.endswith("/test/tag=value/root/b/b.json")
+    assert g.artifacts.c.storage.path.endswith("/test/tag=value/c/{input_fingerprint}/c.json")
diff --git a/tests/arti/internal/import_submodules_test_modules/__init__.py b/tests/arti/internal/import_submodules_test_modules/__init__.py
new file mode 100644
index 00000000..249315e4
--- /dev/null
+++ b/tests/arti/internal/import_submodules_test_modules/__init__.py
@@ -0,0 +1,1 @@
+entries = set[str]()
diff --git a/tests/arti/internal/import_submodules_test_modules/a.py b/tests/arti/internal/import_submodules_test_modules/a.py
new file mode 100644
index 00000000..9c668970
--- /dev/null
+++ b/tests/arti/internal/import_submodules_test_modules/a.py
@@ -0,0 +1,3 @@
+from tests.arti.internal.import_submodules_test_modules import entries
+
+entries.add(__file__)
diff --git a/tests/arti/internal/import_submodules_test_modules/sub/b.py b/tests/arti/internal/import_submodules_test_modules/sub/b.py
new file mode 100644
index 00000000..9c668970
--- /dev/null
+++ b/tests/arti/internal/import_submodules_test_modules/sub/b.py
@@ -0,0 +1,3 @@
+from tests.arti.internal.import_submodules_test_modules import entries
+
+entries.add(__file__)
diff --git a/tests/arti/internal/import_submodules_test_modules/sub/folder/c.py b/tests/arti/internal/import_submodules_test_modules/sub/folder/c.py
new file mode 100644
index 00000000..9c668970
--- /dev/null
+++ b/tests/arti/internal/import_submodules_test_modules/sub/folder/c.py
@@ -0,0 +1,3 @@
+from tests.arti.internal.import_submodules_test_modules import entries
+
+entries.add(__file__)
diff --git a/tests/arti/internal/test_internal.py b/tests/arti/internal/test_internal.py
index c95b694d..44658065 100644
--- a/tests/arti/internal/test_internal.py
+++ b/tests/arti/internal/test_internal.py
@@ -1,5 +1,25 @@
-from arti.internal import version
+import pytest
+
+from arti.internal import version, wrap_exc
 
 
 def test_version() -> None:
     assert version is not None
+
+
+def test_wrap_exc() -> None:
+    root_error = ValueError("test")
+    with pytest.raises(ValueError, match="a - test") as exc:
+        with wrap_exc(ValueError, prefix="a"):
+            raise root_error
+    assert exc.value.__cause__ is root_error
+
+    with pytest.raises(ValueError, match="ab - test") as exc:
+        with wrap_exc(ValueError, prefix="a"):
+            with wrap_exc(ValueError, prefix="b"):
+                raise root_error
+    assert exc.value.__cause__ is root_error
+
+    with wrap_exc(ValueError, prefix="shouldn't run"):
+        x = 5
+    assert x == 5
diff --git a/tests/arti/internal/test_models.py b/tests/arti/internal/test_models.py
index 594f46b0..bd535e8d 100644
--- a/tests/arti/internal/test_models.py
+++ b/tests/arti/internal/test_models.py
@@ -1,9 +1,11 @@
+import re
 from contextlib import nullcontext
 from typing import Annotated, Any, ClassVar, Literal, Optional, Union
 
 import pytest
 from pydantic import ValidationError
 
+from arti.fingerprints import Fingerprint
 from arti.internal.models import Model
 from arti.internal.utils import frozendict
 
@@ -16,6 +18,10 @@ class Concrete(Model):
     pass
 
 
+class Sub(Model):
+    x: int
+
+
 def test_Model() -> None:
     obj = Concrete()
     assert str(obj) == repr(obj)
@@ -24,6 +30,64 @@ def test_Model() -> None:
             model()
 
 
+def test_Model_copy_validation() -> None:
+    v = Sub(x=5)
+    v1 = v.copy(update={"x": 10})
+    assert v.x == 5
+    assert v1.x == 10
+    with pytest.raises(ValueError, match="expected an instance"):
+        v.copy(update={"x": "junk"})
+    v2 = v.copy(update={"x": "junk"}, validate=False)  # Skip validation for "trusted" data.
+    assert v2.x == "junk"  # type: ignore
+
+
+def test_Model_fingerprint() -> None:
+    class A(Model):
+        a: int
+        b: str
+
+    a = A(a=1, b="b")
+    assert a.fingerprint == Fingerprint.from_string('A:{"a": 1, "b": "b"}')
+
+    class B(Model):
+        a: A
+        b: set[str]
+
+    assert B(a=a, b={"b"}).fingerprint == Fingerprint.from_string(
+        f'B:{{"a": {a.fingerprint.key}, "b": ["b"]}}'
+    )
+
+    class Excludes(A):
+        _fingerprint_excludes_ = frozenset(["a"])
+
+    assert Excludes(a=1, b="b").fingerprint == Fingerprint.from_string('Excludes:{"b": "b"}')
+
+    with pytest.raises(ValueError, match=re.escape("Unknown `_fingerprint_excludes_` field(s)")):
+
+        class BadExcludes(A):
+            _fingerprint_excludes_ = frozenset(["z"])
+
+    class Includes(A):
+        _fingerprint_includes_ = frozenset(["a"])
+
+    assert Includes(a=1, b="b").fingerprint == Fingerprint.from_string('Includes:{"a": 1}')
+
+    with pytest.raises(ValueError, match=re.escape("Unknown `_fingerprint_includes_` field(s)")):
+
+        class BadIncludes(A):
+            _fingerprint_includes_ = frozenset(["z"])
+
+
+def test_Model__iter() -> None:
+    class A(Model):
+        a: int
+
+    a = A(a=5)
+    assert dict(a._iter()) == {"a": 5}
+    a.__dict__["b"] = "junk"  # Shouldn't show up!
+    assert dict(a._iter()) == {"a": 5}
+
+
 def test_Model_unknown_kwargs() -> None:
     with pytest.raises(ValidationError, match="extra fields not permitted"):
         Concrete(junk=1)
@@ -39,16 +103,12 @@ class M(Model):
     # frozendict is special cased in the type conversions to automatically convert dicts.
     m = M(a=5, b="b", c=0, d={"a": 1})
     assert isinstance(m.d, frozendict)
-    with pytest.raises(ValidationError, match="Expected an instance of <class 'str'>, got"):
+    with pytest.raises(ValidationError, match="expected an instance of <class 'str'>, got"):
         M(a=5, b=5, c=0)
-    with pytest.raises(ValidationError, match=r"Expected an instance of <class 'int'>, got"):
+    with pytest.raises(ValidationError, match=r"expected an instance of <class 'int'>, got"):
         M(a=5, b="b", c=0.0)
 
 
-class Sub(Model):
-    x: int
-
-
 @pytest.mark.parametrize(
     ("hint", "value", "error_type"),
     [
diff --git a/tests/arti/internal/test_type_hints.py b/tests/arti/internal/test_type_hints.py
index 5812e17d..ef32ee09 100644
--- a/tests/arti/internal/test_type_hints.py
+++ b/tests/arti/internal/test_type_hints.py
@@ -1,5 +1,5 @@
 from collections.abc import Mapping, Sequence
-from typing import Any, Literal, Optional, Tuple, Union, get_origin
+from typing import Annotated, Any, Literal, Optional, Tuple, TypedDict, Union, get_origin
 
 import pytest
 
@@ -16,9 +16,19 @@ class MyTuple(tuple):  # type: ignore
     pass
 
 
+class MyTypedDict(TypedDict):
+    a: int
+
+
 @pytest.mark.parametrize(
     ("klass", "class_or_tuple"),
     (
+        (Annotated[int, 5], Annotated[int, 5]),
+        (Annotated[int, 5], int),
+        (Annotated[list[int], 5], Annotated[list[int], 5]),
+        (Annotated[list[int], 5], Annotated[list, 5]),
+        (Annotated[list[int], 5], list),
+        (Annotated[list[int], 5], list[int]),
         (MyTuple, tuple),
         (NoneType, (int, NoneType)),
         (NoneType, Optional[int]),
@@ -26,9 +36,13 @@ class MyTuple(tuple):  # type: ignore
         (dict[str, int], Mapping[str, int]),
         (dict[str, int], dict),
         (dict[str, int], dict[str, int]),
+        (MyTypedDict, dict),
         (int, (int, str)),
+        (int, Annotated[int, 5]),
         (int, Optional[int]),
         (int, Union[int, str]),
+        (list[int], Annotated[list[int], 5]),
+        (list[int], Annotated[list, 5]),
         (str, Union[int, str]),
         (tuple, Tuple),
         (tuple, tuple),
@@ -52,6 +66,8 @@ def test_lenient_issubclass_true(
 @pytest.mark.parametrize(
     ("klass", "class_or_tuple"),
     (
+        (MyTypedDict, dict[str, int]),  # Might implement in the future
+        (MyTypedDict, dict[str, str]),
         (dict, Mapping[str, int]),
         (dict, dict[str, int]),
         (dict[str, str], Mapping[str, int]),
@@ -81,6 +97,9 @@ def test_lenient_issubclass_error_cases() -> None:
     assert not lenient_issubclass(5, 5)  # type: ignore
     with pytest.raises(TypeError, match="arg 2 must be a class"):
         lenient_issubclass(tuple, 5)  # type: ignore
+    # Might support this in the future
+    with pytest.raises(TypeError, match="TypedDict does not support instance and class checks"):
+        lenient_issubclass(dict, MyTypedDict)
 
 
 def test_is_optional_hint() -> None:
diff --git a/tests/arti/internal/test_utils.py b/tests/arti/internal/test_utils.py
index ddfbedec..2134adae 100644
--- a/tests/arti/internal/test_utils.py
+++ b/tests/arti/internal/test_utils.py
@@ -1,9 +1,12 @@
 import math
 import operator as op
+import os
 import re
 from collections.abc import Callable
 from copy import deepcopy
 from functools import partial
+from pathlib import Path
+from types import ModuleType
 from typing import Any, Union, cast, get_args, get_origin
 
 import pytest
@@ -15,6 +18,7 @@
     classproperty,
     dispatch,
     frozendict,
+    import_submodules,
     int64,
     named_temporary_file,
     ordinal,
@@ -66,7 +70,7 @@ class B1(B):
 
 
 def test_dispatch() -> None:
-    @dispatch
+    @dispatch()
     def dispatch_test(a: A, b: B) -> Any:
         return "good_a_b"
 
@@ -139,6 +143,32 @@ def test_frozendict() -> None:
     assert get_args(frozendict[str, int]) == (str, int)
 
 
+# NOTE: We don't test the thread safety here
+def test_import_submodules() -> None:
+    from tests.arti.internal import import_submodules_test_modules
+
+    basedir = Path(import_submodules_test_modules.__file__).parent
+    modules = {
+        "a",
+        "sub.b",
+        "sub.folder.c",
+    }
+
+    assert import_submodules_test_modules.entries == set()
+    output = import_submodules(
+        import_submodules_test_modules.__path__,  # type: ignore
+        import_submodules_test_modules.__name__,
+    )
+    assert {
+        name.replace(f"{import_submodules_test_modules.__name__}.", "") for name in output
+    } == modules
+    assert all(isinstance(mod, ModuleType) for mod in output.values())
+    assert {
+        str(Path(path).relative_to(basedir)).replace(".py", "")
+        for path in import_submodules_test_modules.entries
+    } == {name.replace(".", os.sep) for name in modules}
+
+
 class _I(_int):
     pass
 
diff --git a/tests/arti/io/test_localfile_io.py b/tests/arti/io/test_localfile_io.py
new file mode 100644
index 00000000..fb08c94b
--- /dev/null
+++ b/tests/arti/io/test_localfile_io.py
@@ -0,0 +1,108 @@
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+import pytest
+
+from arti import io
+from arti.artifacts import Artifact
+from arti.fingerprints import Fingerprint
+from arti.formats import Format
+from arti.formats.json import JSON
+from arti.formats.pickle import Pickle
+from arti.internal.utils import frozendict
+from arti.partitions import CompositeKey, Int64Key
+from arti.storage import StoragePartition
+from arti.storage.local import LocalFile
+from arti.types import Int64, List, Struct
+from arti.views import View
+
+
+class Num(Artifact):
+    type: Int64 = Int64()
+
+
+class PartitionedNum(Artifact):
+    type: List = List(element=Struct(fields={"i": Int64()}), partition_by=("i",))
+
+
+@pytest.mark.parametrize(
+    ["format"],
+    [
+        (JSON(),),
+        (Pickle(),),
+    ],
+)
+def test_localfile_io(format: Format) -> None:
+    with TemporaryDirectory() as _dir:
+        dir = Path(_dir)
+        a, n = Num(format=format, storage=LocalFile(path=str(dir / "a"))), 5
+
+        io.write(
+            n,
+            a.type,
+            a.format,
+            a.storage.generate_partition(
+                keys=CompositeKey(),
+                input_fingerprint=Fingerprint.empty(),
+                with_content_fingerprint=False,
+            ),
+            view=View.get_class_for(int)(),
+        )
+        partitions = a.discover_storage_partitions()
+        assert len(partitions) == 1
+        assert (
+            io.read(
+                a.type,
+                a.format,
+                partitions,
+                view=View.get_class_for(int)(),
+            )
+            == n
+        )
+        with pytest.raises(ValueError, match="Multiple partitions can only be read into a list"):
+            io.read(
+                a.type,
+                a.format,
+                partitions * 2,
+                view=View.get_class_for(int)(),
+            )
+
+
+@pytest.mark.parametrize(
+    ["format"],
+    [
+        (JSON(),),
+        (Pickle(),),
+    ],
+)
+def test_localfile_io_partitioned(format: Format) -> None:
+    with TemporaryDirectory() as _dir:
+        dir = Path(_dir)
+        a = PartitionedNum(format=format, storage=LocalFile(path=str(dir / "{i.key}")))
+        data: dict[frozendict[str, int], StoragePartition] = {
+            frozendict(i=i): a.storage.generate_partition(
+                keys=CompositeKey(i=Int64Key(key=i)),
+                input_fingerprint=Fingerprint.empty(),
+                with_content_fingerprint=False,
+            )
+            for i in [1, 2]
+        }
+        for record, partition in data.items():
+            io.write(
+                [record],
+                a.type,
+                a.format,
+                partition,
+                view=View.get_class_for(list)(),
+            )
+        assert set(data.values()) == set(a.discover_storage_partitions())
+        for record, partition in data.items():
+            assert (
+                io.read(
+                    a.type,
+                    a.format,
+                    [partition],
+                    view=View.get_class_for(list)(),
+                )
+                == [record]
+            )
diff --git a/tests/arti/partitions/test_partitions.py b/tests/arti/partitions/test_partitions.py
index 0165730f..39c30ab6 100644
--- a/tests/arti/partitions/test_partitions.py
+++ b/tests/arti/partitions/test_partitions.py
@@ -16,7 +16,7 @@
     _IntKey,
     key_component,
 )
-from arti.types import Date, Int8, Int16, Int32, Int64, Struct, Type
+from arti.types import Date, Int8, Int16, Int32, Int64, List, Struct, Type
 
 
 def test_PartitionKey_key_components() -> None:
@@ -26,27 +26,22 @@ def test_PartitionKey_key_components() -> None:
 
     components = Int8Key.key_components
     assert isinstance(components, frozenset)
-    assert components == {"hex"}
+    assert components == {"key", "hex"}
 
     assert isinstance(Int8Key.hex, key_component)
     assert isinstance(Int8Key.hex, property)
 
 
 def test_PartitionKey_lookup_from_type() -> None:
-    assert PartitionKey.key_type_for(Int8()) is Int8Key
-    assert PartitionKey.key_type_for(Int16()) is Int16Key
-    assert PartitionKey.key_type_for(Date()) is DateKey
-
-    assert PartitionKey.key_types_from(Int8()) == frozendict()
-    assert (
-        PartitionKey.key_types_from(
-            Struct(
-                fields={"date": Date(), "i": Int8()},
-                partition_by=("date", "i"),
-            )
-        )
-        == frozendict({"date": DateKey, "i": Int8Key})
-    )
+    assert PartitionKey.get_class_for(Int8()) is Int8Key
+    assert PartitionKey.get_class_for(Int16()) is Int16Key
+    assert PartitionKey.get_class_for(Date()) is DateKey
+
+    assert PartitionKey.types_from(Int8()) == frozendict()
+    assert PartitionKey.types_from(Struct(fields={"date": Date(), "i": Int8()})) == frozendict()
+    assert PartitionKey.types_from(
+        List(element=Struct(fields={"date": Date(), "i": Int8()}), partition_by=("date", "i"))
+    ) == frozendict({"date": DateKey, "i": Int8Key})
 
 
 def test_PartitionKey_subclass() -> None:
@@ -54,12 +49,31 @@ class AbstractKey(PartitionKey):
         _abstract_ = True
         _by_type_: ClassVar[dict[type[Type], type[PartitionKey]]] = {}
 
-    with pytest.raises(TypeError, match="must set `matching_type`"):
+        key: int
+
+    with pytest.raises(TypeError, match="NoDefaultKey must set `default_key_components`"):
 
-        class JunkKey(AbstractKey):
+        class NoDefaultKey(AbstractKey):
             pass
 
+    with pytest.raises(TypeError, match="NoMatchingTypeKey must set `matching_type`"):
+
+        class NoMatchingTypeKey(AbstractKey):
+            default_key_components: ClassVar[frozendict[str, str]] = frozendict(key="key")
+
+    with pytest.raises(
+        TypeError,
+        match=re.escape(
+            r"Unknown key_components in UnknownDefaultKey.default_key_components: {'junk'}"
+        ),
+    ):
+
+        class UnknownDefaultKey(AbstractKey):
+            default_key_components: ClassVar[frozendict[str, str]] = frozendict(junk="junk")
+            matching_type = Int8
+
     class SomeKey(AbstractKey):
+        default_key_components: ClassVar[frozendict[str, str]] = frozendict(key="key")
         matching_type = Int8
 
     assert AbstractKey._by_type_ == {Int8: SomeKey}
diff --git a/tests/arti/producers/test_producer.py b/tests/arti/producers/test_producer.py
index b029ba97..88bb797c 100644
--- a/tests/arti/producers/test_producer.py
+++ b/tests/arti/producers/test_producer.py
@@ -1,10 +1,19 @@
-from typing import Any, Optional
+import re
+from typing import Annotated, Optional
 
 import pytest
 
+from arti.annotations import Annotation
+from arti.artifacts import Artifact
 from arti.fingerprints import Fingerprint
 from arti.internal.models import Model
-from arti.producers import Producer
+from arti.internal.utils import frozendict
+from arti.producers import PartitionDependencies, Producer
+from arti.producers import producer as producer_decorator  # Avoid shadowing
+from arti.storage import StoragePartitions
+from arti.types import Int64, List, Struct
+from arti.versions import String as StringVersion
+from arti.views import python as python_views
 from tests.arti.dummies import A1, A2, A3, A4, P1, P2
 
 
@@ -12,11 +21,11 @@ class DummyProducer(Producer):
     a1: A1
 
     @staticmethod
-    def build(a1: A1) -> tuple[A2, A3]:
+    def build(a1: dict) -> tuple[Annotated[dict, A2], Annotated[dict, A3]]:  # type: ignore
         pass
 
     @staticmethod
-    def map(a1: A1) -> Any:
+    def map(a1: StoragePartitions) -> PartitionDependencies:
         pass
 
 
@@ -34,13 +43,116 @@ def test_Producer() -> None:
         assert isinstance(output, expected_output_classes[i])
 
 
+def test_producer_decorator() -> None:
+    @producer_decorator()
+    def dummy_producer(a1: Annotated[dict, A1]) -> Annotated[dict, A2]:  # type: ignore
+        return {}
+
+    assert dummy_producer.__name__ == "dummy_producer"
+    assert dummy_producer._input_artifact_types_ == frozendict(a1=A1)
+    assert len(dummy_producer._output_metadata_) == 1
+    assert dummy_producer._output_metadata_[0][0] == A2
+    assert dummy_producer(a1=A1()).annotations == Producer.__fields__["annotations"].default
+    assert dummy_producer(a1=A1()).version == Producer.__fields__["version"].default
+
+    class MyAnnotation(Annotation):
+        pass
+
+    def mapper() -> PartitionDependencies:
+        return PartitionDependencies()
+
+    @producer_decorator(
+        annotations=(MyAnnotation(),), map=mapper, name="test", version=StringVersion(value="test")
+    )
+    def dummy_producer2(a1: Annotated[dict, A1]) -> Annotated[dict, A2]:  # type: ignore
+        return {}
+
+    assert dummy_producer2.__name__ == "test"
+    assert dummy_producer2.map == mapper
+    assert dummy_producer2(a1=A1()).annotations == (MyAnnotation(),)
+    assert dummy_producer2(a1=A1()).version == StringVersion(value="test")
+
+
+def test_Producer_partitioned_input_validation() -> None:
+    class A(Artifact):
+        type: List = List(element=Struct(fields={"x": Int64()}), partition_by=("x",))
+
+    class P(Producer):
+        a: A
+
+        @staticmethod
+        def build(a: list[dict]) -> Annotated[dict, A2]:  # type: ignore
+            pass
+
+    assert P._input_artifact_types_ == frozendict(a=A)
+    assert P._build_input_views_ == frozendict(a=python_views.List)
+
+    with pytest.raises(ValueError, match="dict.* cannot be used to represent List"):
+
+        class SingularInput(Producer):
+            a: A
+
+            @staticmethod
+            def build(a: dict) -> Annotated[dict, A2]:  # type: ignore
+                pass
+
+    with pytest.raises(ValueError, match=re.escape("list[int] cannot be used to represent List")):
+
+        class IncompatibleInput(Producer):
+            a: A
+
+            @staticmethod
+            def build(a: list[int]) -> Annotated[dict, A]:  # type: ignore
+                pass
+
+
+def test_Producer_output_metadata() -> None:
+    assert DummyProducer._output_metadata_ == ((A2, python_views.Dict), (A3, python_views.Dict))
+
+    class ExplicitView(Producer):
+        a1: A1
+
+        @staticmethod
+        def build(a1: dict) -> Annotated[dict, A2, python_views.Dict]:  # type: ignore
+            pass
+
+    assert ExplicitView._output_metadata_ == ((A2, python_views.Dict),)
+
+    with pytest.raises(ValueError, match="DupView.build 1st return - multiple Views set"):
+
+        class DupView(Producer):
+            a1: A1
+
+            @staticmethod
+            def build(a1: dict) -> Annotated[dict, A2, python_views.Dict, python_views.Int]:  # type: ignore
+                pass
+
+    with pytest.raises(ValueError, match="NoArtifact.build 1st return - Artifact is not set "):
+
+        class NoArtifact(Producer):
+            a1: A1
+
+            @classmethod
+            def build(cls, a1: dict) -> Annotated[None, 5]:  # type: ignore
+                pass
+
+    with pytest.raises(ValueError, match="DupArtifact.build 1st return - multiple Artifacts set"):
+
+        class DupArtifact(Producer):
+            a1: A1
+
+            @staticmethod
+            def build(a1: dict) -> Annotated[dict, A1, A2]:  # type: ignore
+                pass
+
+
 def test_Producer_string_annotation() -> None:
     # This may be from `x: "Type"` or `from __future__ import annotations`.
     class StrAnnotation(Producer):
         a1: "A1"
 
         @staticmethod
-        def build(a1: "A1") -> "A2":
+        def build(a1: "dict") -> "Annotated[dict, A2]":  # type: ignore
             pass
 
     assert isinstance(StrAnnotation(a1=A1()).out(), A2)
@@ -48,7 +160,30 @@ def build(a1: "A1") -> "A2":
 
 def test_Producer_fingerprint() -> None:
     p1 = P1(a1=A1())
-    assert p1.fingerprint == Fingerprint.from_string("P1") ^ p1.version.fingerprint
+    assert p1.fingerprint == Fingerprint.from_string(
+        f'P1:{{"a1": {p1.a1.fingerprint.key}, "version": {p1.version.fingerprint.key}}}'
+    )
+
+
+def test_Producer_compute_input_fingerprint() -> None:
+    p1 = P1(a1=A1())
+    assert p1.compute_input_fingerprint(
+        frozendict(a1=StoragePartitions())
+    ) == Fingerprint.from_string(p1._class_key_).combine(p1.version.fingerprint)
+
+    storage_partition = p1.a1.storage.storage_partition_type(
+        keys={}, content_fingerprint=Fingerprint.from_int(10)
+    )
+    assert p1.compute_input_fingerprint(
+        frozendict(a1=StoragePartitions([storage_partition]))
+    ) == Fingerprint.from_string(p1._class_key_).combine(
+        p1.version.fingerprint, storage_partition.content_fingerprint
+    )
+
+    with pytest.raises(
+        ValueError, match=re.escape("Mismatched dependency inputs; expected {'a1'}, got {'junk'}")
+    ):
+        p1.compute_input_fingerprint(frozendict(junk=StoragePartitions()))
 
 
 def test_Producer_out() -> None:
@@ -59,20 +194,22 @@ def test_Producer_out() -> None:
     # multi return Producer
     p2 = P2(a2=a2)
     a3_, a4_ = p2.out(a3, a4)
-    for (producer, inp, out, type_) in (
-        (p1, a2, a2_, A2),
-        (p2, a3, a3_, A3),
-        (p2, a4, a4_, A4),
+    for (producer, inp, out, type_, position) in (
+        (p1, a2, a2_, A2, 0),
+        (p2, a3, a3_, A3, 0),
+        (p2, a4, a4_, A4, 1),
     ):
         assert inp is not out
         assert isinstance(out, type_)
-        assert out.producer is producer
-        check_model_matches(inp, out, exclude={"producer"})
+        assert out.producer_output is not None
+        assert out.producer_output.producer == producer
+        assert out.producer_output.position == position
+        check_model_matches(inp, out, exclude={"producer_output"})
     assert list(p1) == [a2_]
     assert list(p2) == [a3_, a4_]
 
 
-@pytest.mark.xfail(raises=ValueError)
+@pytest.mark.xfail(raises=NotImplementedError)
 def test_Producer_map_defaults() -> None:
     p1 = P1(a1=A1())
     # We can make .map defaulting a bit smarter by inspecting how the input artifacts are
@@ -80,6 +217,79 @@ def test_Producer_map_defaults() -> None:
     p1.map()  # ValueError
 
 
+def test_Producer_map_artifacts() -> None:
+    class P(Producer):
+        a1: A1
+
+        @staticmethod
+        def build(a1: dict) -> Annotated[dict, A2]:  # type: ignore
+            pass
+
+        @staticmethod
+        def map(a1: StoragePartitions) -> PartitionDependencies:
+            pass
+
+    assert P._map_input_metadata_ == frozendict(a1=A1)
+
+    with pytest.raises(
+        ValueError,
+        match="BadMapParam.map a1 param - type hint must be `StoragePartitions`",
+    ):
+
+        class BadMapParam(P):
+            @staticmethod
+            def map(a1: list) -> PartitionDependencies:  # type: ignore
+                pass
+
+
+def test_Producer_build_outputs_check() -> None:
+    class A(Artifact):
+        type: Int64 = Int64()
+
+    class B(Artifact):
+        type: Int64 = Int64()
+
+    class C(Artifact):
+        type: List = List(element=Struct(fields={"a": Int64()}), partition_by=("a",))
+
+    class D(Artifact):
+        type: List = List(element=Struct(fields={"a": Int64(), "b": Int64()}), partition_by=("b",))
+
+    class NoPartitioning(Producer):
+        @staticmethod
+        def build() -> tuple[Annotated[int, A], Annotated[int, B]]:
+            pass
+
+    class MatchingPartitioning(Producer):
+        @staticmethod
+        def build() -> tuple[Annotated[list[dict], C], Annotated[list[dict], C]]:  # type: ignore
+            pass
+
+        @staticmethod
+        def map() -> PartitionDependencies:
+            return PartitionDependencies()
+
+    for first_output in [Annotated[int, A], Annotated[list[dict], C]]:  # type: ignore
+        with pytest.raises(
+            ValueError, match="all output Artifacts must have the same partitioning scheme"
+        ):
+
+            class MixedPartitioning(Producer):
+                @staticmethod
+                def build() -> tuple[first_output, Annotated[list[dict], D]]:  # type: ignore
+                    pass
+
+    with pytest.raises(
+        ValueError,
+        match=r"BadProducer.map - must be implemented when the `build` outputs are partitioned",
+    ):
+
+        class BadProducer(Producer):  # noqa: F811
+            @staticmethod
+            def build() -> Annotated[list[dict], C]:  # type: ignore
+                pass
+
+
 def test_Producer_bad_signature() -> None:  # noqa: C901
     # pylint: disable=function-redefined
 
@@ -87,7 +297,7 @@ def test_Producer_bad_signature() -> None:  # noqa: C901
     class OkProducer(Producer):
         _abstract_ = True
 
-    with pytest.raises(ValueError, match="Producers must implement"):
+    with pytest.raises(ValueError, match="BadProducer.build - must be implemented"):
 
         class BadProducer(Producer):
             pass
@@ -99,7 +309,7 @@ class BadProducer(Producer):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             @classmethod
-            def build(cls, a1: A1) -> A2:
+            def build(cls, a1: dict) -> Annotated[dict, A2]:  # type: ignore
                 pass
 
     with pytest.raises(
@@ -109,11 +319,11 @@ def build(cls, a1: A1) -> A2:
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             @classmethod
-            def build(cls) -> A2:
+            def build(cls) -> Annotated[dict, A2]:  # type: ignore
                 pass
 
             @classmethod
-            def map(cls, a1: A1) -> A2:
+            def map(cls, a1: StoragePartitions) -> PartitionDependencies:
                 pass
 
     with pytest.raises(
@@ -126,7 +336,7 @@ class BadProducer(Producer):  # type: ignore # noqa: F811
             a2: A2
 
             @classmethod
-            def build(cls, a1: A1) -> A3:
+            def build(cls, a1: dict) -> Annotated[dict, A3]:  # type: ignore
                 pass
 
     with pytest.raises(ValueError, match="must have a type hint"):
@@ -153,94 +363,61 @@ class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1
 
             @classmethod
-            def build(cls, a1: A1 = A1()):  # type: ignore
+            def build(cls, a1: dict = A1()):  # type: ignore
                 pass
 
-    with pytest.raises(ValueError, match="parameter must be usable as a keyword argument"):
+    with pytest.raises(ValueError, match="must be usable as a keyword argument"):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1
 
             @classmethod
-            def build(cls, a1: A1, /):  # type: ignore
-                pass
-
-    with pytest.raises(ValueError, match="parameter must be usable as a keyword argument"):
-
-        class BadProducer(Producer):  # type: ignore # noqa: F811
-            @classmethod
-            def build(cls, *args: A1):  # type: ignore
+            def build(cls, a1: dict, /):  # type: ignore
                 pass
 
-    with pytest.raises(ValueError, match="parameter must be usable as a keyword argument"):
-
-        class BadProducer(Producer):  # type: ignore # noqa: F811
-            @classmethod
-            def build(cls, **kwargs: A1):  # type: ignore
-                pass
-
-    with pytest.raises(ValueError, match="A return value must be set"):
+    with pytest.raises(ValueError, match="must be usable as a keyword argument"):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1
 
             @classmethod
-            def build(cls, a1: A1):  # type: ignore
+            def build(cls, *a1: dict):  # type: ignore
                 pass
 
-    with pytest.raises(ValueError, match="return value must be an Artifact"):
+    with pytest.raises(ValueError, match="must be usable as a keyword argument"):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1
 
             @classmethod
-            def build(cls, a1: A1) -> None:
+            def build(cls, **a1: dict):  # type: ignore
                 pass
 
-    with pytest.raises(ValueError, match="return value must be an Artifact"):
+    with pytest.raises(ValueError, match="a return value must be set"):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1
 
             @classmethod
-            def build(cls, a1: A1) -> str:
+            def build(cls, a1: dict):  # type: ignore
                 pass
 
-    with pytest.raises(ValueError, match="return value must be an Artifact"):
-
-        class BadProducer(Producer):  # type: ignore # noqa: F811
-            a1: A1
-
-            @classmethod
-            def build(cls, a1: A1) -> tuple[A2, str]:
-                pass
-
-    with pytest.raises(
-        ValueError,
-        match=r"BadProducer.build - `a1` parameter type hint must match the field: <class 'tests.arti.dummies.A1'>",
-    ):
+    with pytest.raises(ValueError, match="missing return signature"):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1
 
             @classmethod
-            def build(cls, a1: A2) -> A2:
+            def build(cls, a1: dict) -> None:  # type: ignore
                 pass
 
-    with pytest.raises(
-        ValueError,
-        match=r"BadProducer.map - `a1` parameter type hint must match the field: <class 'tests.arti.dummies.A1'>",
-    ):
+    with pytest.raises(ValueError, match="2nd return - must be an Annotated hint"):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1
 
             @classmethod
-            def build(cls, a1: A1) -> A2:
-                pass
-
-            @classmethod
-            def map(self, a1: A2) -> Any:
+            def build(cls, a1: dict) -> tuple[Annotated[dict, A2], str]:  # type: ignore
                 pass
 
     with pytest.raises(
@@ -251,7 +428,7 @@ class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1 = None  # type: ignore
 
             @classmethod
-            def build(cls, a1: A1):  # type: ignore
+            def build(cls, a1: dict):  # type: ignore
                 pass
 
     with pytest.raises(
@@ -262,7 +439,7 @@ class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: Optional[A1]
 
             @classmethod
-            def build(cls, a1: A1):  # type: ignore
+            def build(cls, a1: dict):  # type: ignore
                 pass
 
     with pytest.raises(
@@ -274,29 +451,36 @@ class BadProducer(Producer):  # type: ignore # noqa: F811
             a1: A1 = A1()
 
             @classmethod
-            def build(cls, a1: A1) -> A2:
+            def build(cls, a1: dict) -> A2:  # type: ignore
+                pass
+
+    with pytest.raises(ValueError, match=r"str.* cannot be used to represent Struct"):
+
+        class BadProducer(Producer):  # type: ignore # noqa: F811
+            @classmethod
+            def build(cls) -> Annotated[str, A2]:
                 pass
 
     with pytest.raises(
         ValueError,
-        match=r"BadProducer.build - must be a @classmethod or @staticmethod.",
+        match=r"BadProducer.build - must be a @classmethod or @staticmethod",
     ):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
-            def build(cls) -> A2:
+            def build(cls) -> Annotated[dict, A2]:  # type: ignore
                 pass
 
     with pytest.raises(
         ValueError,
-        match=r"BadProducer.map - must be a @classmethod or @staticmethod.",
+        match=r"BadProducer.map - must be a @classmethod or @staticmethod",
     ):
 
         class BadProducer(Producer):  # type: ignore # noqa: F811
             @classmethod
-            def build(cls) -> A2:
+            def build(cls) -> Annotated[dict, A2]:  # type: ignore
                 pass
 
-            def map(cls) -> A2:
+            def map(cls) -> PartitionDependencies:
                 pass
 
 
@@ -307,19 +491,23 @@ def test_Producer_bad_init() -> None:
         DummyProducer(junk=5)
     with pytest.raises(ValueError, match="field required"):
         DummyProducer()
-    with pytest.raises(ValueError, match="Expected an instance of"):
+    with pytest.raises(ValueError, match="expected an instance of"):
         DummyProducer(a1=5)
-    with pytest.raises(ValueError, match="Expected an instance of"):
+    with pytest.raises(ValueError, match="expected an instance of"):
         DummyProducer(a1=A2())
 
 
 def test_Producer_bad_out() -> None:
     producer = DummyProducer(a1=A1())
-    with pytest.raises(ValueError, match="Expected 2 arguments of"):
+    with pytest.raises(ValueError, match="expected 2 arguments of"):
         producer.out(1)  # type: ignore
-    with pytest.raises(ValueError, match="Expected the 1st argument to be"):
+    with pytest.raises(
+        ValueError, match=r"DummyProducer.out\(\) 1st argument - expected instance of"
+    ):
         producer.out(1, 2)  # type: ignore
-    with pytest.raises(ValueError, match="Expected the 2nd argument to be"):
+    with pytest.raises(
+        ValueError, match=r"DummyProducer.out\(\) 2nd argument - expected instance of"
+    ):
         producer.out(A2(), A2())
     output = producer.out(A2(), A3())
     with pytest.raises(ValueError, match="is produced by"):
diff --git a/tests/arti/storage/test_local_storage.py b/tests/arti/storage/test_local_storage.py
index 6078ea4b..74767490 100644
--- a/tests/arti/storage/test_local_storage.py
+++ b/tests/arti/storage/test_local_storage.py
@@ -2,27 +2,30 @@
 import re
 from contextlib import contextmanager
 from datetime import date
+from itertools import repeat
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Iterator
+from typing import Any, Iterator
 
 import pytest
 
 from arti.fingerprints import Fingerprint
-from arti.partitions import DateKey
+from arti.partitions import CompositeKey, CompositeKeyTypes, DateKey
+from arti.storage import InputFingerprints, Storage
 from arti.storage.local import LocalFile, LocalFilePartition
 
 
 @pytest.fixture()
-def date_keys() -> frozenset[DateKey]:
-    return frozenset(
-        {
+def date_keys() -> list[CompositeKey]:
+    return [
+        CompositeKey(date=date_key)
+        for date_key in (
             DateKey(key=date(1970, 1, 1)),
             DateKey(key=date(1970, 1, 2)),
             DateKey(key=date(1970, 1, 3)),
             DateKey(key=date(2021, 1, 1)),
-        }
-    )
+        )
+    ]
 
 
 @contextmanager
@@ -31,59 +34,82 @@ def tmp_dir() -> Iterator[Path]:
         yield Path(_tmpdir)
 
 
-@contextmanager
-def tmp_date_files(date_partitions: frozenset[DateKey], filename: str) -> Iterator[Path]:
+def generate_partition_files(storage: Storage[Any], input_fingerprints: InputFingerprints) -> None:
+    for pk, input_fingerprint in input_fingerprints.items():
+        partition = storage.generate_partition(
+            pk, input_fingerprint, with_content_fingerprint=False
+        )
+        partition_path = Path(partition.path)
+        partition_path.parent.mkdir(parents=True)
+        partition_path.touch()
+
+
+def test_local_partitioning(date_keys: list[CompositeKey]) -> None:
     with tmp_dir() as tmpdir:
-        for pk in date_partitions:
-            path = tmpdir / str(pk.Y) / f"{pk.m:02}" / f"{pk.d:02}"
-            path.mkdir(parents=True)
-            (path / filename).touch()
-        yield tmpdir
-
-
-def test_local_partitioning(date_keys: frozenset[DateKey]) -> None:
-    with tmp_date_files(date_keys, "test") as tmpdir:
-        partitions = LocalFile(
-            path=str(tmpdir / "{data_date.Y}" / "{data_date.m}" / "{data_date.d}" / "test")
-        ).discover_partitions(data_date=DateKey)
+        storage = LocalFile(path=str(tmpdir / "{date.Y}" / "{date.m}" / "{date.d}" / "test"))
+        generate_partition_files(
+            storage, InputFingerprints(zip(date_keys, repeat(Fingerprint.empty())))
+        )
+        partitions = storage.discover_partitions(CompositeKeyTypes(date=DateKey))
+        assert len(partitions) > 0
         for partition in partitions:
             assert isinstance(partition, LocalFilePartition)
-            assert set(partition.keys) == {"data_date"}
-            assert partition.keys["data_date"] in date_keys
-            assert isinstance(partition.keys["data_date"], DateKey)
+            assert set(partition.keys) == {"date"}
+            assert partition.keys in date_keys
+            assert isinstance(partition.keys["date"], DateKey)
 
 
-def test_local_partitioning_filtered(date_keys: frozenset[DateKey]) -> None:
-    with tmp_date_files(date_keys, "test") as tmpdir:
-        for year in {k.Y for k in date_keys}:
-            partitions = LocalFile(
+def test_local_partitioning_filtered(date_keys: list[CompositeKey]) -> None:
+    for year in {k["date"].Y for k in date_keys}:  # type: ignore
+        with tmp_dir() as tmpdir:
+            storage = LocalFile(path=str(tmpdir / "{date.Y}" / "{date.m}" / "{date.d}" / "test"))
+            storage = LocalFile(
                 path=str(
-                    tmpdir
-                    / ("{data_date.Y[" + str(year) + "]}")
-                    / "{data_date.m}"
-                    / "{data_date.d}"
-                    / "test"
+                    tmpdir / ("{date.Y[" + str(year) + "]}") / "{date.m}" / "{date.d}" / "test"
                 )
-            ).discover_partitions(data_date=DateKey)
+            )
+            # Generate files for *all* years - we want discover_partitions to do the filtering.
+            generate_partition_files(
+                storage, InputFingerprints(zip(date_keys, repeat(Fingerprint.empty())))
+            )
+            partitions = storage.discover_partitions(CompositeKeyTypes(date=DateKey))
+            assert len(partitions) > 0
             for partition in partitions:
                 assert isinstance(partition, LocalFilePartition)
-                assert set(partition.keys) == {"data_date"}
-                assert partition.keys["data_date"] in date_keys
-                assert isinstance(partition.keys["data_date"], DateKey)
-                assert partition.keys["data_date"].Y == year
+                assert set(partition.keys) == {"date"}
+                assert partition.keys in date_keys
+                assert isinstance(partition.keys["date"], DateKey)
+                assert partition.keys["date"].Y == year
+
+
+def test_local_partitioning_with_input_fingerprints(date_keys: list[CompositeKey]) -> None:
+    with tmp_dir() as tmpdir:
+        storage = LocalFile(
+            path=str(tmpdir / "{date.Y}" / "{date.m}" / "{date.d}" / "{input_fingerprint}" / "test")
+        )
+        input_fingerprint = Fingerprint.from_int(42)
+        input_fingerprints = InputFingerprints(zip(date_keys, repeat(input_fingerprint)))
+        generate_partition_files(storage, input_fingerprints)
+        partitions = storage.discover_partitions(
+            CompositeKeyTypes(date=DateKey), input_fingerprints
+        )
+        assert len(partitions) > 0
+        for partition in partitions:
+            assert isinstance(partition, LocalFilePartition)
+            assert set(partition.keys) == {"date"}
+            assert partition.keys in date_keys
+            assert isinstance(partition.keys["date"], DateKey)
+            assert partition.input_fingerprint == input_fingerprint
 
 
-def test_local_partitioning_errors(date_keys: frozenset[DateKey]) -> None:
-    with tmp_date_files(date_keys, "test") as tmpdir:
+def test_local_partitioning_errors() -> None:
+    with tmp_dir() as tmpdir:
+        storage = LocalFile(path=str(tmpdir / "{date.Y}" / "{date.m}" / "{date.d}" / "test"))
         with pytest.raises(
             ValueError,
-            match=re.escape("No 'data_date' partition key found, expected one of ('date',)"),
+            match=re.escape("No 'date' partition key found, expected one of ('data_date',)"),
         ):
-            LocalFile(
-                path=str(
-                    tmpdir / "{data_date.Y[2021]}" / "{data_date.m}" / "{data_date.d}" / "test"
-                )
-            ).discover_partitions(date=DateKey)
+            storage.discover_partitions(CompositeKeyTypes(data_date=DateKey))
 
 
 def test_local_file_partition_fingerprint() -> None:
@@ -93,7 +119,7 @@ def test_local_file_partition_fingerprint() -> None:
         with path.open("w") as f:
             f.write("hello world")
 
-        partition = LocalFilePartition(keys={}, path=str(path)).with_fingerprint()
-        assert partition.fingerprint == Fingerprint.from_string(
+        partition = LocalFilePartition(keys={}, path=str(path)).with_content_fingerprint()
+        assert partition.content_fingerprint == Fingerprint.from_string(
             hashlib.sha1(text.encode()).hexdigest()
         )
diff --git a/tests/arti/storage/test_storage.py b/tests/arti/storage/test_storage.py
index 320cc3c7..7109eefe 100644
--- a/tests/arti/storage/test_storage.py
+++ b/tests/arti/storage/test_storage.py
@@ -1,14 +1,19 @@
+from __future__ import annotations
+
+from typing import Optional
+
 import pytest
 
 from arti.fingerprints import Fingerprint
-from arti.partitions import Int8Key, PartitionKey
-from arti.storage import Storage, StoragePartition
+from arti.internal.utils import frozendict
+from arti.partitions import CompositeKey, CompositeKeyTypes, DateKey, Int8Key
+from arti.storage import InputFingerprints, Storage, StoragePartition
 
 
 class MockStoragePartition(StoragePartition):
     path: str
 
-    def compute_fingerprint(self) -> Fingerprint:
+    def compute_content_fingerprint(self) -> Fingerprint:
         return Fingerprint.from_string(self.path)
 
 
@@ -16,7 +21,7 @@ class MockStorage(Storage[MockStoragePartition]):
     path: str
 
     def discover_partitions(
-        self, **key_types: type[PartitionKey]
+        self, key_types: CompositeKeyTypes, input_fingerprints: Optional[InputFingerprints] = None
     ) -> tuple[MockStoragePartition, ...]:
         assert all(v is Int8Key for v in key_types.values())  # Simplifies logic here...
         return tuple(
@@ -25,12 +30,15 @@ def discover_partitions(
         )
 
 
-def test_StoragePartition_fingerprint() -> None:
-    sp = MockStoragePartition(path="/tmp/test", keys={"key": Int8Key(key=5)})
-    assert sp.fingerprint is None
-    modified = sp.with_fingerprint()
-    assert sp.fingerprint is None
-    assert modified.fingerprint == Fingerprint.from_string(sp.path)
+def test_StoragePartition_content_fingerprint() -> None:
+    sp = MockStoragePartition(path="/tmp/test", keys={})
+    assert sp.content_fingerprint == Fingerprint.empty()
+    populated = sp.with_content_fingerprint()
+    assert sp.content_fingerprint == Fingerprint.empty()
+    assert populated.content_fingerprint == Fingerprint.from_string(sp.path)
+    assert populated.with_content_fingerprint(keep_existing=True) is populated
+    assert populated.with_content_fingerprint(keep_existing=False) == populated
+    assert populated.with_content_fingerprint(keep_existing=False) is not populated
 
 
 def test_Storage_init_subclass() -> None:
@@ -56,10 +64,141 @@ class S(Storage[MockStoragePartition]):
     assert S.storage_partition_type is MockStoragePartition  # type: ignore
 
 
+def test_Storage_resolve_extension() -> None:
+    s = MockStorage(path="{extension}")
+    assert s.resolve_extension("test") == MockStorage(path="test")
+    assert s.resolve_extension(None) == MockStorage(path="")
+
+
+def test_Storage_resolve_graph_name() -> None:
+    s = MockStorage(path="/{graph_name}/junk")
+    assert s.resolve_graph_name("test") == MockStorage(path="/test/junk")
+    # Confirm we strip trailing (ie: dup) separators when resolving an empty placeholder.
+    assert s.resolve_graph_name("") == MockStorage(path="/junk")
+
+
+def test_Storage_resolve_input_fingerprint() -> None:
+    s = MockStorage(path="/{input_fingerprint}/junk")
+    assert s.resolve_input_fingerprint(Fingerprint.from_int(10)) == MockStorage(path="/10/junk")
+    assert s.resolve_input_fingerprint(Fingerprint.empty()) == MockStorage(path="/junk")
+
+
+def test_Storage_resolve_names() -> None:
+    s = MockStorage(path="/{names}")
+    assert s.resolve_names(("a", "b")) == MockStorage(path="/a/b")
+    assert s.resolve_names(()) == MockStorage(path="/")
+
+    s = MockStorage(path="/{names}/junk/{name}")
+    assert s.resolve_names(("a", "b")) == MockStorage(path="/a/b/junk/b")
+    assert s.resolve_names(()) == MockStorage(path="/junk/")
+
+
+@pytest.mark.parametrize(
+    ("spec", "expected", "key_types"),
+    (
+        (
+            "/tmp/test/{partition_key_spec}",
+            "/tmp/test/",
+            {},
+        ),
+        (
+            "/tmp/test/{partition_key_spec}",
+            "/tmp/test/date_Y={date.Y}/date_m={date.m:02}/date_d={date.d:02}",
+            {"date": DateKey},
+        ),
+        (
+            "/tmp/test/{partition_key_spec}",
+            "/tmp/test/a_key={a.key}/b_key={b.key}",
+            {"a": Int8Key, "b": Int8Key},
+        ),
+        (
+            "/tmp/test/{tag}/{partition_key_spec}",
+            "/tmp/test/{tag}/a_key={a.key}",
+            {"a": Int8Key},
+        ),
+    ),
+)
+def test_Storage_resolve_partition_key_spec(
+    spec: str, expected: str, key_types: CompositeKeyTypes
+) -> None:
+    assert MockStorage(path=spec).resolve_partition_key_spec(key_types).path == expected
+
+
+def test_Storage_resolve_partition_key_spec_extra() -> None:
+    class TablePartition(StoragePartition):
+        dataset: str
+        name: str
+
+        def compute_content_fingerprint(self) -> Fingerprint:
+            return Fingerprint.empty()
+
+    class Table(Storage[TablePartition]):
+        key_value_sep = "_"
+        partition_name_component_sep = "_"
+        segment_sep = "__"
+
+        dataset: str = "s_{tag}"
+        name: str = "{partition_key_spec}"
+
+        def discover_partitions(
+            self,
+            key_types: CompositeKeyTypes,
+            input_fingerprints: Optional[InputFingerprints] = None,
+        ) -> tuple[TablePartition, ...]:
+            return ()
+
+    t = Table().resolve_partition_key_spec(CompositeKeyTypes(a=Int8Key, b=Int8Key))
+    assert t.dataset == "s_{tag}"
+    assert t.name == "a_key_{a.key}__b_key_{b.key}"
+
+
+def test_Storage_resolve_path_tags() -> None:
+    s = MockStorage(path="{path_tags}")
+    assert s.resolve_path_tags(frozendict(a="b")) == MockStorage(path="a=b")
+    assert s.resolve_path_tags(frozendict()) == MockStorage(path="")
+
+
 def test_Storage_discover_partitions() -> None:
     s = MockStorage(path="/test/{i.key}/file")
-    partitions = s.discover_partitions(i=Int8Key)
+    partitions = s.discover_partitions(CompositeKeyTypes(i=Int8Key))
     for i, sp in enumerate(sorted(partitions, key=lambda x: x.path)):
         assert sp.path == f"/test/{i}/file"
         assert isinstance(sp.keys["i"], Int8Key)
         assert sp.keys["i"].key == i
+
+
+def test_Storage_generate_partition() -> None:
+    keys = CompositeKey(i=Int8Key(key=5))
+    input_fingerprint = Fingerprint.from_int(10)
+    s = MockStorage(path="{i.key:02}/{input_fingerprint}")
+    expected_partition = MockStoragePartition(
+        input_fingerprint=input_fingerprint,
+        keys=keys,
+        path="05/10",
+    )
+
+    output = s.generate_partition(keys=keys, input_fingerprint=input_fingerprint)
+    assert output == expected_partition.with_content_fingerprint()
+
+    output = s.generate_partition(
+        keys=keys, input_fingerprint=input_fingerprint, with_content_fingerprint=True
+    )
+    assert output == expected_partition.with_content_fingerprint()
+
+    output = s.generate_partition(
+        keys=keys, input_fingerprint=input_fingerprint, with_content_fingerprint=False
+    )
+    assert output == expected_partition
+
+    # Check behavior when the Storage spec doesn't align with the passed in keys/fingerprint. We'll
+    # probably want nicer error messages for these eventually.
+    with pytest.raises(KeyError, match="i"):
+        s.generate_partition(
+            keys=CompositeKey(j=Int8Key(key=5)), input_fingerprint=input_fingerprint
+        )
+    with pytest.raises(ValueError, match="requires an input_fingerprint, but none was provided"):
+        s.generate_partition(keys=keys, input_fingerprint=Fingerprint.empty())
+    with pytest.raises(ValueError, match="does not specify a {input_fingerprint} template"):
+        MockStorage(path="hard coded").generate_partition(
+            keys=keys, input_fingerprint=Fingerprint.from_string("fake")
+        )
diff --git a/tests/arti/storage/test_storage_internal.py b/tests/arti/storage/test_storage_internal.py
index a1a3b911..1aa2bd2f 100644
--- a/tests/arti/storage/test_storage_internal.py
+++ b/tests/arti/storage/test_storage_internal.py
@@ -1,18 +1,25 @@
 import re
+from typing import Optional
 
 import parse
 import pytest
 
+from arti.fingerprints import Fingerprint
 from arti.internal.utils import frozendict
-from arti.partitions import CompositeKey, Int8Key, PartitionKey
+from arti.partitions import CompositeKey, CompositeKeyTypes, Int8Key, PartitionKey
 from arti.storage._internal import (
-    WildcardFormatDict,
+    FormatDict,
+    FormatPlaceholder,
     WildcardPlaceholder,
-    extract_partition_keys,
-    parse_partition_keys,
+    extract_placeholders,
+    parse_spec,
+    partial_format,
     spec_to_wildcard,
+    strip_partition_indexes,
 )
 
+PathPlaceholders = dict[str, tuple[Optional[Fingerprint], CompositeKey]]
+
 
 @pytest.fixture
 def PKs() -> dict[str, type[PartitionKey]]:
@@ -30,28 +37,52 @@ def paths() -> set[str]:
 
 
 @pytest.fixture
-def paths_to_keys() -> dict[str, CompositeKey]:
+def paths_to_placeholders() -> PathPlaceholders:
     return {
-        "/p/1/0x1": frozendict({"x": Int8Key(key=1), "y": Int8Key(key=1)}),
-        "/p/2/0x2": frozendict({"x": Int8Key(key=2), "y": Int8Key(key=2)}),
-        "/p/3/0x3": frozendict({"x": Int8Key(key=3), "y": Int8Key(key=3)}),
+        "/p/1/0x1": (Fingerprint.empty(), frozendict({"x": Int8Key(key=1), "y": Int8Key(key=1)})),
+        "/p/2/0x2": (Fingerprint.empty(), frozendict({"x": Int8Key(key=2), "y": Int8Key(key=2)})),
+        "/p/3/0x3": (Fingerprint.empty(), frozendict({"x": Int8Key(key=3), "y": Int8Key(key=3)})),
     }
 
 
 @pytest.mark.parametrize(
-    ("partition_key_type", "attribute", "valid_attributes"),
+    ("spec", "key"),
+    (
+        ("baseline", None),
+        ("{}", None),
+        ("{a}", "a"),
+        ("{a.5}", "a"),
+        ("{a[1]}", "a"),
+        ("{a[b]}", "a"),
+        ("{a[b].c}", "a"),
+        ("{a[b].c[d]}", "a"),
+        ("{a[b].c[d]:2d}", "a"),
+        ("hello {world}", "world"),
+    ),
+)
+def test_FormatPlaceholder(spec: str, key: str) -> None:
+    if key:
+        out = spec.format(**{key: FormatPlaceholder(key)})
+    else:
+        out = spec.format(FormatPlaceholder(""))
+    assert out == spec
+
+
+@pytest.mark.parametrize(
+    ("key", "partition_key_types", "attribute"),
     (
-        (Int8Key, "key", {"hex", "key"}),
-        (Int8Key, "hex", {"hex", "key"}),
+        ("test", CompositeKeyTypes(test=Int8Key), "key"),
+        ("test", CompositeKeyTypes(test=Int8Key), "hex"),
     ),
 )
 def test_WildcardPlaceholder(
-    partition_key_type: type[PartitionKey], attribute: str, valid_attributes: set[str]
+    key: str, partition_key_types: CompositeKeyTypes, attribute: str
 ) -> None:
-    placeholder = WildcardPlaceholder(name="test", key_type=partition_key_type)
-    assert placeholder._name == "test"
+    partition_key_type = partition_key_types[key]
+
+    placeholder = WildcardPlaceholder(key, partition_key_types)
+    assert placeholder._key == key
     assert placeholder._key_type == partition_key_type
-    assert placeholder._valid_attributes == valid_attributes
     assert placeholder._attribute is None
     assert getattr(placeholder, attribute) is placeholder
     assert placeholder._attribute == attribute
@@ -60,38 +91,43 @@ def test_WildcardPlaceholder(
     assert placeholder[5] == 5
     assert placeholder[10] == 10
 
-    placeholder = WildcardPlaceholder(name="test", key_type=partition_key_type)
+    partial = WildcardPlaceholder.with_key_types(partition_key_types)
+    applied = partial(key)
+    assert applied._key == key
+    assert applied._key_type == partition_key_type
+
+    placeholder = WildcardPlaceholder(key, partition_key_types)
     getattr(placeholder, attribute)
     with pytest.raises(
-        ValueError, match=f"'test.{attribute}.{attribute}' cannot be used in a partition path"
+        ValueError, match=f"'{key}.{attribute}.{attribute}' cannot be used in a partition path"
     ):
         getattr(placeholder, attribute)
 
-    placeholder = WildcardPlaceholder(name="test", key_type=partition_key_type)
+    placeholder = WildcardPlaceholder(key, partition_key_types)
     with pytest.raises(
         AttributeError,
         match=f"'{partition_key_type.__name__}' has no field or key component 'abc123'",
     ):
         placeholder.abc123
 
-    placeholder = WildcardPlaceholder(name="test", key_type=partition_key_type)
+    placeholder = WildcardPlaceholder(key, partition_key_types)
     with pytest.raises(
         ValueError,
-        match="'test' cannot be used directly in a partition path; access one of the key components",
+        match=f"'{key}' cannot be used directly in a partition path; access one of the key components",
     ):
         str(placeholder)
     with pytest.raises(
         ValueError,
-        match="'test' cannot be used directly in a partition path; access one of the key components",
+        match=f"'{key}' cannot be used directly in a partition path; access one of the key components",
     ):
         placeholder[5]
 
 
-def test_WildcardFormatDict() -> None:
-    d = WildcardFormatDict({"test": Int8Key}, tag="x")
+def test_FormatDict() -> None:
+    d = FormatDict(WildcardPlaceholder.with_key_types({"test": Int8Key}), tag="x")
     test_placeholder = d["test"]
     assert isinstance(test_placeholder, WildcardPlaceholder)
-    assert test_placeholder._name == "test"
+    assert test_placeholder._key == "test"
     assert test_placeholder._key_type == Int8Key
     assert d["tag"] == "x"
 
@@ -101,65 +137,98 @@ def test_WildcardFormatDict() -> None:
         d["junk"]
 
 
+@pytest.mark.parametrize(
+    ("spec", "expected", "kwargs"),
+    (
+        ("baseline", "baseline", {}),
+        ("{a}", "a", {"a": "a"}),
+        ("{a}", "{a}", {}),
+        ("{hello} {world}", "hello {world}", {"hello": "hello"}),
+        ("{hello} {wo[rld]}", "hello {wo[rld]}", {"hello": "hello"}),
+    ),
+)
+def test_partial_format(spec: str, expected: str, kwargs: dict[str, str]) -> None:
+    assert partial_format(spec, **kwargs) == expected
+
+
+# To support positional args, we would need to introduce a FormatSequence (w/
+# __getitem__) similar to FormatDict. Easy enough, but don't need (yet).
+@pytest.mark.xfail(raises=IndexError)
+def test_partial_format_positional_args() -> None:
+    partial_format("{}")
+
+
+@pytest.mark.parametrize(
+    ("spec", "expected"),
+    (
+        ("baseline", "baseline"),
+        ("{a}", "{a}"),
+        ("{a[b]}", "{a}"),
+    ),
+)
+def test_strip_partition_indexes(spec: str, expected: str) -> None:
+    assert strip_partition_indexes(spec) == expected
+
+
 def test_spec_to_wildcard(PKs: dict[str, type[PartitionKey]]) -> None:
     assert spec_to_wildcard("/p/{x.key}/", PKs) == "/p/*/"
     assert spec_to_wildcard("/p/{x.key[5]}/", PKs) == "/p/5/"
     assert spec_to_wildcard("/p/{x.key[5]}/{y.hex}/", PKs) == "/p/5/*/"
 
 
-def test_extract_partition_keys(
+def test_extract_placeholders(
     PKs: dict[str, type[PartitionKey]],
     spec: str,
     paths: set[str],
-    paths_to_keys: dict[str, CompositeKey],
+    paths_to_placeholders: PathPlaceholders,
 ) -> None:
     parser = parse.compile(spec, case_sensitive=True)
 
     assert {
-        path: extract_partition_keys(
+        path: extract_placeholders(
             key_types=PKs,
             parser=parser,
             path=path,
             spec=spec,
         )
         for path in paths
-    } == paths_to_keys
+    } == paths_to_placeholders
     with pytest.raises(
         ValueError, match=re.escape("Unable to parse 'fake' with '/p/{x.key}/{y.hex}'.")
     ):
-        extract_partition_keys(
+        extract_placeholders(
             key_types=PKs,
             parser=parser,
             path="fake",
             spec=spec,
         )
     assert (
-        extract_partition_keys(
+        extract_placeholders(
             error_on_no_match=False,
             key_types=PKs,
             parser=parser,
             path="fake",
             spec=spec,
         )
-        is None
+        == (Fingerprint.empty(), CompositeKey())
     )
 
 
-def test_parse_partition_keys(
+def test_parse_spec(
     PKs: dict[str, type[PartitionKey]],
     spec: str,
     paths: set[str],
-    paths_to_keys: dict[str, CompositeKey],
+    paths_to_placeholders: PathPlaceholders,
 ) -> None:
-    pks = parse_partition_keys(paths, spec=spec, key_types=PKs)
-    assert pks == paths_to_keys
+    pks = parse_spec(paths, key_types=PKs, spec=spec)
+    assert pks == paths_to_placeholders
 
     with pytest.raises(
         ValueError, match=re.escape("Unable to parse '/p/1/' with '/p/{x.key}/{y.hex}'")
     ):
-        parse_partition_keys({"/p/1/"}, spec="/p/{x.key}/{y.hex}", key_types=PKs)
+        parse_spec({"/p/1/"}, spec="/p/{x.key}/{y.hex}", key_types=PKs)
     with pytest.raises(
         ValueError,
         match=re.escape("Expected to find partition keys for ['x', 'y'], only found ['x']."),
     ):
-        parse_partition_keys({"/p/1/"}, spec="/p/{x.key}/", key_types=PKs)
+        parse_spec({"/p/1/"}, spec="/p/{x.key}/", key_types=PKs)
diff --git a/tests/arti/test_fingerprints.py b/tests/arti/test_fingerprints.py
index 5c40cf8d..aa420fc6 100644
--- a/tests/arti/test_fingerprints.py
+++ b/tests/arti/test_fingerprints.py
@@ -47,7 +47,13 @@ def test_Fingerprint_math() -> None:
     # empty cascades
     assert Fingerprint.empty().combine(f1).is_empty
     assert f1.combine(Fingerprint.empty()).is_empty
-
-    for val in [None, 0, "a"]:
+    # bitwise operators
+    assert f2 & 15 == 1
+    assert f2 << 15 == 32768
+    assert f2 >> 15 == 0
+    assert f2 ^ 15 == 14
+    assert f2 | 15 == 15
+
+    for val in [None, "a"]:
         with pytest.raises(TypeError):
             f1.combine(val)  # type: ignore
diff --git a/tests/arti/test_thresholds.py b/tests/arti/test_thresholds.py
new file mode 100644
index 00000000..82ac8f46
--- /dev/null
+++ b/tests/arti/test_thresholds.py
@@ -0,0 +1,5 @@
+from arti.thresholds import Threshold
+
+
+def test_Threshold() -> None:
+    assert Threshold  # NOTE: Nothing here yet...
diff --git a/tests/arti/types/test_pydantic_adapters.py b/tests/arti/types/test_pydantic_adapters.py
index 59af70df..330eb23a 100644
--- a/tests/arti/types/test_pydantic_adapters.py
+++ b/tests/arti/types/test_pydantic_adapters.py
@@ -29,7 +29,7 @@ def test_pydantic_conversion() -> None:
     assert isinstance(arti_type.fields["x"], _Int)
     assert isinstance(arti_type.fields["y"], String)
     assert isinstance(arti_type.fields["tags"], List)
-    assert isinstance(arti_type.fields["tags"].value_type, String)
+    assert isinstance(arti_type.fields["tags"].element, String)
 
 
 def test_pydantic_field_naming() -> None:
@@ -90,7 +90,7 @@ def compare_model_to_type(model: type[BaseModel], generated: Type) -> None:  # n
             expected_args = get_args(expected_type)
             if lenient_issubclass(expected_origin, Mapping):
                 assert isinstance(spec, Map)
-                for (sub_type, sub_spec) in zip(expected_args, (spec.key_type, spec.value_type)):
+                for (sub_type, sub_spec) in zip(expected_args, (spec.key, spec.value)):
                     if lenient_issubclass(sub_type, BaseModel):
                         compare_model_to_type(sub_type, sub_spec)
                     else:
@@ -104,7 +104,7 @@ def compare_model_to_type(model: type[BaseModel], generated: Type) -> None:  # n
                     assert len(expected_args) == 2
                     assert expected_args[1] == ...
                 assert isinstance(spec, List)
-                sub_type, sub_spec = expected_args[0], spec.value_type
+                sub_type, sub_spec = expected_args[0], spec.element
                 if lenient_issubclass(sub_type, BaseModel):
                     compare_model_to_type(sub_type, sub_spec)
                 else:
diff --git a/tests/arti/types/test_python_adapters.py b/tests/arti/types/test_python_adapters.py
index 098d24f6..23a9bee9 100644
--- a/tests/arti/types/test_python_adapters.py
+++ b/tests/arti/types/test_python_adapters.py
@@ -59,7 +59,7 @@ def test_python_datetime() -> None:
 
 
 def test_python_frozenset() -> None:
-    a = Set(value_type=Int64())
+    a = Set(element=Int64())
     p = frozenset[int]
 
     assert python_type_system.to_system(a, hints={}) == set[int]  # set has higher priority
@@ -67,7 +67,7 @@ def test_python_frozenset() -> None:
 
 
 def test_python_list() -> None:
-    a = List(value_type=Int64())
+    a = List(element=Int64())
     p = list[int]
 
     assert python_type_system.to_system(a, hints={}) == p
@@ -111,7 +111,7 @@ def test_python_literal_errors() -> None:
 
 
 def test_python_map() -> None:
-    a = Map(key_type=String(), value_type=Int64())
+    a = Map(key=String(), value=Int64())
     p = dict[str, int]
 
     assert python_type_system.to_system(a, hints={}) == p
@@ -147,7 +147,7 @@ def test_python_optional_priority() -> None:
 
 
 def test_python_set() -> None:
-    a = Set(value_type=Int64())
+    a = Set(element=Int64())
     p = set[int]
 
     assert python_type_system.to_system(a, hints={}) == p
@@ -168,12 +168,12 @@ class P(TypedDict):
 
 
 def test_python_tuple() -> None:
-    a = List(value_type=Int64())
+    a = List(element=Int64())
     # mypy errors here for some reason...?
     p = tuple[int, ...]  # type: ignore
 
     assert python_type_system.to_system(a, hints={}) == list[int]  # list has higher priority
-    assert PyTuple.to_system(List(value_type=Int64()), hints={}) == p
+    assert PyTuple.to_system(List(element=Int64()), hints={}) == p
     assert python_type_system.to_artigraph(p, hints={}) == a
 
     # We don't (currently) support structure based tuples
diff --git a/tests/arti/types/test_sgqlc_adapters.py b/tests/arti/types/test_sgqlc_adapters.py
index 4fcd3091..ae0df414 100644
--- a/tests/arti/types/test_sgqlc_adapters.py
+++ b/tests/arti/types/test_sgqlc_adapters.py
@@ -123,7 +123,7 @@ def test_nested_sgqlc_type() -> None:
     assert isinstance(a.fields["id"], at._Int)
     assert not a.fields["id"].nullable
     assert isinstance(a.fields["value"], at.List)
-    assert isinstance(a.fields["value"].value_type, at.Struct)
+    assert isinstance(a.fields["value"].element, at.Struct)
     # root sgqlc
     assert s.__name__ == MySuperType.__name__
     assert is_sgqlc_nullable(s)
@@ -137,7 +137,7 @@ def test_nested_sgqlc_type() -> None:
     a_container = a.fields["value"]
     assert isinstance(a_container, at.List)
     assert a_container.nullable
-    a_value = a_container.value_type
+    a_value = a_container.element
     assert isinstance(a_value, at.Struct)
     assert a_value.name == MyType.__name__
     assert not a_value.nullable
@@ -183,7 +183,7 @@ def test_sgqlc_list_adapter() -> None:
 
     arti_t = sgqlc_type_system.to_artigraph(base, hints={})
     assert isinstance(arti_t, at.List)
-    assert isinstance(arti_t.value_type, at.Int64)
+    assert isinstance(arti_t.element, at.Int64)
 
     sgqlc_t = sgqlc_type_system.to_system(arti_t, hints={})
     assert issubclass(sgqlc_t, st.Int)
@@ -209,8 +209,8 @@ def test_sgqlc_list_of_non_null_adapting() -> None:
     arti_t = sgqlc_type_system.to_artigraph(base, hints={})
     assert isinstance(arti_t, at.List)
     assert arti_t.nullable
-    assert isinstance(arti_t.value_type, at.Enum)
-    assert not arti_t.value_type.nullable
+    assert isinstance(arti_t.element, at.Enum)
+    assert not arti_t.element.nullable
 
     sgqlc_t = sgqlc_type_system.to_system(arti_t, hints={})
     assert issubclass(sgqlc_t, st.Enum)
@@ -225,8 +225,8 @@ def test_sgqlc_non_null_list_of_adapting() -> None:
     arti_t = sgqlc_type_system.to_artigraph(base, hints={})
     assert isinstance(arti_t, at.List)
     assert not arti_t.nullable
-    assert isinstance(arti_t.value_type, at.Enum)
-    assert arti_t.value_type.nullable
+    assert isinstance(arti_t.element, at.Enum)
+    assert arti_t.element.nullable
 
     sgqlc_t = sgqlc_type_system.to_system(arti_t, hints={})
     assert issubclass(sgqlc_t, st.Enum)
@@ -241,8 +241,8 @@ def test_sgqlc_non_null_list_of_non_null_adapting() -> None:
     arti_t = sgqlc_type_system.to_artigraph(base, hints={})
     assert isinstance(arti_t, at.List)
     assert not arti_t.nullable
-    assert isinstance(arti_t.value_type, at.Enum)
-    assert not arti_t.value_type.nullable
+    assert isinstance(arti_t.element, at.Enum)
+    assert not arti_t.element.nullable
 
     sgqlc_t = sgqlc_type_system.to_system(arti_t, hints={})
     assert issubclass(sgqlc_t, st.Enum)
diff --git a/tests/arti/types/test_types.py b/tests/arti/types/test_types.py
index 2b595e0c..ba71af4e 100644
--- a/tests/arti/types/test_types.py
+++ b/tests/arti/types/test_types.py
@@ -11,6 +11,7 @@
     Float32,
     Float64,
     Int32,
+    List,
     Struct,
     Timestamp,
     Type,
@@ -102,7 +103,7 @@ def test_Enum_errors() -> None:
     with pytest.raises(ValueError, match="cannot be empty"):
         Enum(type=float32, items=[])
 
-    with pytest.raises(ValueError, match="Expected an instance of"):
+    with pytest.raises(ValueError, match="expected an instance of"):
         Enum(type=float32, items={v: v for v in items})
 
     mismatch_prefix = "incompatible Float32() (<class 'float'>) item(s)"
@@ -111,65 +112,73 @@ def test_Enum_errors() -> None:
     with pytest.raises(ValueError, match=re.escape(f"{mismatch_prefix}: [3]")):
         Enum(type=float32, items=(1.0, 2.0, 3))
 
-    with pytest.raises(ValueError, match="Expected an instance of <class 'arti.types.Type'>"):
+    with pytest.raises(ValueError, match="expected an instance of <class 'arti.types.Type'>"):
         # NOTE: using a python type instead of an Artigraph type
         Enum(type=float, items=items)
 
 
-def test_Struct() -> None:
+def test_List() -> None:
+    lst = List(element=Int32())
+    assert lst.element == Int32()
+    assert lst.partition_by == ()
+    assert lst.cluster_by == ()
+    assert lst.partition_fields == frozendict()
+    assert lst.is_partitioned is False
+
+
+def test_List_partitioned() -> None:
     fields = {"x": Int32(), "y": Int32()}
-    s = Struct(fields=fields)
-    assert isinstance(s.fields, frozendict)
-    assert s.fields == frozendict(fields)
 
-    s = Struct(fields=fields, partition_by=("x",), cluster_by=("y",))
-    assert s.partition_by == ("x",)
-    assert s.cluster_by == ("y",)
-    assert s.partition_fields == frozendict({"x": fields["x"]})
+    lst = List(element=Struct(fields=fields), partition_by=("x",), cluster_by=("y",))
+    assert lst.partition_by == ("x",)
+    assert lst.cluster_by == ("y",)
+    assert lst.partition_fields == frozendict({"x": fields["x"]})
+    assert lst.is_partitioned is True
+
+    with pytest.raises(ValueError, match="requires element to be a Struct"):
+        List(element=Int32(), partition_by=("x",))
+    with pytest.raises(ValueError, match="requires element to be a Struct"):
+        List(element=Int32(), cluster_by=("x",))
 
 
 @pytest.mark.parametrize(["param"], (("partition_by",), ("cluster_by",)))
-def test_Struct_field_references(param: str) -> None:
-    match = re.escape("Unknown field(s): {'z'}")
+def test_List_field_references(param: str) -> None:
+    match = re.escape("unknown field(s): {'z'}")
 
     with pytest.raises(ValueError, match=match):
-        Struct(fields={"x": Int32(), "y": Int32()}, **{param: ("z",)})
+        List(element=Struct(fields={"x": Int32(), "y": Int32()}), **{param: ("z",)})
 
     # Test with a bad `fields` and confirm the error *does not* contain the "unknown field" error.
-    with pytest.raises(ValueError) as exc:
-        Struct(fields="junk", **{param: ("z",)})
+    with pytest.raises(ValueError, match="expected an instance of") as exc:
+        List(element="junk", **{param: ("z",)})
     with pytest.raises(AssertionError):
         exc.match(match)
 
 
-def test_Struct_partition_cluster_overlapping() -> None:
-    match = re.escape("Clustering fields overlap with partition fields: {'x'}")
+def test_List_partition_cluster_overlapping() -> None:
+    match = re.escape("clustering fields overlap with partition fields: {'x'}")
 
     with pytest.raises(ValueError, match=match):
-        Struct(fields={"x": Int32(), "y": Int32()}, partition_by=("x",), cluster_by=("x",))
+        List(
+            element=Struct(fields={"x": Int32(), "y": Int32()}),
+            partition_by=("x",),
+            cluster_by=("x",),
+        )
 
     # Test with a bad `partition_by` and confirm the error *does not* contain the "unknown field" error.
-    with pytest.raises(ValueError) as exc:
-        Struct(fields={"x": Int32(), "y": Int32()}, partition_by="x", cluster_by=("y",))
+    with pytest.raises(ValueError, match="expected an instance of") as exc:
+        List(
+            element=Struct(fields={"x": Int32(), "y": Int32()}), partition_by="x", cluster_by=("y",)
+        )
     with pytest.raises(AssertionError):
         exc.match(match)
 
 
-@pytest.mark.parametrize(
-    ["sub_params"],
-    (
-        ({"partition_by": ("x",)},),
-        ({"cluster_by": ("x",)},),
-        ({"partition_by": ("x",), "cluster_by": ("y",)},),
-    ),
-)
-def test_Struct_incorrect_nesting(sub_params: dict[str, Any]) -> None:
-    sub = Struct(fields={"x": Int32(), "y": Int32()}, **sub_params)
-    with pytest.raises(
-        ValueError,
-        match=re.escape("Nested Structs ({'sub'}) cannot set `partition_by` or `cluster_by`"),
-    ):
-        Struct(fields={"sub": sub})
+def test_Struct() -> None:
+    fields = {"x": Int32(), "y": Int32()}
+    s = Struct(fields=fields)
+    assert isinstance(s.fields, frozendict)
+    assert s.fields == frozendict(fields)
 
 
 def test_Timestamp() -> None:
@@ -209,3 +218,7 @@ def test_TypeSystem(
     dummy.register_adapter(Int32Adapter)
     assert isinstance(dummy.to_artigraph(MyInt, hints={}), Int32)
     assert dummy.to_system(Int32(), hints={}) is MyInt
+
+    dummy.check_similarity(arti=Int32(), python_type=MyInt)
+    with pytest.raises(ValueError, match="MyInt'> cannot be used to represent Float64"):
+        dummy.check_similarity(arti=Float64(), python_type=MyInt)
diff --git a/tests/arti/versions/test_version.py b/tests/arti/versions/test_version.py
index 82c67b5f..155b703c 100644
--- a/tests/arti/versions/test_version.py
+++ b/tests/arti/versions/test_version.py
@@ -7,7 +7,7 @@
 
 def test_GitCommit() -> None:
     assert GitCommit(sha="test").sha == "test"
-    assert GitCommit(sha="test").fingerprint == 8581389452482819506
+    assert GitCommit(sha="test").fingerprint == -4338587361150152260
     assert len(GitCommit().sha) == 40  # Length of git sha
 
 
@@ -37,8 +37,8 @@ class P:
 
     assert isinstance(P.version, String)
     assert P.version.value == "    class P:\n        version: Version = _Source()\n"
-    assert P.version.fingerprint == -4528092110694557253
-    assert P().version.fingerprint == -4528092110694557253
+    assert P.version.fingerprint == -5384942283309384283
+    assert P().version.fingerprint == -5384942283309384283
 
     class P2:
         version = _Source()
@@ -47,7 +47,7 @@ class P2:
 
 
 def test_String() -> None:
-    assert String(value="ok").fingerprint == 5227454011934222951
+    assert String(value="ok").fingerprint == 5676437635634367418
 
 
 def test_Timestamp() -> None:
@@ -64,8 +64,5 @@ def test_Timestamp() -> None:
 
 def test_Version() -> None:
     # Version sets an @abstractmethod, so ABC catches it before our Model._abstract_ validator.
-    with pytest.raises(
-        TypeError,
-        match="Can't instantiate abstract class Version with abstract method fingerprint",
-    ):
-        Version()  # type: ignore
+    with pytest.raises(ValueError, match="cannot be instantiated directly"):
+        Version()
diff --git a/tests/arti/views/test_python.py b/tests/arti/views/test_python.py
index 4e0ba599..298b4a25 100644
--- a/tests/arti/views/test_python.py
+++ b/tests/arti/views/test_python.py
@@ -14,30 +14,40 @@ def test_python_View() -> None:
     for val, view, python_type in [
         (date(1970, 1, 1), Date, date),
         (datetime(1970, 1, 1, 0), Datetime, datetime),
-        (dict(a=1), Dict, dict),
+        (dict(a=1), Dict, dict[str, int]),
         (1.0, Float, float),
         (1, Int, int),
         (None, Null, NoneType),
         ("", Str, str),
     ]:
+        assert View.get_class_for(python_type) is view
+
+        test_type = view.type_system.to_artigraph(python_type, hints={})
+        test_format = Pickle()
+        test_view = view()
+
         binary = pickle.dumps(val)
         with named_temporary_file("w+b") as f:
+            test_storage_partition = LocalFilePartition(keys={}, path=f.name)
+
             f.write(binary)
             f.seek(0)
 
-            test_format = Pickle()
-            test_storage_partition = LocalFilePartition(keys={}, path=f.name)
-            test_view = view()
-            assert View._registry_[python_type] is view
             # read returns a list, matching the passed partitions
-            data, *tail = read(
-                format=test_format, storage_partitions=[test_storage_partition], view=test_view
+            data = read(
+                type=test_type,
+                format=test_format,
+                storage_partitions=(test_storage_partition,),
+                view=test_view,
             )
             assert data == val
-            assert not tail
 
             f.truncate()
             write(
-                data, format=test_format, storage_partition=test_storage_partition, view=test_view
+                data,
+                type=test_type,
+                format=test_format,
+                storage_partition=test_storage_partition,
+                view=test_view,
             )
             assert f.read() == binary
diff --git a/tests/arti/views/test_views.py b/tests/arti/views/test_views.py
index fddd6908..92ce7f56 100644
--- a/tests/arti/views/test_views.py
+++ b/tests/arti/views/test_views.py
@@ -3,27 +3,45 @@
 import pytest
 from pydantic import ValidationError
 
+from arti.types.python import python_type_system
 from arti.views import View
 
 
-def test_View() -> None:
+@pytest.fixture
+def MockView() -> type[View]:
+    class V(View):
+        _abstract_ = True
+        _by_python_type_: ClassVar[dict[type, type[View]]] = {}
+        type_system = python_type_system
+
+    return V
+
+
+def test_View_init() -> None:
     with pytest.raises(ValidationError, match="cannot be instantiated directly"):
         View()
 
 
-def test_View_registry() -> None:
-    class V(View):
-        _abstract_ = True
-        _registry_: ClassVar[dict[type, type[View]]] = {}
-
-    class Int(V):
+def test_View_registry(MockView: type[View]) -> None:
+    class Int(MockView):  # type: ignore
         python_type = int
 
-    class Int2(V):
-        python_type = int
+    class Int2(MockView):  # type: ignore
         priority = Int.priority + 1
+        python_type = int
 
-    class Str(V):
+    class Str(MockView):  # type: ignore
         python_type = str
 
-    assert V._registry_ == {int: Int2, str: Str}
+    assert MockView._by_python_type_ == {int: Int2, str: Str}
+
+
+def test_View_get_class_for(MockView: type[View]) -> None:
+    with pytest.raises(ValueError, match="cannot be matched to a View, try setting one explicitly"):
+        MockView.get_class_for(list)
+
+    class List(MockView):  # type: ignore
+        python_type = list
+
+    assert MockView.get_class_for(list) is List
+    assert MockView.get_class_for(list[int]) is List