diff --git a/src/citrine/__version__.py b/src/citrine/__version__.py
index fcd7ddb9e..9ce9954cf 100644
--- a/src/citrine/__version__.py
+++ b/src/citrine/__version__.py
@@ -1 +1 @@
-__version__ = "3.9.0"
+__version__ = "3.10.0"
diff --git a/src/citrine/informatics/data_sources.py b/src/citrine/informatics/data_sources.py
index d1b746eaa..153d9e6ac 100644
--- a/src/citrine/informatics/data_sources.py
+++ b/src/citrine/informatics/data_sources.py
@@ -1,17 +1,23 @@
 """Tools for working with Descriptors."""
+from abc import abstractmethod
 from typing import Type, List, Mapping, Optional, Union
 from uuid import UUID
+from warnings import warn
 
 from citrine._serialization import properties
 from citrine._serialization.polymorphic_serializable import PolymorphicSerializable
 from citrine._serialization.serializable import Serializable
 from citrine.informatics.descriptors import Descriptor
 from citrine.resources.file_link import FileLink
+from citrine.resources.gemtables import GemTable
 
-__all__ = ['DataSource',
-           'CSVDataSource',
-           'GemTableDataSource',
-           'ExperimentDataSourceRef']
+__all__ = [
+    'DataSource',
+    'CSVDataSource',
+    'GemTableDataSource',
+    'ExperimentDataSourceRef',
+    'SnapshotDataSource',
+]
 
 
 class DataSource(PolymorphicSerializable['DataSource']):
@@ -28,19 +34,43 @@ def __eq__(self, other):
         else:
             return False
 
+    @classmethod
+    def _subclass_list(self) -> List[Type[Serializable]]:
+        return [CSVDataSource, GemTableDataSource, ExperimentDataSourceRef, SnapshotDataSource]
+
     @classmethod
     def get_type(cls, data) -> Type[Serializable]:
         """Return the subtype."""
         if "type" not in data:
             raise ValueError("Can only get types from dicts with a 'type' key")
-        types: List[Type[Serializable]] = [
-            CSVDataSource, GemTableDataSource, ExperimentDataSourceRef
-        ]
-        res = next((x for x in types if x.typ == data["type"]), None)
+        res = next((x for x in cls._subclass_list() if x.typ == data["type"]), None)
         if res is None:
-            raise ValueError("Unrecognized type: {}".format(data["type"]))
+            raise ValueError(f"Unrecognized type: {data['type']}")
         return res
 
+    @property
+    @abstractmethod
+    def _data_source_type(self) -> str:
+        """The data source type string, which is the leading term of the data_source_id."""
+
+    @classmethod
+    def from_data_source_id(cls, data_source_id: str) -> "DataSource":
+        """Build a DataSource from a datasource_id."""
+        terms = data_source_id.split("::")
+        res = next((x for x in cls._subclass_list() if x._data_source_type == terms[0]), None)
+        if res is None:
+            raise ValueError(f"Unrecognized type: {terms[0]}")
+        return res._data_source_id_builder(*terms[1:])
+
+    @classmethod
+    @abstractmethod
+    def _data_source_id_builder(cls, *args) -> "DataSource":
+        """Build a DataSource based on a parsed data_source_id."""
+
+    @abstractmethod
+    def to_data_source_id(self) -> str:
+        """Generate the data_source_id for this DataSource."""
+
 
 class CSVDataSource(Serializable['CSVDataSource'], DataSource):
     """A data source based on a CSV file stored on the data platform.
@@ -65,6 +95,8 @@ class CSVDataSource(Serializable['CSVDataSource'], DataSource):
         properties.String, properties.Object(Descriptor), "column_definitions")
     identifiers = properties.Optional(properties.List(properties.String), "identifiers")
 
+    _data_source_type = "csv"
+
     def __init__(self,
                  *,
                  file_link: FileLink,
@@ -74,6 +106,21 @@ def __init__(self,
         self.column_definitions = column_definitions
         self.identifiers = identifiers
 
+    @classmethod
+    def _data_source_id_builder(cls, *args) -> DataSource:
+        # TODO Figure out how to populate the column definitions
+        warn("A CSVDataSource was derived from a data_source_id "
+             "but is missing its column_definitions and identities",
+             UserWarning)
+        return CSVDataSource(
+            file_link=FileLink(url=args[0], filename=args[1]),
+            column_definitions={}
+        )
+
+    def to_data_source_id(self) -> str:
+        """Generate the data_source_id for this DataSource."""
+        return f"{self._data_source_type}::{self.file_link.url}::{self.file_link.filename}"
+
 
 class GemTableDataSource(Serializable['GemTableDataSource'], DataSource):
     """A data source based on a GEM Table hosted on the data platform.
@@ -92,6 +139,8 @@ class GemTableDataSource(Serializable['GemTableDataSource'], DataSource):
     table_id = properties.UUID("table_id")
     table_version = properties.Integer("table_version")
 
+    _data_source_type = "gemd"
+
     def __init__(self,
                  *,
                  table_id: UUID,
@@ -99,6 +148,26 @@ def __init__(self,
         self.table_id: UUID = table_id
         self.table_version: Union[int, str] = table_version
 
+    @classmethod
+    def _data_source_id_builder(cls, *args) -> DataSource:
+        return GemTableDataSource(table_id=UUID(args[0]), table_version=args[1])
+
+    def to_data_source_id(self) -> str:
+        """Generate the data_source_id for this DataSource."""
+        return f"{self._data_source_type}::{self.table_id}::{self.table_version}"
+
+    @classmethod
+    def from_gemtable(cls, table: GemTable) -> "GemTableDataSource":
+        """Generate a DataSource that corresponds to a GemTable.
+
+        Parameters
+        ----------
+        table: GemTable
+            The GemTable object to reference
+
+        """
+        return GemTableDataSource(table_id=table.uid, table_version=table.version)
+
 
 class ExperimentDataSourceRef(Serializable['ExperimentDataSourceRef'], DataSource):
     """A reference to a data source based on an experiment result hosted on the data platform.
@@ -113,5 +182,42 @@ class ExperimentDataSourceRef(Serializable['ExperimentDataSourceRef'], DataSourc
     typ = properties.String('type', default='experiments_data_source', deserializable=False)
     datasource_id = properties.UUID("datasource_id")
 
+    _data_source_type = "experiments"
+
     def __init__(self, *, datasource_id: UUID):
         self.datasource_id: UUID = datasource_id
+
+    @classmethod
+    def _data_source_id_builder(cls, *args) -> DataSource:
+        return ExperimentDataSourceRef(datasource_id=UUID(args[0]))
+
+    def to_data_source_id(self) -> str:
+        """Generate the data_source_id for this DataSource."""
+        return f"{self._data_source_type}::{self.datasource_id}"
+
+
+class SnapshotDataSource(Serializable['SnapshotDataSource'], DataSource):
+    """A reference to a data source based on a Snapshot on the data platform.
+
+    Parameters
+    ----------
+    snapshot_id: UUID
+        Unique identifier for the Snapshot Data Source
+
+    """
+
+    typ = properties.String('type', default='snapshot_data_source', deserializable=False)
+    snapshot_id = properties.UUID("snapshot_id")
+
+    _data_source_type = "snapshot"
+
+    def __init__(self, *, snapshot_id: UUID):
+        self.snapshot_id = snapshot_id
+
+    @classmethod
+    def _data_source_id_builder(cls, *args) -> DataSource:
+        return SnapshotDataSource(snapshot_id=UUID(args[0]))
+
+    def to_data_source_id(self) -> str:
+        """Generate the data_source_id for this DataSource."""
+        return f"{self._data_source_type}::{self.snapshot_id}"
diff --git a/src/citrine/informatics/workflows/design_workflow.py b/src/citrine/informatics/workflows/design_workflow.py
index a1073b5b6..a63848fe4 100644
--- a/src/citrine/informatics/workflows/design_workflow.py
+++ b/src/citrine/informatics/workflows/design_workflow.py
@@ -3,6 +3,7 @@
 
 from citrine._rest.resource import Resource
 from citrine._serialization import properties
+from citrine.informatics.data_sources import DataSource
 from citrine.informatics.workflows.workflow import Workflow
 from citrine.resources.design_execution import DesignExecutionCollection
 from citrine._rest.ai_resource_metadata import AIResourceMetadata
@@ -31,11 +32,12 @@ class DesignWorkflow(Resource['DesignWorkflow'], Workflow, AIResourceMetadata):
     design_space_id = properties.Optional(properties.UUID, 'design_space_id')
     predictor_id = properties.Optional(properties.UUID, 'predictor_id')
     predictor_version = properties.Optional(
-        properties.Union([properties.Integer(), properties.String()]), 'predictor_version')
+        properties.Union([properties.Integer, properties.String]), 'predictor_version')
     branch_root_id: Optional[UUID] = properties.Optional(properties.UUID, 'branch_root_id')
     """:Optional[UUID]: Root ID of the branch that contains this workflow."""
     branch_version: Optional[int] = properties.Optional(properties.Integer, 'branch_version')
     """:Optional[int]: Version number of the branch that contains this workflow."""
+    data_source = properties.Optional(properties.Object(DataSource), "data_source")
 
     status_description = properties.String('status_description', serializable=False)
     """:str: more detailed description of the workflow's status"""
@@ -50,16 +52,35 @@ def __init__(self,
                  design_space_id: Optional[UUID] = None,
                  predictor_id: Optional[UUID] = None,
                  predictor_version: Optional[Union[int, str]] = None,
+                 data_source: Optional[DataSource] = None,
                  description: Optional[str] = None):
         self.name = name
         self.design_space_id = design_space_id
         self.predictor_id = predictor_id
         self.predictor_version = predictor_version
+        self.data_source = data_source
         self.description = description
 
     def __str__(self):
         return '<DesignWorkflow {!r}>'.format(self.name)
 
+    @classmethod
+    def _pre_build(cls, data: dict) -> dict:
+        """Run data modification before building."""
+        data_source_id = data.pop("data_source_id", None)
+        if data_source_id is not None:
+            data["data_source"] = DataSource.from_data_source_id(data_source_id).dump()
+        return data
+
+    def _post_dump(self, data: dict) -> dict:
+        """Run data modification after dumping."""
+        data_source = data.pop("data_source", None)
+        if data_source is not None:
+            data["data_source_id"] = DataSource.build(data_source).to_data_source_id()
+        else:
+            data["data_source_id"] = None
+        return data
+
     @property
     def design_executions(self) -> DesignExecutionCollection:
         """Return a resource representing all visible executions of this workflow."""
@@ -67,3 +88,18 @@ def design_executions(self) -> DesignExecutionCollection:
             raise AttributeError('Cannot initialize execution without project reference!')
         return DesignExecutionCollection(
             project_id=self.project_id, session=self._session, workflow_id=self.uid)
+
+    @property
+    def data_source_id(self) -> Optional[str]:
+        """A resource referencing the workflow's data source."""
+        if self.data_source is None:
+            return None
+        else:
+            return self.data_source.to_data_source_id()
+
+    @data_source_id.setter
+    def data_source_id(self, value: Optional[str]):
+        if value is None:
+            self.data_source = None
+        else:
+            self.data_source = DataSource.from_data_source_id(value)
diff --git a/tests/conftest.py b/tests/conftest.py
index 74112dfc5..4548a208e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -949,16 +949,3 @@ def predictor_evaluation_workflow_dict(generic_entity, example_cv_evaluator_dict
         "evaluators": [example_cv_evaluator_dict, example_holdout_evaluator_dict]
     })
     return ret
-
-@pytest.fixture
-def design_workflow_dict(generic_entity):
-    ret = generic_entity.copy()
-    ret.update({
-        "name": "Example Design Workflow",
-        "description": "A description! Of the Design Workflow! So you know what it's for!",
-        "design_space_id": str(uuid.uuid4()),
-        "predictor_id": str(uuid.uuid4()),
-        "predictor_version": random.randint(1, 10),
-        "branch_id": str(uuid.uuid4()),
-    })
-    return ret
diff --git a/tests/informatics/test_data_source.py b/tests/informatics/test_data_source.py
index c5819247b..7890c9241 100644
--- a/tests/informatics/test_data_source.py
+++ b/tests/informatics/test_data_source.py
@@ -3,10 +3,14 @@
 
 import pytest
 
-from citrine.informatics.data_sources import DataSource, CSVDataSource, ExperimentDataSourceRef, GemTableDataSource
-from citrine.informatics.descriptors import RealDescriptor, FormulationDescriptor
+from citrine.informatics.data_sources import (
+    DataSource, CSVDataSource, ExperimentDataSourceRef, GemTableDataSource, SnapshotDataSource
+)
+from citrine.informatics.descriptors import RealDescriptor
 from citrine.resources.file_link import FileLink
+from citrine.resources.gemtables import GemTable
 
+from tests.utils.factories import GemTableDataFactory
 
 @pytest.fixture(params=[
     CSVDataSource(file_link=FileLink("foo.spam", "http://example.com"),
@@ -15,7 +19,8 @@
     GemTableDataSource(table_id=uuid.uuid4(), table_version=1),
     GemTableDataSource(table_id=uuid.uuid4(), table_version="2"),
     GemTableDataSource(table_id=uuid.uuid4(), table_version="2"),
-    ExperimentDataSourceRef(datasource_id=uuid.uuid4())
+    ExperimentDataSourceRef(datasource_id=uuid.uuid4()),
+    SnapshotDataSource(snapshot_id=uuid.uuid4())
 ])
 def data_source(request):
     return request.param
@@ -39,3 +44,24 @@ def test_invalid_deser():
 
     with pytest.raises(ValueError):
         DataSource.build({"type": "foo"})
+
+
+def test_data_source_id(data_source):
+    if isinstance(data_source, CSVDataSource):
+        # TODO: There's no obvious way to recover the column_definitions & identifiers from the ID
+        with pytest.warns(UserWarning):
+            transformed = DataSource.from_data_source_id(data_source.to_data_source_id())
+        assert isinstance(data_source, CSVDataSource)
+        assert transformed.file_link == data_source.file_link
+    else:
+        assert data_source == DataSource.from_data_source_id(data_source.to_data_source_id())
+
+def test_from_gem_table():
+    table = GemTable.build(GemTableDataFactory())
+    data_source = GemTableDataSource.from_gemtable(table)
+    assert data_source.table_id == table.uid
+    assert data_source.table_version == table.version
+
+def test_invalid_data_source_id():
+    with pytest.raises(ValueError):
+        DataSource.from_data_source_id(f"Undefined::{uuid.uuid4()}")
diff --git a/tests/informatics/test_workflows.py b/tests/informatics/test_workflows.py
index f73ad3eee..7d5fcdcb1 100644
--- a/tests/informatics/test_workflows.py
+++ b/tests/informatics/test_workflows.py
@@ -1,20 +1,18 @@
 """Tests for citrine.informatics.workflows."""
-import json
+from multiprocessing.reduction import register
 from uuid import uuid4, UUID
 
-import mock
 import pytest
 
-from citrine._session import Session
-from citrine.informatics.design_candidate import DesignMaterial, DesignVariable, DesignCandidate, ChemicalFormula, \
+from citrine.informatics.design_candidate import DesignMaterial, DesignCandidate, ChemicalFormula, \
     MeanAndStd, TopCategories, Mixture, MolecularStructure
 from citrine.informatics.executions import DesignExecution
 from citrine.informatics.predict_request import PredictRequest
-from citrine.informatics.workflows import DesignWorkflow, Workflow
+from citrine.informatics.workflows import DesignWorkflow
 from citrine.resources.design_execution import DesignExecutionCollection
 from citrine.resources.design_workflow import DesignWorkflowCollection
 
-from tests.utils.factories import BranchDataFactory
+from tests.utils.factories import BranchDataFactory, DesignWorkflowDataFactory
 from tests.utils.session import FakeSession, FakeCall
 
 
@@ -48,10 +46,8 @@ def execution_collection(session) -> DesignExecutionCollection:
 
 
 @pytest.fixture
-def design_workflow(collection, design_workflow_dict) -> DesignWorkflow:
-    workflow = collection.build(design_workflow_dict)
-    collection.session.calls.clear()
-    return workflow
+def design_workflow(collection) -> DesignWorkflow:
+    return collection.build(DesignWorkflowDataFactory(register=True))
 
 @pytest.fixture
 def design_execution(execution_collection, design_execution_dict) -> DesignExecution:
diff --git a/tests/resources/test_design_workflows.py b/tests/resources/test_design_workflows.py
index d37eb0443..077c7346c 100644
--- a/tests/resources/test_design_workflows.py
+++ b/tests/resources/test_design_workflows.py
@@ -6,10 +6,16 @@
 
 from citrine.informatics.workflows import DesignWorkflow
 from citrine.resources.design_workflow import DesignWorkflowCollection
-from tests.utils.factories import BranchDataFactory
+from tests.utils.factories import (
+    BranchDataFactory, DesignWorkflowDataFactory, TableDataSourceFactory
+)
 from tests.utils.session import FakeSession, FakeCall
 
-PARTIAL_DW_ARGS = (("predictor_id", uuid.uuid4), ("design_space_id", uuid.uuid4))
+PARTIAL_DW_ARGS = (
+    ("data_source_id", lambda: TableDataSourceFactory().to_data_source_id()),
+    ("predictor_id", lambda: str(uuid.uuid4())),
+    ("design_space_id", lambda: str(uuid.uuid4()))
+)
 OPTIONAL_ARGS = PARTIAL_DW_ARGS + (("predictor_version", lambda: random.randint(1, 10)),)
 
 
@@ -42,24 +48,9 @@ def collection(branch_data, collection_without_branch) -> DesignWorkflowCollecti
 
 
 @pytest.fixture
-def workflow(collection, branch_data, design_workflow_dict) -> DesignWorkflow:
-    design_workflow_dict["branch_root_id"] = branch_data["metadata"]["root_id"]
-    design_workflow_dict["branch_version"] = branch_data["metadata"]["version"]
-
-    collection.session.set_response(branch_data)
-    workflow = collection.build(design_workflow_dict)
-    collection.session.calls.clear()
-
-    workflow.uid = uuid.uuid4()
-    return workflow
-
-
-@pytest.fixture
-def workflow_minimal(collection, workflow) -> DesignWorkflow:
-    workflow.predictor_id = None
-    workflow.predictor_version = None
-    workflow.design_space_id = None
-    return workflow
+def workflow(collection, branch_data) -> DesignWorkflow:
+    workflow_data = DesignWorkflowDataFactory(branch=branch_data, register=True)
+    return collection.build(workflow_data)
 
 
 def all_combination_lengths(vals, maxlen=None):
@@ -79,6 +70,7 @@ def assert_workflow(actual, expected, *, include_branch=False):
     assert actual.design_space_id == expected.design_space_id
     assert actual.predictor_id == expected.predictor_id
     assert actual.predictor_version == expected.predictor_version
+    assert actual.data_source_id == expected.data_source_id
     assert actual.project_id == expected.project_id
     if include_branch:
         assert actual._branch_id == expected._branch_id
@@ -86,34 +78,31 @@ def assert_workflow(actual, expected, *, include_branch=False):
         assert actual.branch_version == expected.branch_version
 
 
-def test_basic_methods(workflow, collection, design_workflow_dict):
+def test_basic_methods(workflow, collection):
     assert 'DesignWorkflow' in str(workflow)
     assert workflow.design_executions.project_id == workflow.project_id
 
 
 @pytest.mark.parametrize("optional_args", all_combination_lengths(OPTIONAL_ARGS))
-def test_register(session, branch_data, workflow_minimal, collection, optional_args):
-    workflow = workflow_minimal
-    branch_root_id = branch_data['metadata']['root_id']
-    branch_version = branch_data['metadata']['version']
-
-    # Set a random value for all optional args selected for this run.
-    for name, factory in optional_args:
-        setattr(workflow, name, factory())
+def test_register(session, branch_data, collection, optional_args):
+    kw_args = {argument: None for argument, factory in OPTIONAL_ARGS}
+    kw_args.update({argument: factory() for argument, factory in optional_args})
+    workflow_data = DesignWorkflowDataFactory(**kw_args, branch=branch_data)
 
     # Given
-    post_dict = {**workflow.dump(), "branch_root_id": str(branch_root_id), "branch_version": branch_version}
-    session.set_responses({**post_dict, 'status_description': 'status'})
+    post_dict = {k: v for k, v in workflow_data.items() if k != 'status_description'}
+    session.set_responses(workflow_data)
 
     # When
-    new_workflow = collection.register(workflow)
+    old_workflow = collection.build(workflow_data)
+    new_workflow = collection.register(old_workflow)
 
     # Then
     assert session.calls == [FakeCall(method='POST', path=workflow_path(collection), json=post_dict)]
 
     assert new_workflow.branch_root_id == collection.branch_root_id
     assert new_workflow.branch_version == collection.branch_version
-    assert_workflow(new_workflow, workflow)
+    assert_workflow(new_workflow, old_workflow)
 
 
 def test_register_conflicting_branches(session, branch_data, workflow, collection):
@@ -139,8 +128,8 @@ def test_register_conflicting_branches(session, branch_data, workflow, collectio
     assert_workflow(new_workflow, workflow)
 
 
-def test_register_partial_workflow_without_branch(session, workflow_minimal, collection_without_branch):
-    workflow = workflow_minimal
+def test_register_partial_workflow_without_branch(session, collection_without_branch):
+    workflow = DesignWorkflow.build(DesignWorkflowDataFactory())
     with pytest.raises(RuntimeError):
         collection_without_branch.register(workflow)
 
@@ -180,15 +169,11 @@ def test_list_archived(branch_data, workflow, collection: DesignWorkflowCollecti
     )
 
 
-def test_missing_project(design_workflow_dict):
+def test_missing_project():
     """Make sure we get an attribute error if there is no project id."""
 
-    workflow = DesignWorkflow(
-        name=design_workflow_dict["name"],
-        predictor_id=design_workflow_dict["predictor_id"],
-        predictor_version=design_workflow_dict["predictor_version"],
-        design_space_id=design_workflow_dict["design_space_id"]
-    )
+    workflow = DesignWorkflow.build(DesignWorkflowDataFactory())
+    assert workflow.project_id is None  # Verify test assumption still holds
 
     with pytest.raises(AttributeError):
         workflow.design_executions
@@ -258,3 +243,17 @@ def test_update_branch_not_found(collection, workflow):
     # When
     with pytest.raises(ValueError):
         collection.update(workflow)
+
+def test_data_source_id(workflow):
+    original_id = workflow.data_source_id
+    assert workflow.data_source.to_data_source_id() == original_id
+    workflow.data_source.table_version += 1
+    assert workflow.data_source.to_data_source_id() != original_id
+    assert workflow.data_source.to_data_source_id() == workflow.data_source_id
+
+    workflow.data_source_id = None
+    assert workflow.data_source_id is None
+    assert workflow.data_source is None
+
+    workflow.data_source_id = original_id
+    assert workflow.data_source is not None
diff --git a/tests/seeding/test_find_or_create.py b/tests/seeding/test_find_or_create.py
index 897f82529..944563ceb 100644
--- a/tests/seeding/test_find_or_create.py
+++ b/tests/seeding/test_find_or_create.py
@@ -349,9 +349,9 @@ def test_create_or_update_unique_found_design_workflow(session):
     branch_data = BranchDataFactory()
     root_id = UUID(branch_data["metadata"]["root_id"])
     version = branch_data["metadata"]["version"]
-    dw1_dict = DesignWorkflowDataFactory()
-    dw2_dict = DesignWorkflowDataFactory(branch_root_id=root_id, branch_version=version)
-    dw3_dict = DesignWorkflowDataFactory()
+    dw1_dict = DesignWorkflowDataFactory(register=True)
+    dw2_dict = DesignWorkflowDataFactory(register=True, branch=branch_data)
+    dw3_dict = DesignWorkflowDataFactory(register=True)
     session.set_responses(
         # List
         {"response": [dw1_dict, dw2_dict, dw3_dict]},  # Return the design workflows
diff --git a/tests/utils/factories.py b/tests/utils/factories.py
index 962262451..e19aa69c1 100644
--- a/tests/utils/factories.py
+++ b/tests/utils/factories.py
@@ -14,6 +14,7 @@
 from citrine.gemd_queries.criteria import *
 from citrine.gemd_queries.filter import *
 from citrine.informatics.scores import LIScore
+from citrine.informatics.workflows import DesignWorkflow
 from citrine.resources.dataset import Dataset
 from citrine.resources.file_link import _Uploader
 from citrine.resources.material_run import MaterialRun
@@ -376,6 +377,18 @@ class TableDataSourceDataFactory(factory.DictFactory):
     table_id = factory.Faker("uuid4")
     table_version = factory.Faker('random_digit_not_null')
 
+from citrine.informatics.data_sources import GemTableDataSource
+
+class TableDataSourceFactory(factory.Factory):
+    class Meta:
+        model = GemTableDataSource
+
+    class Params:
+        data_factory = factory.SubFactory(TableDataSourceDataFactory)
+
+    table_id = factory.LazyAttribute(lambda o: o.data_factory["table_id"])
+    table_version = factory.LazyAttribute(lambda o: o.data_factory["table_version"])
+
 
 class StatusDataFactory(factory.DictFactory):
     # TODO  Create trait and info / detail content
@@ -470,17 +483,46 @@ class DesignSpaceDataFactory(factory.DictFactory):
 
 
 class DesignWorkflowDataFactory(factory.DictFactory):
-    id = factory.Faker('uuid4')
+    class Params:
+        data_source = factory.SubFactory(TableDataSourceFactory)
+        branch = factory.SubFactory(BranchDataFactory)
+        times = factory.List([factory.Faker("unix_milliseconds") for i in range(3)])
+        register = factory.Trait(
+            id = factory.Faker('uuid4'),
+            branch_id = factory.LazyAttribute(lambda o: o.branch["id"]),
+            created_by = factory.Faker('uuid4'),
+            updated_by = factory.LazyAttribute(lambda o: o.created_by),
+            create_time = factory.LazyAttribute(lambda o: sorted(o.times)[0]),
+            update_time = factory.LazyAttribute(lambda o: sorted(o.times)[0]),
+            # TODO: Create a Trait for statuses
+            status = "SUCCEEDED",
+            status_description = "READY",
+            status_info = [],
+            status_detail = []
+        )
+        update = factory.Trait(
+            register = True,
+            updated_by = factory.Faker('uuid4'),
+            update_time = factory.LazyAttribute(lambda o: sorted(o.times)[1])
+        )
+        archive = factory.Trait(
+            update = True,
+            archived = True,
+            archived_by = factory.Faker('uuid4'),
+            archive_time = factory.LazyAttribute(lambda o: sorted(o.times)[2]),
+        )
+
+    type = DesignWorkflow.typ
     name = factory.Faker("company")
     description = factory.Faker("catch_phrase")
-    archived = False
     design_space_id = factory.Faker("uuid4")
     predictor_id = factory.Faker("uuid4")
-    branch_id = factory.Faker("uuid4")
-    # TODO  Create Trait and status_detail content
-    status = "SUCCEEDED"
-    status_description = "READY"
-    status_detail = []
+    predictor_version = factory.Faker("random_digit_not_null")
+    data_source_id = factory.LazyAttribute(lambda o: o.data_source.to_data_source_id())
+    branch_root_id = factory.LazyAttribute(lambda o: o.branch["metadata"]["root_id"])
+    branch_version = factory.LazyAttribute(lambda o: o.branch["metadata"]["version"])
+    archived = False
+    status_description = ""  # TODO: Should be None, but property not defined as Optional
 
 
 class JobSubmissionResponseFactory(factory.DictFactory):
@@ -488,6 +530,9 @@ class JobSubmissionResponseFactory(factory.DictFactory):
 
 
 class DatasetDataFactory(factory.DictFactory):
+    class Params:
+        times = factory.List([factory.Faker("unix_milliseconds") for i in range(3)])
+
     id = factory.Faker('uuid4')
     name = factory.Faker('company')
     summary = factory.Faker('catch_phrase')
@@ -497,10 +542,9 @@ class DatasetDataFactory(factory.DictFactory):
     updated_by = factory.Faker('uuid4')
     deleted_by = factory.Faker('uuid4')
     unique_name = None  # TODO Update tests to include unique_name
-    # Unfortunately, this does not respect chronology
-    update_time = factory.Faker("unix_milliseconds")
-    create_time = factory.Faker("unix_milliseconds")
-    delete_time = factory.Faker("unix_milliseconds")
+    create_time = factory.LazyAttribute(lambda o: sorted(o.times)[0])
+    update_time = factory.LazyAttribute(lambda o: sorted(o.times)[1])
+    delete_time = factory.LazyAttribute(lambda o: sorted(o.times)[2])
     public = False