Repository: SFDO-Tooling/CumulusCI
Pull request #3285 (2022-07-12 15:44:43)

Add extract declarations concept to CumulusCI
Add extract declarations concept to CumulusCI.

Documentation will be added later when we unveil to the public.




Patch:
diff --git a/cumulusci/salesforce_api/org_schema.py b/cumulusci/salesforce_api/org_schema.py
index 10a890bb5..30eac3a5a 100644
--- a/cumulusci/salesforce_api/org_schema.py
+++ b/cumulusci/salesforce_api/org_schema.py
@@ -130,7 +130,7 @@ class Schema:
         try:
             return self.sobjects.filter_by(name=name).one()
         except exc.NoResultFound:
-            raise KeyError(f"No sobject named {name}")
+            raise KeyError(f"No sobject named `{name}`")
 
     def __contains__(self, name):
         return self.sobjects.filter_by(name=name).all()
@@ -432,7 +432,7 @@ def get_org_schema(
 class ZippableTempDb:
     """A database that loads and saves from a tempdir to a zippped cache"""
 
-    def __enter__(self) -> Path:
+    def __enter__(self) -> "ZippableTempDb":
         self.tempdir = TemporaryDirectory()
         self.tempfile = Path(self.tempdir.name) / "temp_org_schema.db"
         return self
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/calculate_dependencies.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/calculate_dependencies.py
new file mode 100644
index 000000000..82b36450a
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/calculate_dependencies.py
@@ -0,0 +1,99 @@
+import typing as T
+
+from cumulusci.salesforce_api.filterable_objects import NOT_EXTRACTABLE
+from cumulusci.salesforce_api.org_schema import Schema
+
+from .synthesize_extract_declarations import (
+    SimplifiedExtractDeclaration,
+    synthesize_declaration_for_sobject,
+)
+
+
+class Dependency(T.NamedTuple):
+    table_name_from: str
+    table_name_to: str
+    field_name: str
+
+
+def _calculate_dependencies_for_declarations(
+    decls: T.Sequence[SimplifiedExtractDeclaration], schema: Schema
+) -> T.Dict[str, T.List[Dependency]]:
+    """Ensure that required lookups are fulfilled for list of SimplifiedExtractDeclarations
+
+    Do this by adding their referent tables (in full) to the extract.
+    Module-internal helper function.
+    """
+    dependencies = {}
+    for decl in decls:
+        new_dependencies = _calculate_dependencies_for_sobject(
+            decl.sf_object, decl.fields, schema, only_required_fields=False
+        )
+        dependencies.update(new_dependencies)
+    return dependencies
+
+
+def _calculate_dependencies_for_sobject(
+    source_sfobject: str,
+    fields: T.List[str],
+    schema: Schema,
+    only_required_fields: bool,
+):
+    """Ensure that required lookups are fulfilled for a single SObject
+
+    Do this by adding its referent tables (in full) to the extract.
+    Module-internal helper function.
+    """
+    dependencies = {}
+    for field_name in fields:
+        field_info = schema[source_sfobject].fields[field_name]
+        if not field_info.createable:  # pragma: no cover
+            continue
+        references = field_info.referenceTo
+        if len(references) == 1:
+            target = references[0]
+
+            target_disallowed = target in NOT_EXTRACTABLE
+            field_disallowed = target_disallowed or not field_info.createable
+            field_allowed = not (only_required_fields or field_disallowed)
+            if field_info.requiredOnCreate or field_allowed:
+                dependencies.setdefault(source_sfobject, []).append(
+                    Dependency(source_sfobject, target, field_name)
+                )
+
+    return dependencies
+
+
+def extend_declarations_to_include_referenced_tables(
+    decl_list: T.Sequence[SimplifiedExtractDeclaration], schema: Schema
+) -> T.Sequence[SimplifiedExtractDeclaration]:
+    """Extend the declarations to complete required or requested lookups recursively"""
+    decls = {decl.sf_object: decl for decl in decl_list}
+    dependencies = _calculate_dependencies_for_declarations(decl_list, schema)
+    to_process = list(decls.keys())
+
+    while to_process:
+        sf_object = to_process.pop()
+        assert isinstance(sf_object, str)
+        my_dependencies = dependencies.get(sf_object, ())
+        for dep in my_dependencies:
+            target_table = dep.table_name_to
+            if target_table not in decls and target_table not in NOT_EXTRACTABLE:
+                required_fields = [
+                    field.name
+                    for field in schema[target_table].fields.values()
+                    if field.requiredOnCreate
+                ]
+                decls[target_table] = synthesize_declaration_for_sobject(
+                    target_table, required_fields
+                )
+
+                new_dependencies = _calculate_dependencies_for_sobject(
+                    target_table,
+                    decls[target_table].fields,
+                    schema,
+                    only_required_fields=True,
+                )
+                dependencies.update(new_dependencies)
+                to_process.append(target_table)
+
+    return list(decls.values())
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/extract_yml.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/extract_yml.py
new file mode 100644
index 000000000..bef031492
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/extract_yml.py
@@ -0,0 +1,123 @@
+import re
+import typing as T
+from enum import Enum
+from pathlib import Path
+
+from pydantic import Field, validator
+
+from cumulusci.tasks.bulkdata.step import DataApi
+from cumulusci.utils.yaml.model_parser import CCIDictModel, HashableBaseModel
+
+object_decl = re.compile(r"objects\((\w+)\)", re.IGNORECASE)
+field_decl = re.compile(r"fields\((\w+)\)", re.IGNORECASE)
+
+
+class SFObjectGroupTypes(Enum):
+    all = "all"
+    custom = "custom"
+    standard = "standard"
+
+
+class SFFieldGroupTypes(Enum):
+    all = "all"
+    custom = "custom"
+    standard = "standard"
+    required = "required"
+
+
+class ExtractDeclaration(HashableBaseModel):
+    where: T.Optional[str] = None
+    fields_: T.Union[T.List[str], str] = Field(["FIELDS(ALL)"], alias="fields")
+    api: DataApi = DataApi.SMART
+    sf_object: T.Optional[str] = None  # injected, not implied
+
+    @staticmethod
+    def parse_field_complex_type(fieldspec):
+        """If it's something like FIELDS(...), parse it out"""
+        if match := field_decl.match(fieldspec):
+            matching_group = match.groups()[0].lower()
+            field_group_type = getattr(SFFieldGroupTypes, matching_group, None)
+            return field_group_type
+        else:
+            return None
+
+    def assert_sf_object_fits_pattern(self):
+        if self.is_group:
+            assert (
+                self.group_type in SFObjectGroupTypes
+            ), f"Expected OBJECTS(ALL), OBJECTS(CUSTOM) or OBJECTS(STANDARD), not `{self.group_type.upper()}`"
+        else:
+            assert self.sf_object.isidentifier(), (
+                "Value should start with OBJECTS( or be a simple alphanumeric field name"
+                f" (underscores allowed) not {self.sf_object}"
+            )
+        return self.sf_object
+
+    def assert_check_where_against_complex(self):
+        """Check that a where clause was not used with a group declaration."""
+        assert not (
+            self.where and self.is_group
+        ), "Cannot specify a `where` clause on a declaration for multiple kinds of objects."
+
+    @validator("fields_")
+    def normalize_fields(cls, vals):
+        if isinstance(vals, str):
+            vals = [vals]
+        for val in vals:
+            cls.validate_field(val)
+        return vals
+
+    @classmethod
+    def validate_field(cls, val):
+        assert cls.parse_field_complex_type(val) or val.isidentifier(), val
+        if group_type := cls.parse_field_complex_type(val):
+            assert group_type, (
+                "group_type in OBJECT(group_type) should be one of "
+                f"{tuple(SFFieldGroupTypes.__members__.keys())}, "
+                f"not {val}"
+            )
+
+        return val
+
+    @property
+    def is_group(self):
+        return bool(self.group_type)
+
+    @property
+    def group_type(self):
+        """If it's something like OBJECT(...), parse it out"""
+        val = self._parse_group_type(self.sf_object)
+        if val:
+            group_type = getattr(SFObjectGroupTypes, val, None)
+            assert group_type, (
+                "group_type in OBJECT(group_type) should be one of "
+                f"{tuple(SFObjectGroupTypes.__members__.keys())}, "
+                f"not {val}"
+            )
+            return group_type
+
+    @staticmethod
+    def _parse_group_type(val):
+        if "(" in val:
+            return object_decl.match(val)[1].lower()
+        else:
+            return None
+
+
+class ExtractRulesFile(CCIDictModel):
+    version: int = 1
+    extract: T.Dict[str, ExtractDeclaration]
+
+    @validator("extract")
+    def inject_sf_object_name(cls, val):
+        for sf_object, decl in val.items():
+            decl.sf_object = sf_object
+            decl.assert_sf_object_fits_pattern()
+            decl.assert_check_where_against_complex()
+
+        return val
+
+    @classmethod
+    def parse_extract(cls, source: T.Union[str, Path, T.IO]):
+        """Return the extract key after parsing an extract file."""
+        return super().parse_from_yaml(source).extract
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/hardcoded_default_declarations.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/hardcoded_default_declarations.py
new file mode 100644
index 000000000..3fa6f8220
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/hardcoded_default_declarations.py
@@ -0,0 +1,49 @@
+from .extract_yml import ExtractDeclaration
+
+_DEFAULT_DECLARATIONS = [
+    ExtractDeclaration(
+        sf_object="Account", where="Name != 'Sample Account for Entitlements'"
+    ),
+    ExtractDeclaration(sf_object="BusinessHours", where="Name != 'Default'"),
+    ExtractDeclaration(sf_object="ContentWorkspace", where="Name != 'Asset Library'"),
+    ExtractDeclaration(sf_object="Entitlement", where="Name != 'Sample Entitlement'"),
+    ExtractDeclaration(
+        sf_object="FieldServiceMobileSettings",
+        where="DeveloperName != 'Field_Service_Mobile_Settings'",
+    ),
+    ExtractDeclaration(
+        sf_object="PricebookEntry",
+        where="Pricebook2.Id != NULL and Pricebook2.Name != 'Standard Price Book'",
+    ),
+    ExtractDeclaration(sf_object="Pricebook2", where="Name != 'Standard Price Book'"),
+    ExtractDeclaration(
+        sf_object="WebLink", where="Name != 'ViewCampaignInfluenceReport'"
+    ),
+    ExtractDeclaration(
+        sf_object="Folder",
+        where="DeveloperName NOT IN ('SharedApp', 'EinsteinBotReports')",
+    ),
+    ExtractDeclaration(
+        sf_object="MilestoneType",
+        where="Name NOT IN ('First Response to Customer', 'Escalate Case', 'Close Case')",
+    ),
+    ExtractDeclaration(
+        sf_object="WorkBadgeDefinition",
+        where="Name NOT IN ('Thanks', 'You\\'re a RockStar!', 'Team Player',"
+        "'All About Execution', 'Teacher', 'Top Performer', 'Hot Lead',"
+        "'Key Win', 'Customer Hero', 'Competition Crusher',"
+        "'Deal Maker', 'Gold Star')",
+    ),
+    ExtractDeclaration(
+        sf_object="EmailTemplate",
+        where="DeveloperName NOT IN ('CommunityLockoutEmailTemplate',"
+        "'CommunityVerificationEmailTemplate',"
+        "'CommunityChgEmailVerOldTemplate',"
+        "'CommunityChgEmailVerNewTemplate',"
+        "'CommunityDeviceActEmailTemplate',"
+        "'CommunityWelcomeEmailTemplate',"
+        "'CommunityChangePasswordEmailTemplate',"
+        "'CommunityForgotPasswordEmailTemplate' )",
+    ),
+]
+DEFAULT_DECLARATIONS = {decl.sf_object: decl for decl in _DEFAULT_DECLARATIONS}
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/synthesize_extract_declarations.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/synthesize_extract_declarations.py
new file mode 100644
index 000000000..97d27f0b2
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/synthesize_extract_declarations.py
@@ -0,0 +1,258 @@
+import collections
+import re
+import typing as T
+
+from pydantic import validator
+
+from cumulusci.salesforce_api.org_schema import NOT_EXTRACTABLE, Field, Schema
+from cumulusci.utils.iterators import partition
+
+from .extract_yml import ExtractDeclaration, SFFieldGroupTypes, SFObjectGroupTypes
+from .hardcoded_default_declarations import DEFAULT_DECLARATIONS
+
+
+class SimplifiedExtractDeclaration(ExtractDeclaration):
+    # a model where sf_object references a single sf_object
+    # and every field is a single field name, rather
+    # than a group declaration like FIELDS(xxx)
+    @validator("sf_object")
+    def sf_object_fits_pattern(cls, val):
+        assert val.isidentifier()
+        return val
+
+    @classmethod
+    def from_template_and_fields(
+        cls, template: ExtractDeclaration, fields: T.List[str]
+    ):
+        """Generate a simplified declaration with specified properties and fields"""
+        data = dict(template)
+        data["fields"] = fields
+        del data["fields_"]
+        return cls(**data)
+
+
+def flatten_declarations(
+    declarations: T.Iterable[ExtractDeclaration], schema: Schema
+) -> T.List[SimplifiedExtractDeclaration]:
+    """Convert short-form, abstract Extract declarations like this:
+
+    OBJECTS(CUSTOM):
+        fields: FIELDS(REQUIRED)
+
+    to concrete ones like this (a "Simplified" declaration):
+
+    Custom__c:
+        fields:
+            - Name
+            - CustomField__c
+    Custom2__c:
+        fields:
+            - Name
+            - CustomField2__c
+
+    Also detects dependencies between tables and pulls in required fields
+    from referenced tables recursively.
+    """
+    assert schema.includes_counts, "Schema object was not set up with `includes_counts`"
+    merged_declarations = _simplify_sfobject_declarations(declarations, schema)
+    simplified_declarations = [
+        _expand_field_definitions(decl, schema[decl.sf_object].fields)
+        for decl in merged_declarations
+    ]
+    from .calculate_dependencies import extend_declarations_to_include_referenced_tables
+
+    simplified_declarations = extend_declarations_to_include_referenced_tables(
+        simplified_declarations, schema
+    )
+
+    return simplified_declarations
+
+
+def _simplify_sfobject_declarations(declarations, schema: Schema):
+    """Generate a new list of declarations such that all sf_object patterns
+    (like OBJECTS(CUSTOM)) have been expanded into many declarations
+    with specific names and defaults have been merged in."""
+    simple_declarations, group_declarations = partition(
+        lambda d: d.is_group, declarations
+    )
+    simple_declarations = list(simple_declarations)
+    simple_declarations = _normalize_user_supplied_simple_declarations(
+        simple_declarations, DEFAULT_DECLARATIONS
+    )
+    simple_declarations = _merge_group_declarations_with_simple_declarations(
+        simple_declarations, group_declarations, schema
+    )
+    return simple_declarations
+
+
+def _merge_group_declarations_with_simple_declarations(
+    simple_declarations: T.Iterable[ExtractDeclaration],
+    group_declarations: T.Iterable[ExtractDeclaration],
+    schema: Schema,
+) -> T.List[ExtractDeclaration]:
+    """Expand group declarations to simple declarations and merge
+    with existing simple declarations"""
+    simple_declarations = list(simple_declarations)
+    group_declarations = list(group_declarations)
+
+    specific_sobject_decl_names = [obj.sf_object for obj in simple_declarations]
+
+    simplified_declarations = [
+        _expand_group_sobject_declaration(decl, schema) for decl in group_declarations
+    ]
+    for decl_set in simplified_declarations:
+        for decl in decl_set:
+            if decl.sf_object not in specific_sobject_decl_names and not any(
+                re.match(pat, decl.sf_object, re.IGNORECASE) for pat in NOT_EXTRACTABLE
+            ):
+                simple_declarations.append(decl)
+
+    return simple_declarations
+
+
+def _expand_group_sobject_declaration(decl: ExtractDeclaration, schema: Schema):
+    """Expand a group sobject declaration to a list of simple declarations"""
+    if decl.group_type == SFObjectGroupTypes.standard:
+
+        def matches_obj(obj):
+            return not obj.custom
+
+    elif decl.group_type == SFObjectGroupTypes.custom:
+
+        def matches_obj(obj):
+            return obj.custom
+
+    elif decl.group_type == SFObjectGroupTypes.all:
+
+        def matches_obj(obj):
+            return True
+
+    else:  # pragma: no cover
+        assert 0, decl.group_type
+
+    matching_objects = [
+        obj["name"] for obj in schema.sobjects if matches_obj(obj) and obj.count > 1
+    ]
+    decls = [
+        synthesize_declaration_for_sobject(obj, decl.fields) for obj in matching_objects
+    ]
+
+    return decls
+
+
+def _expand_field_definitions(
+    sobject_decl: ExtractDeclaration, schema_fields: T.Dict[str, Field]
+) -> SimplifiedExtractDeclaration:
+    """Expand group declarations to concrete ones. e.g. FIELDS(STANDARD) -> "LastName",
+
+    Also include all required fields whether asked for or not.
+    """
+    simple_declarations, group_declarations = partition(
+        lambda d: "(" in d, sobject_decl.fields
+    )
+    declarations = list(simple_declarations)
+
+    # expand group declarations to concrete ones.
+    # e.g. FIELDS(STANDARD) -> "LastName",
+    for c in group_declarations:
+        declarations.extend(_find_matching_field_declarations(c, schema_fields))
+
+    # add in all of the required fields
+    declarations.extend(
+        field.name
+        for field in schema_fields.values()
+        if (field.createable and not field.nillable and field.name not in declarations)
+    )
+    # get rid of OwnerId because we don't move users between orgs
+    if "OwnerId" in declarations:
+        index = declarations.index("OwnerId")
+        del declarations[index]
+
+    return SimplifiedExtractDeclaration.from_template_and_fields(
+        sobject_decl, fields=declarations
+    )
+
+
+def _find_matching_field_declarations(
+    field_group_type: str, schema_fields: T.Dict[str, Field]
+) -> T.Iterable[str]:
+    """Look in schema for field declarations matching a pattern like "Custom", "Standard", etc."""
+    ctype = ExtractDeclaration.parse_field_complex_type(field_group_type)
+    assert ctype, f"Could not parse {field_group_type}"  # Should be impossible
+
+    if ctype == SFFieldGroupTypes.standard:
+        # find updateable standard fields
+        return (
+            field.name
+            for field in schema_fields.values()
+            if field.createable and not field.custom
+        )
+    elif ctype == SFFieldGroupTypes.custom:
+        return (
+            field.name
+            for field in schema_fields.values()
+            if field.createable and field.custom
+        )
+    elif ctype == SFFieldGroupTypes.required:
+        # required fields are always exported
+        return ()
+    elif ctype == SFFieldGroupTypes.all:
+        return (field.name for field in schema_fields.values() if field.createable)
+    else:  # pragma: no cover
+        raise NotImplementedError(type)
+
+
+def synthesize_declaration_for_sobject(
+    sf_object: str, fields: list
+) -> SimplifiedExtractDeclaration:
+    """Fake a declaration for an sobject that was mentioned
+    indirectly"""
+    default = DEFAULT_DECLARATIONS.get(sf_object)
+    if default:
+        return SimplifiedExtractDeclaration.from_template_and_fields(default, fields)
+    else:
+        return SimplifiedExtractDeclaration(sf_object=sf_object, fields=fields)
+
+
+def _normalize_user_supplied_simple_declarations(
+    simple_declarations: T.List[ExtractDeclaration],
+    default_declarations: T.Mapping[str, ExtractDeclaration],
+) -> T.List[ExtractDeclaration]:
+    """Merge info provided by the user with things we already know about each SObject
+
+    For example, if we extract WorkBadgeDefinition, don't extract the
+    built-in ones.
+    """
+    duplicates = _find_duplicates(simple_declarations, lambda x: x.sf_object)
+
+    assert not duplicates, f"Duplicate declarations not allowed: {duplicates}"
+    simple_declarations = {
+        decl.sf_object: _merge_declarations_with_defaults(
+            decl, default_declarations.get(decl.sf_object, _DEFAULT_DEFAULTS)
+        )
+        for decl in simple_declarations
+    }
+    return list(simple_declarations.values())
+
+
+def _find_duplicates(input: T.Iterable[T.Tuple], key: T.Callable[[str], str]):
+    """Find duplicates in an iterable"""
+    counts = collections.Counter((key(v), v) for v in input)
+    duplicates = [name for name, count in counts.items() if count > 1]
+    return duplicates
+
+
+def _merge_declarations_with_defaults(
+    user_decl: ExtractDeclaration, default_decl: ExtractDeclaration
+):
+    """Merge two declarations with one taking priority over the other"""
+    default_decl = default_decl or _DEFAULT_DEFAULTS
+    return ExtractDeclaration(
+        sf_object=user_decl.sf_object,
+        where=user_decl.where or default_decl.where,
+        fields=user_decl.fields or default_decl.fields,
+        api=user_decl.api,
+    )
+
+
+_DEFAULT_DEFAULTS = ExtractDeclaration(fields="FIELDS(REQUIRED)")


Test patch:
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_extract_yml.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_extract_yml.py
new file mode 100644
index 000000000..8daf124e2
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_extract_yml.py
@@ -0,0 +1,142 @@
+from io import StringIO
+
+import pytest
+from pydantic import ValidationError
+
+from cumulusci.tasks.bulkdata.extract_dataset_utils.extract_yml import (
+    ExtractDeclaration,
+    ExtractRulesFile,
+)
+
+
+class TestExtractYML:
+    def test_simple_parse(self):
+        yaml = """
+            version: 1
+            extract:
+                Opportunity:
+                    fields:
+                        - Name
+                        - ContactId
+                        - AccountId
+                OBJECTS(CUSTOM):
+                    fields: FIELDS(CUSTOM)
+                OBJECTS(STANDARD):
+                    fields: FIELDS(CUSTOM)
+        """
+        parse_result = ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+        assert parse_result.version == 1
+        extract = parse_result.extract
+        assert tuple(extract.keys()) == (
+            "Opportunity",
+            "OBJECTS(CUSTOM)",
+            "OBJECTS(STANDARD)",
+        )
+        assert extract["OBJECTS(STANDARD)"].fields == ["FIELDS(CUSTOM)"]
+        assert extract["OBJECTS(STANDARD)"].sf_object == "OBJECTS(STANDARD)"
+        assert extract["OBJECTS(STANDARD)"].group_type.value == "standard"
+        assert extract["Opportunity"].group_type is None
+
+    def test_default_version(self):
+        yaml = """
+            extract:
+                Opportunity:
+                    fields:
+                        - Name
+                        - ContactId
+                        - AccountId
+        """
+        parse_result = ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+        assert parse_result.version == 1
+        extract = parse_result.extract
+        assert tuple(extract.keys()) == ("Opportunity",)
+
+    def test_mini_parser(self):
+        assert (
+            ExtractDeclaration.parse_field_complex_type("FIELDS(STANDARD)").value
+            == "standard"
+        )
+        assert ExtractDeclaration.parse_field_complex_type("ffff") is None
+
+    def test_weird_syntaxes(self):
+        yaml = """
+            extract:
+                OBJECTS(XYZZY):
+                    fields: FIELDS(CUSTOM)
+        """
+        with pytest.raises(ValidationError, match="xyzzy"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+
+        yaml = """
+            extract:
+                OBJECTS(XYZZY):
+                    fields:
+                        - FIELDS(CUSTOM)
+        """
+        with pytest.raises(ValidationError, match="xyzzy"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+        yaml = """
+            extract:
+                OBJECTS(POPULATED):
+                    fields: FIELDS(XYZZY)
+        """
+        with pytest.raises(ValidationError, match="XYZZY"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+
+        yaml = """
+            extract:
+                OBJECTS(POPULATED):
+                    fields:
+                        - FIELDS(XYZZY)
+        """
+        with pytest.raises(ValidationError, match="XYZZY"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+
+        yaml = """
+            extract:
+                OBJECTS(POPULATED):
+                    fields:
+                        - abcde
+                        - FIELDS(XYZZY)
+        """
+        with pytest.raises(ValidationError, match="XYZZY"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+
+        yaml = """
+            extract:
+                abcd.efgh:
+                    fields:
+                        - Id
+        """
+        with pytest.raises(ValidationError, match="abcd.efgh"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+
+        yaml = """
+            extract:
+                OBJECTS(POPULATED):
+                    fields: XYZZY(CUSTOM)
+        """
+        with pytest.raises(ValidationError, match="XYZZY"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+
+        yaml = """
+            extract:
+                OBJECTS(POPULATED):
+                    fields:
+                        - XYZZY(CUSTOM)
+        """
+        with pytest.raises(ValidationError, match="XYZZY"):
+            ExtractRulesFile.parse_from_yaml(StringIO(yaml))
+
+    def test_parse_real_file(self, cumulusci_test_repo_root):
+        parse_result = ExtractRulesFile.parse_from_yaml(
+            cumulusci_test_repo_root / "datasets/test.extract.yml"
+        )
+        assert parse_result.version == 1
+        extract = parse_result.extract
+        assert tuple(extract.keys()) == (
+            "Opportunity",
+            "Contact",
+            "Account",
+            "OBJECTS(CUSTOM)",
+        )
diff --git a/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_synthesize_extract_declarations.py b/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_synthesize_extract_declarations.py
new file mode 100644
index 000000000..c359a1c3c
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/extract_dataset_utils/tests/test_synthesize_extract_declarations.py
@@ -0,0 +1,421 @@
+import json
+import typing as T
+from contextlib import contextmanager
+from functools import lru_cache
+from io import StringIO
+from pathlib import Path
+from unittest import mock
+
+from sqlalchemy import create_engine
+
+from cumulusci.core.config import OrgConfig
+from cumulusci.salesforce_api.org_schema import get_org_schema
+from cumulusci.tasks.bulkdata.extract_dataset_utils.extract_yml import ExtractRulesFile
+from cumulusci.tasks.bulkdata.extract_dataset_utils.synthesize_extract_declarations import (
+    flatten_declarations,
+)
+from cumulusci.tasks.bulkdata.step import DataApi
+from cumulusci.tests.util import read_mock
+from cumulusci.utils.tests.test_org_schema import FakeSF
+
+
+class TestSynthesizeExtractDeclarations:
+    def test_synthesize_extract_declarations(self, org_config):
+        declarations = """
+            extract:
+                    OBJECTS(ALL):
+                        fields:
+                            FIELDS(REQUIRED)
+        """
+        object_counts = {"Account": 10, "Contact": 0, "Case": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+        )
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+
+            assert tuple(decl.dict() for decl in decls) == tuple(
+                (
+                    {
+                        "where": "Name != 'Sample Account for Entitlements'",
+                        "fields_": mock.ANY,
+                        "api": DataApi.SMART,
+                        "sf_object": "Account",
+                    },
+                    {
+                        "where": None,
+                        "fields_": mock.ANY,
+                        "api": DataApi.SMART,
+                        "sf_object": "Case",
+                    },
+                )
+            )
+
+    def test_find_custom_objects(self, org_config):
+        declarations = """
+            extract:
+                    OBJECTS(CUSTOM):
+                        fields:
+                            FIELDS(ALL)
+        """
+        object_counts = {"Account": 10, "Contact": 0, "Case": 5, "Custom__c": 10}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+            describe_for("Custom__c"),
+        )
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+
+            assert tuple(decl.dict() for decl in decls) == tuple(
+                (
+                    {
+                        "where": None,
+                        "fields_": mock.ANY,
+                        "api": DataApi.SMART,
+                        "sf_object": "Custom__c",
+                    },
+                )
+            )
+
+    def test_find_standard_objects(self, org_config):
+        declarations = """
+            extract:
+                    OBJECTS(STANDARD):
+                        fields:
+                            FIELDS(REQUIRED)
+        """
+        object_counts = {"Account": 10, "Contact": 0, "Case": 5, "Custom__c": 10}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Case"),
+            describe_for("Custom__c"),
+        )
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            filters=[],
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+
+            assert tuple(decl.dict() for decl in decls) == tuple(
+                (
+                    {
+                        "where": "Name != 'Sample Account for Entitlements'",
+                        "fields_": mock.ANY,
+                        "api": DataApi.SMART,
+                        "sf_object": "Account",
+                    },
+                    {
+                        "where": None,
+                        "fields_": mock.ANY,
+                        "api": DataApi.SMART,
+                        "sf_object": "Case",
+                    },
+                )
+            )
+
+    def test_synthesize_extract_declarations__custom(self, org_config):
+        declarations = """
+            extract:
+                    OBJECTS(CUSTOM):
+                        fields:
+                            FIELDS(ALL)
+        """
+        object_counts = {"Account": 0, "Contact": 2, "Custom__c": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Custom__c"),
+        )
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+
+            assert tuple(decl.dict() for decl in decls) == tuple(
+                (
+                    {
+                        "where": None,
+                        "fields_": mock.ANY,
+                        "api": DataApi.SMART,
+                        "sf_object": "Custom__c",
+                    },
+                )
+            )
+
+    def test_synthesize_extract_declarations__custom_fields(self, org_config):
+        declarations = """
+            extract:
+                    OBJECTS(CUSTOM):
+                        fields:
+                            FIELDS(custom)
+        """
+        object_counts = {"Account": 0, "Contact": 2, "Custom__c": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Custom__c"),
+        )
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+
+            assert tuple(decl.dict() for decl in decls) == tuple(
+                (
+                    {
+                        "where": None,
+                        "fields_": ["CustomField__c"],
+                        "api": DataApi.SMART,
+                        "sf_object": "Custom__c",
+                    },
+                )
+            )
+
+    def test_synthesize_extract_declarations__standard_fields(self, org_config):
+        declarations = """
+            extract:
+                    OBJECTS(CUSTOM):
+                        fields:
+                            FIELDS(standard)
+        """
+        object_counts = {"Account": 0, "Contact": 2, "Custom__c": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Custom__c"),
+        )
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+
+            assert tuple(decl.dict() for decl in decls) == tuple(
+                (
+                    {
+                        "where": None,
+                        "fields_": ["Name"],
+                        "api": DataApi.SMART,
+                        "sf_object": "Custom__c",
+                    },
+                )
+            )
+
+    def test_required_lookups__pulled_in(self, org_config):
+        """Bringing in the AccountId should force Account to come in.
+
+        Including all Account/Contact required fields."""
+        declarations = """
+            extract:
+                Contact:
+                    fields:
+                        AccountId
+        """
+        object_counts = {"Account": 3, "Contact": 2, "Custom__c": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Custom__c"),
+        )
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+
+            assert tuple(decl.dict() for decl in decls) == tuple(
+                (
+                    {
+                        "where": mock.ANY,
+                        "fields_": ["AccountId", "LastName"],
+                        "api": DataApi.SMART,
+                        "sf_object": "Contact",
+                    },
+                    {
+                        "where": mock.ANY,
+                        "fields_": [
+                            "Name",
+                        ],
+                        "api": DataApi.SMART,
+                        "sf_object": "Account",
+                    },
+                )
+            )
+
+    def test_parse_real_file(self, cumulusci_test_repo_root, org_config):
+        declarations = ExtractRulesFile.parse_extract(
+            cumulusci_test_repo_root / "datasets/test_minimal.extract.yml"
+        )
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Opportunity"),
+            describe_for("Custom__c"),
+        )
+        object_counts = {
+            "Account": 100,
+            "Opportunity": 10,
+            "Custom__c": 5,
+            "Contact": 0,
+        }
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+            decls = {decl.sf_object: decl for decl in decls}
+            assert decls["Opportunity"].fields == [
+                "Name",
+                "ContactId",
+                "AccountId",
+                "CloseDate",  # pull these in because they required
+                "IsPrivate",
+                "StageName",
+            ]
+            assert "Account" in decls
+            assert "Contact" in decls  # pulled in because of ContactId
+            assert "Custom__c" in decls
+            assert set(decls["Custom__c"].fields) == set(
+                schema["Custom__c"].fields.keys()
+            ) - set(["Id"])
+
+    def test_synthesize_extract_declarations__where_clause(self, org_config):
+        declarations = """
+            extract:
+                Contact:
+                    fields:
+                        FIELDS(REQUIRED)
+                Account:
+                    where: Name Like '%Foo%'
+                    fields:
+                        - Description
+
+        """
+        declarations = ExtractRulesFile.parse_extract(StringIO(declarations))
+        object_counts = {"Account": 2, "Contact": 2, "Custom__c": 5}
+        obj_describes = (
+            describe_for("Account"),
+            describe_for("Contact"),
+            describe_for("Custom__c"),
+        )
+        with _fake_get_org_schema(
+            org_config,
+            obj_describes,
+            object_counts,
+            include_counts=True,
+        ) as schema:
+            decls = flatten_declarations(declarations.values(), schema)
+            decls = {decl.sf_object: decl for decl in decls}
+            assert "Account" in decls
+            assert "Contact" in decls
+            assert "Custom__c" not in decls
+
+            assert set(decls["Account"].fields) == set(["Name", "Description"])
+
+
+@lru_cache(maxsize=None)
+def describe_for(sobject: str):
+    return json.loads(read_mock(sobject))
+
+
+@contextmanager
+def _fake_get_org_schema(
+    org_config: OrgConfig,
+    org_describes: T.List[dict],
+    object_counts: T.Dict[str, int],
+    **kwargs,
+):
+    with mock.patch(
+        "cumulusci.salesforce_api.org_schema.count_sobjects",
+        lambda *args: (
+            object_counts,
+            [],
+            [],
+        ),
+    ), mock.patch(
+        "cumulusci.salesforce_api.org_schema.ZippableTempDb", FakeZippableTempDb
+    ), mock.patch(
+        "cumulusci.salesforce_api.org_schema.deep_describe",
+        return_value=((desc, "Sat, 1 Jan 2000 00:00:01 GMT") for desc in org_describes),
+    ), get_org_schema(
+        FakeSF(), org_config, **kwargs
+    ) as schema:
+        yield schema
+
+
+class FakeZippableTempDb:
+    "Fast no-IO database for testing"
+
+    def __enter__(self, *args, **kwargs):
+        return self
+
+    def __exit__(self, *args, **kwargs):
+        return ""
+
+    def zip_database(self, target_path: Path):
+        pass
+
+    def unzip_database(self, zipped_db: Path):
+        pass
+
+    def clear(self):
+        pass
+
+    def create_engine(self):
+        self.engine = create_engine("sqlite:///")
+        return self.engine
+
+
+@contextmanager
+def faketempdb():
+    yield Path("")
+
+
+# TODO: Decide upon what happens in this cases:
+#       OBJECTS(STANDARD)
+#          fields:
+#           - X
+#       Account:
+#          fields:
+#           - Y
+#    Do you get X and Y? (Union semantics?) Or just Y (override semantics)
+# Consider all kinds of conflicting/complementary declarations.
diff --git a/datasets/test.extract.yml b/datasets/test.extract.yml
new file mode 100644
index 000000000..dcf4b1ed4
--- /dev/null
+++ b/datasets/test.extract.yml
@@ -0,0 +1,13 @@
+version: 1
+extract:
+    Opportunity: # Extract these Opp fields plus all required fields
+        fields:
+            - Name
+            - ContactId
+            - AccountId
+    Contact: # Extract all Contact fields
+        fields: FIELDS(ALL)
+    Account: # Extract all Account fields
+        fields: FIELDS(ALL)
+    OBJECTS(CUSTOM):
+        fields: FIELDS(ALL)
diff --git a/datasets/test_minimal.extract.yml b/datasets/test_minimal.extract.yml
new file mode 100644
index 000000000..8828b813b
--- /dev/null
+++ b/datasets/test_minimal.extract.yml
@@ -0,0 +1,13 @@
+version: 1
+extract:
+    Opportunity: # Extract these Opp fields plus all required fields
+        fields:
+            - Name
+            - ContactId
+            - AccountId
+    Contact: # Extract required Contact fields
+        fields: FIELDS(REQUIRED)
+    Account: # Extract required Account fields
+        fields: FIELDS(REQUIRED)
+    OBJECTS(CUSTOM): # Extract every field on Custom objects
+        fields: FIELDS(ALL)