Add IOType ABC (typed-artifact contract)

This PR, on top of Netflix#3117, adds the tiny contract that extension authors
target when they want to ship typed artifacts. Only the abstract base
lives in OSS; concrete scalar/tensor/struct/etc. types and the bridge
serializer live downstream (per review with Romain — core should not
carry opinions about which types exist, how enums are wire-encoded,
tensor byte order, dataclass inference, etc.).

Standard Python primitives (``int``, ``str``, ``list``, ``dict``, ...)
continue to flow through ``PickleSerializer`` unchanged. Wrapping is
opt-in, for types that need metadata or invariants attached.

IOType contract
- ``serialize(format=...)`` / ``deserialize(data, format=..., **kw)``
  mirror the signature that Netflix#3117 added on ``ArtifactSerializer``. The
  same ``WIRE`` / ``STORAGE`` constants govern dispatch so a single
  subclass owns both representations:
  - ``STORAGE`` → ``(List[SerializedBlob], metadata_dict)`` for the
    datastore save path.
  - ``WIRE`` → ``str`` for CLI args, protobuf payloads, and
    cross-process IPC.
- Subclasses implement four hooks: ``_wire_serialize``,
  ``_wire_deserialize``, ``_storage_serialize``,
  ``_storage_deserialize``.
- ``type_name`` + ``to_spec()`` support JSON schema generation.
- ``IOType`` itself is abstract; instantiating without implementing
  the four hooks raises ``TypeError``.

Tests
- ``test/unit/io_types/test_base.py`` — covers abstract instantiation,
  WIRE and STORAGE round-trips, default format, invalid format
  rejection, descriptor-mode spec output, and equality/hash.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Saeid Barati

metaflow/io_types/__init__.py

@@ -0,0 +1 @@
+from .base import IOType

metaflow/io_types/base.py

@@ -0,0 +1,138 @@
+"""Typed-artifact contract for Metaflow.
+
+This module defines the minimal :class:`IOType` abstract base class. OSS
+Metaflow ships the contract; concrete types (scalars, tensors, enums,
+dataclass-backed structs, etc.) live in extensions — they embody
+deployment-specific opinions about encoding, byte order, and dataclass
+inference that do not belong in core.
+
+:class:`IOType` mirrors the ``format`` argument introduced on
+:class:`metaflow.datastore.artifacts.serializer.ArtifactSerializer` so a
+single subclass can own both representations:
+
+- ``STORAGE`` — blob-based, persisted through the datastore.
+- ``WIRE`` — string-based, for CLI args, protobuf payloads, and
+  cross-process IPC.
+
+Subclasses implement four hooks (``_wire_serialize``, ``_wire_deserialize``,
+``_storage_serialize``, ``_storage_deserialize``); callers use the public
+``serialize(format=...)`` / ``deserialize(data, format=...)`` methods.
+"""
+
+from abc import ABCMeta, abstractmethod
+
+from metaflow.datastore.artifacts.serializer import STORAGE, WIRE
+
+
+_UNSET = object()
+
+
+class IOType(object, metaclass=ABCMeta):
+    """
+    Base class for typed Metaflow artifacts.
+
+    An :class:`IOType` instance plays two roles:
+
+    - **Descriptor** (no value): ``Int64`` in a spec describes an int64
+      field.
+    - **Wrapper** (with value): ``Int64(42)`` wraps a value for typed
+      serialization.
+
+    Subclasses implement four internal operations, dispatched by the
+    ``format`` argument of the public :meth:`serialize` / :meth:`deserialize`
+    methods.
+    """
+
+    type_name = None  # e.g. "text", "json", "int64" — set by subclasses.
+
+    def __init__(self, value=_UNSET):
+        self._value = value
+
+    @property
+    def value(self):
+        """The wrapped Python value, or ``_UNSET`` if this is a pure descriptor."""
+        return self._value
+
+    # -- Public API --------------------------------------------------------
+
+    def serialize(self, format=STORAGE):
+        """
+        Serialize the wrapped value. Must be side-effect-free.
+
+        Parameters
+        ----------
+        format : str
+            ``STORAGE`` (default) returns ``(List[SerializedBlob], dict)``.
+            ``WIRE`` returns a ``str``.
+        """
+        if format == WIRE:
+            return self._wire_serialize()
+        if format == STORAGE:
+            return self._storage_serialize()
+        raise ValueError("format must be %r or %r, got %r" % (STORAGE, WIRE, format))
+
+    @classmethod
+    def deserialize(cls, data, format=STORAGE, **kwargs):
+        """
+        Reconstruct an :class:`IOType` from serialized data.
+
+        Parameters
+        ----------
+        data : Union[str, List[bytes]]
+            ``str`` when ``format=WIRE``; ``List[bytes]`` when ``format=STORAGE``.
+        format : str
+            ``STORAGE`` (default) or ``WIRE``.
+        **kwargs
+            Forwarded to the underlying ``_storage_deserialize`` hook
+            (e.g. metadata the datastore produced at save time).
+        """
+        if format == WIRE:
+            return cls._wire_deserialize(data)
+        if format == STORAGE:
+            return cls._storage_deserialize(data, **kwargs)
+        raise ValueError("format must be %r or %r, got %r" % (STORAGE, WIRE, format))
+
+    # -- Subclass hooks ----------------------------------------------------
+
+    @abstractmethod
+    def _wire_serialize(self):
+        """Value -> string (for CLI args, protobuf, external APIs)."""
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    def _wire_deserialize(cls, s):
+        """String -> :class:`IOType` instance."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def _storage_serialize(self):
+        """Value -> ``(List[SerializedBlob], metadata_dict)``. Side-effect-free."""
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    def _storage_deserialize(cls, blobs, **kwargs):
+        """``(List[bytes], metadata)`` -> :class:`IOType` instance."""
+        raise NotImplementedError
+
+    # -- Spec generation ---------------------------------------------------
+
+    def to_spec(self):
+        """JSON type spec. Works with or without a wrapped value."""
+        return {"type": self.type_name}
+
+    # -- Dunder ------------------------------------------------------------
+
+    def __repr__(self):
+        if self._value is _UNSET:
+            return "%s()" % self.__class__.__name__
+        return "%s(%r)" % (self.__class__.__name__, self._value)
+
+    def __eq__(self, other):
+        if type(self) is not type(other):
+            return NotImplemented
+        return self._value == other._value
+
+    def __hash__(self):
+        return hash((type(self), self._value))

test/unit/io_types/__init__.py

@@ -0,0 +1,67 @@
+"""Contract tests for the IOType ABC."""
+
+import pytest
+
+from metaflow.datastore.artifacts.serializer import STORAGE, WIRE, SerializedBlob
+from metaflow.io_types import IOType
+
+
+class _TextIOType(IOType):
+    """Minimal concrete IOType used only to exercise the base-class dispatch."""
+
+    type_name = "test_text"
+
+    def _wire_serialize(self):
+        return self._value
+
+    @classmethod
+    def _wire_deserialize(cls, s):
+        return cls(s)
+
+    def _storage_serialize(self):
+        blob = self._value.encode("utf-8")
+        return [SerializedBlob(blob)], {"length": len(blob)}
+
+    @classmethod
+    def _storage_deserialize(cls, blobs, **kwargs):
+        return cls(blobs[0].decode("utf-8"))
+
+
+def test_cannot_instantiate_abstract_base():
+    with pytest.raises(TypeError):
+        IOType()  # missing hook implementations
+
+
+def test_wire_roundtrip():
+    wire = _TextIOType("hi").serialize(format=WIRE)
+    assert wire == "hi"
+    assert _TextIOType.deserialize(wire, format=WIRE) == _TextIOType("hi")
+
+
+def test_storage_roundtrip():
+    blobs, meta = _TextIOType("hi").serialize(format=STORAGE)
+    assert meta["length"] == 2
+    raw = [b.value for b in blobs]
+    assert _TextIOType.deserialize(raw, format=STORAGE) == _TextIOType("hi")
+
+
+def test_default_format_is_storage():
+    out = _TextIOType("hi").serialize()
+    assert isinstance(out, tuple)  # (blobs, metadata)
+
+
+def test_invalid_format_raises():
+    with pytest.raises(ValueError):
+        _TextIOType("hi").serialize(format="bogus")
+    with pytest.raises(ValueError):
+        _TextIOType.deserialize("hi", format="bogus")
+
+
+def test_descriptor_has_no_value():
+    assert _TextIOType().to_spec() == {"type": "test_text"}
+
+
+def test_eq_and_hash():
+    assert _TextIOType("x") == _TextIOType("x")
+    assert _TextIOType("x") != _TextIOType("y")
+    assert hash(_TextIOType("x")) == hash(_TextIOType("x"))