Repository: SFDO-Tooling/CumulusCI
Pull request #1618 (2020-03-16 18:02:54)

Pydantic-based task option parsing.
Opt-in pydantic-based task option parsing.


Br4nd0R: @davisagli - I see that you had brought up the issue of wanting to ensure context is preserved so we know what task and option are the culprits in case of an error when options are being passed via `cumulusci.yml`. I did some testing and confirmed with @prescod that it should be fairly apparent during flow execution with our current logging which task failed, and the current error messaging displays the culprit option.

We don't have access to the actual name of the task being run when validating the options; however, we do have the ability to display the `class_path` of the currently executing task.

Let me know if you can think of any other scenarios where the currently executing task may be ambiguous, we definitely could have missed an edge case somewhere.


prescod: @davisagli This one is ready for you.
jstvz: Closing since it's been more than a year since the last activity. Feel free to re-open as needed.

Patch:
diff --git a/cumulusci/core/tasks.py b/cumulusci/core/tasks.py
index 2ab7a68d2..c2e95c9db 100644
--- a/cumulusci/core/tasks.py
+++ b/cumulusci/core/tasks.py
@@ -9,7 +9,9 @@ import re
 import threading
 import time
 from contextlib import nullcontext
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Type, Union
+
+from pydantic.error_wrappers import ValidationError
 
 from cumulusci import __version__
 from cumulusci.core.config import TaskConfig
@@ -25,6 +27,8 @@ from cumulusci.core.exceptions import (
 from cumulusci.core.flowrunner import FlowCoordinator, StepSpec, StepVersion
 from cumulusci.utils import cd
 from cumulusci.utils.logging import redirect_output_to_logger
+from cumulusci.utils.metaprogramming import classproperty
+from cumulusci.utils.options import CCIOptions
 
 CURRENT_TASK = threading.local()
 
@@ -51,7 +55,7 @@ class BaseTask:
     """
 
     task_docs: str = ""
-    task_options: dict = {}
+    Options: Optional[Type[CCIOptions]] = None
     salesforce_task: bool = False  # Does this task require a salesforce org?
     name: Optional[str]
     stepnum: Optional[StepVersion]
@@ -110,6 +114,16 @@ class BaseTask:
         else:
             self.logger = logging.getLogger(__name__)
 
+    @classproperty
+    def task_options(cls):  # type: ignore  -- doesn't like the signature override below
+        "Convert Options dict into old fashioned task_options syntax"
+        if cls.Options:
+            return cls.Options.as_task_options()
+        else:
+            return {}
+
+    task_options: Union[classproperty, Dict]
+
     def _init_options(self, kwargs):
         """Initializes self.options"""
         if self.task_config.options is None:
@@ -129,6 +143,32 @@ class BaseTask:
                 )
                 self.options[option] = value
 
+        if self.Options:
+            try:
+                specials = ["debug_before", "debug_after", "no_prompt"]
+                options_without_specials = {
+                    opt: val for opt, val in self.options.items() if opt not in specials
+                }
+                self.parsed_options = self.Options(**options_without_specials)
+            except ValidationError as e:
+                try:
+                    errors = [
+                        f"Error in '{error['loc'][0]}' option: '{error['msg']}'"
+                        for error in e.errors()
+                    ]
+                    plural = "s" if len(errors) > 1 else ""
+                    errorstrs = ", ".join(errors)
+                    message = f"Task Options Error{plural}: {errorstrs}"
+                except (AttributeError, IndexError):
+                    message = f"Task Options Error: {e.errors()}"
+                if "extra fields not permitted" in message:
+                    message = message.replace("extra fields", "extra options")
+                raise TaskOptionsError(message) from e
+            except (TaskOptionsError, TypeError) as e:
+                raise TaskOptionsError(
+                    f"Task Options Error: Error in '{self.options}' option: '{e}'"
+                )
+
     def _validate_options(self):
         missing_required = []
         for name, config in list(self.task_options.items()):
@@ -252,7 +292,10 @@ class BaseTask:
             "is_required": True,
         }
         ui_step.update(self.task_config.config.get("ui_options", {}))
-        task_config = {"options": self.options, "checks": self.task_config.checks or []}
+        task_config = {
+            "options": self.options,
+            "checks": self.task_config.checks or [],
+        }
         ui_step.update(
             {
                 "path": step.path,
diff --git a/cumulusci/core/utils.py b/cumulusci/core/utils.py
index 727ccb935..88cd57065 100644
--- a/cumulusci/core/utils.py
+++ b/cumulusci/core/utils.py
@@ -18,6 +18,7 @@ from cumulusci.core.exceptions import (
     CumulusCIException,
     TaskOptionsError,
 )
+from cumulusci.utils.options import parse_list_of_pairs_dict_arg
 
 
 def import_global(path: str):
@@ -136,22 +137,10 @@ def process_list_arg(arg):
 
 def process_list_of_pairs_dict_arg(arg):
     """Process an arg in the format "aa:bb,cc:dd" """
-    if isinstance(arg, dict):
-        return arg
-    elif isinstance(arg, str):
-        rc = {}
-        for key_value in arg.split(","):
-            subparts = key_value.split(":", 1)
-            if len(subparts) == 2:
-                key, value = subparts
-                if key in rc:
-                    raise TaskOptionsError(f"Var specified twice: {key}")
-                rc[key] = value
-            else:
-                raise TaskOptionsError(f"Var is not a name/value pair: {key_value}")
-        return rc
-    else:
-        raise TaskOptionsError(f"Arg is not a dict or string ({type(arg)}): {arg}")
+    try:
+        return parse_list_of_pairs_dict_arg(arg)
+    except TypeError as e:
+        raise TaskOptionsError(e) from e
 
 
 def decode_to_unicode(content):
@@ -230,7 +219,8 @@ def remove_overridden_flow_steps_in_config(
 ):
     """If any steps of flows from the config_to_override are being overridden in overriding_config,
     then we need to set those steps in the config_to_override to an empty dict so that we don't have
-    both a "task" and a "flow" listed in a flow step after merging the configs with `dictmerge()`."""
+    both a "task" and a "flow" listed in a flow step after merging the configs with `dictmerge()`.
+    """
 
     if "flows" not in config_to_override or "flows" not in overriding_config:
         return
diff --git a/cumulusci/tasks/github/commit_status.py b/cumulusci/tasks/github/commit_status.py
index 6277664b0..7793fd802 100644
--- a/cumulusci/tasks/github/commit_status.py
+++ b/cumulusci/tasks/github/commit_status.py
@@ -5,7 +5,6 @@ from cumulusci.tasks.salesforce.BaseSalesforceApiTask import BaseSalesforceApiTa
 
 
 class GetPackageDataFromCommitStatus(BaseGithubTask, BaseSalesforceApiTask):
-
     api_version = "52.0"
 
     task_options = {
diff --git a/cumulusci/tasks/github/merge.py b/cumulusci/tasks/github/merge.py
index 798f91bdb..1134de004 100644
--- a/cumulusci/tasks/github/merge.py
+++ b/cumulusci/tasks/github/merge.py
@@ -18,7 +18,7 @@ class MergeBranch(BaseGithubTask):
 
     If source_branch begins with branch_prefix, then the commit is merged to all child branches of source_branch.
     """
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "commit": {
             "description": "The commit to merge into feature branches.  Defaults to the current head commit."
         },
@@ -112,7 +112,6 @@ class MergeBranch(BaseGithubTask):
         main_descendents = []
         release_branches = []
         for branch in repo_branches:
-
             # check for adding future release branches
             if update_future_releases and self._is_future_release_branch(
                 branch.name, next_release
diff --git a/cumulusci/tasks/github/publish.py b/cumulusci/tasks/github/publish.py
index 0c8f228f3..82b54123d 100644
--- a/cumulusci/tasks/github/publish.py
+++ b/cumulusci/tasks/github/publish.py
@@ -18,7 +18,7 @@ from cumulusci.utils.git import split_repo_url
 
 
 class PublishSubtree(BaseGithubTask):
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "repo_url": {"description": "The url to the public repo", "required": True},
         "branch": {
             "description": "The branch to update in the target repo",
diff --git a/cumulusci/tasks/github/release_report.py b/cumulusci/tasks/github/release_report.py
index 6e511a751..b88310653 100644
--- a/cumulusci/tasks/github/release_report.py
+++ b/cumulusci/tasks/github/release_report.py
@@ -6,7 +6,7 @@ from cumulusci.tasks.github.base import BaseGithubTask
 
 
 class ReleaseReport(BaseGithubTask):
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "date_start": {
             "description": "Filter out releases created before this date (YYYY-MM-DD)"
         },
diff --git a/cumulusci/tasks/github/tag.py b/cumulusci/tasks/github/tag.py
index 64d33ed80..6d54c7eec 100644
--- a/cumulusci/tasks/github/tag.py
+++ b/cumulusci/tasks/github/tag.py
@@ -5,8 +5,7 @@ from cumulusci.tasks.github.base import BaseGithubTask
 
 
 class CloneTag(BaseGithubTask):
-
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "src_tag": {
             "description": "The source tag to clone.  Ex: beta/1.0-Beta_2",
             "required": True,
diff --git a/cumulusci/tasks/marketing_cloud/api.py b/cumulusci/tasks/marketing_cloud/api.py
index 9ae269ac5..0b7792cd1 100644
--- a/cumulusci/tasks/marketing_cloud/api.py
+++ b/cumulusci/tasks/marketing_cloud/api.py
@@ -7,7 +7,7 @@ from .base import BaseMarketingCloudTask
 
 
 class CreateSubscriberAttribute(BaseMarketingCloudTask):
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "attribute_name": {
             "description": "The name of the Subscriber Attribute to deploy via the Marketing Cloud API.",
             "required": True,
@@ -37,7 +37,7 @@ class CreateSubscriberAttribute(BaseMarketingCloudTask):
 
 
 class CreateUser(BaseMarketingCloudTask):
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "parent_bu_mid": {
             "description": "Specify the MID for Parent BU.",
             "required": True,
@@ -126,7 +126,7 @@ class CreateUser(BaseMarketingCloudTask):
 
 
 class UpdateUserRole(BaseMarketingCloudTask):
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "account_mid": {
             "description": "Specify the Account MID.",
             "required": True,
diff --git a/cumulusci/tasks/metadata/ee_src.py b/cumulusci/tasks/metadata/ee_src.py
index 5d2bfc1a8..fab12dc55 100644
--- a/cumulusci/tasks/metadata/ee_src.py
+++ b/cumulusci/tasks/metadata/ee_src.py
@@ -9,7 +9,7 @@ from cumulusci.utils import remove_xml_element_directory
 
 
 class CreateUnmanagedEESrc(BaseTask):
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "path": {
             "description": "The path containing metadata to process for managed deployment",
             "required": True,
@@ -63,7 +63,7 @@ class CreateUnmanagedEESrc(BaseTask):
 
 
 class RevertUnmanagedEESrc(BaseTask):
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "path": {
             "description": "The path containing metadata to process for managed deployment",
             "required": True,
diff --git a/cumulusci/tasks/robotframework/lint.py b/cumulusci/tasks/robotframework/lint.py
index e8e7f6467..40726e26e 100644
--- a/cumulusci/tasks/robotframework/lint.py
+++ b/cumulusci/tasks/robotframework/lint.py
@@ -54,7 +54,7 @@ class RobotLint(BaseTask):
 
     """
 
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "configure": {
             "description": "List of rule configuration values, in the form of rule:args.",
             "default": None,
diff --git a/cumulusci/tasks/salesforce/SOQLQuery.py b/cumulusci/tasks/salesforce/SOQLQuery.py
index 0ed51fae3..96f5aa8ce 100644
--- a/cumulusci/tasks/salesforce/SOQLQuery.py
+++ b/cumulusci/tasks/salesforce/SOQLQuery.py
@@ -4,7 +4,7 @@ from cumulusci.tasks.salesforce.BaseSalesforceApiTask import BaseSalesforceApiTa
 class SOQLQuery(BaseSalesforceApiTask):
     name = "SOQLQuery"
 
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "object": {"required": True, "description": "The object to query"},
         "query": {
             "required": True,
diff --git a/cumulusci/tasks/salesforce/custom_settings.py b/cumulusci/tasks/salesforce/custom_settings.py
index 5f358c85c..b7a4b4093 100644
--- a/cumulusci/tasks/salesforce/custom_settings.py
+++ b/cumulusci/tasks/salesforce/custom_settings.py
@@ -37,7 +37,7 @@ class LoadCustomSettings(BaseSalesforceApiTask):
                      MyField__c: 2
     """
 
-    task_options = {
+    task_options = {  # TODO: should use `class Options instead`
         "settings_path": {
             "description": "The path to a YAML settings file",
             "required": True,
diff --git a/cumulusci/tasks/util.py b/cumulusci/tasks/util.py
index e47322dff..afb574dfc 100644
--- a/cumulusci/tasks/util.py
+++ b/cumulusci/tasks/util.py
@@ -7,49 +7,59 @@ from defusedxml.minidom import parse
 
 from cumulusci.core.exceptions import TaskOptionsError
 from cumulusci.core.tasks import BaseTask
-from cumulusci.core.utils import process_bool_arg, process_list_arg
 from cumulusci.utils import download_extract_zip, find_replace, find_replace_regex
+from cumulusci.utils.options import (
+    CCIOptions,
+    DirectoryPath,
+    Field,
+    ListOfStringsOption,
+)
 
 
 class DownloadZip(BaseTask):
     name = "Download"
-    task_options = {
-        "url": {"description": "The url of the zip file to download", "required": True},
-        "dir": {
-            "description": "The directory where the zip should be extracted",
-            "required": True,
-        },
-        "subfolder": {
-            "description": (
+
+    class Options(CCIOptions):
+        url: str = Field(..., description="The url of the zip file to download")
+        dir: str = Field(
+            ..., description="The directory where the zip should be extracted"
+        )
+        subfolder: str = Field(
+            None,
+            description=(
                 "The subfolder of the target zip to extract. Defaults to"
                 + " extracting the root of the zip file to the destination."
-            )
-        },
-    }
+            ),
+        )
+
+    parsed_options: Options
 
     def _run_task(self):
-        if not os.path.exists(self.options["dir"]):
-            os.makedirs(self.options["dir"])
+        if not os.path.exists(self.parsed_options.dir):
+            os.makedirs(self.parsed_options.dir)
 
         download_extract_zip(
-            self.options["url"], self.options["dir"], self.options.get("subfolder")
+            self.parsed_options.url,
+            self.parsed_options.dir,
+            self.parsed_options.subfolder,
         )
 
 
 class ListMetadataTypes(BaseTask):
     name = "ListMetadataTypes"
-    task_options = {
-        "package_xml": {
-            "description": (
+
+    class Options(CCIOptions):
+        package_xml: str = Field(
+            None,
+            description=(
                 "The project package.xml file."
                 + " Defaults to <project_root>/src/package.xml"
-            )
-        }
-    }
+            ),
+        )
 
     def _init_options(self, kwargs):
         super(ListMetadataTypes, self)._init_options(kwargs)
-        if "package_xml" not in self.options:
+        if not self.options.get("package_xml"):
             self.options["package_xml"] = os.path.join(
                 self.project_config.repo_root, "src", "package.xml"
             )
@@ -72,36 +82,47 @@ class ListMetadataTypes(BaseTask):
 
 class Sleep(BaseTask):
     name = "Sleep"
-    task_options = {
-        "seconds": {"description": "The number of seconds to sleep", "required": True}
-    }
+
+    class Options(CCIOptions):
+        seconds: int = Field(..., description="The number of seconds to sleep")
+
+    parsed_options: Options
 
     def _run_task(self):
-        self.logger.info("Sleeping for {} seconds".format(self.options["seconds"]))
-        time.sleep(float(self.options["seconds"]))
+        self.logger.info("Sleeping for {} seconds".format(self.parsed_options.seconds))
+        time.sleep(float(self.parsed_options.seconds))
         self.logger.info("Done")
 
 
 class Delete(BaseTask):
     name = "Delete"
-    task_options = {
-        "path": {
-            "description": "The path to delete.  If path is a directory, recursively deletes the directory: BE CAREFUL!!!  If path is a list, all paths will be deleted",
-            "required": True,
-        },
-        "chdir": {
-            "description": "Change directories before deleting path(s).  This is useful if you have a common list of relative paths to delete that you want to call against different directories."
-        },
-    }
+
+    class Options(CCIOptions):
+        path: ListOfStringsOption = Field(
+            default=...,
+            description=(
+                "The path to delete. "
+                "If path is a directory, recursively deletes the directory: BE CAREFUL!!! "
+                "If path is a list, all paths will be deleted"
+            ),
+        )
+        chdir: DirectoryPath = Field(
+            default=None,
+            description=(
+                "Change directories before deleting path(s). "
+                "This is useful if you have a common list of relative paths to "
+                "delete that you want to call against different directories."
+            ),
+        )
 
     def _run_task(self):
-        chdir = self.options.get("chdir")
+        chdir = self.parsed_options.chdir
         cwd = os.getcwd()
         if chdir:
             self.logger.info("Changing directory to {}".format(chdir))
             os.chdir(chdir)
 
-        path = self.options["path"]
+        path = self.parsed_options.path
         if not isinstance(path, list):
             path = [path]
         for path_item in path:
@@ -124,75 +145,68 @@ class Delete(BaseTask):
             os.remove(path)
 
 
-class FindReplace(BaseTask):
-    task_options = {
-        "find": {"description": "The string to search for", "required": True},
-        "replace": {
-            "description": "The string to replace matches with. Defaults to an empty string",
-            "required": True,
-        },
-        "env_replace": {
-            "description": "If True, treat the value of the replace option as the name of an environment variable, and use the value of that variable as the replacement string. Defaults to False",
-            "required": False,
-        },
-        "path": {"description": "The path to recursively search", "required": True},
-        "file_pattern": {
-            "description": "A UNIX like filename pattern used for matching filenames, or a list of them. See python fnmatch docs for syntax. If passed via command line, use a comma separated string. Defaults to *"
-        },
-        "max": {
-            "description": "The max number of matches to replace.  Defaults to replacing all matches."
-        },
-    }
+class FindReplaceOptions(CCIOptions):
+    find: str = Field(..., description="The string to search for")
+    replace: str = Field(
+        "",
+        description="The string to replace matches with. Defaults to an empty string",
+    )
+    path: DirectoryPath = Field(..., description="The path to recursively search")
+    file_pattern: ListOfStringsOption = Field(
+        "*",
+        description="A UNIX like filename pattern used for matching filenames, or a list of them. See python fnmatch docs for syntax. If passed via command line, use a comma separated string. Defaults to *",
+    )
+    env_replace: bool = Field(
+        False,
+        description="If True, treat the value of the replace option as the name of an environment variable, and use the value of that variable as the replacement string. Defaults to False",
+    )
 
-    def _init_options(self, kwargs):
-        super(FindReplace, self)._init_options(kwargs)
 
-        if "replace" not in self.options:
-            self.options["replace"] = ""
-        self.options["file_pattern"] = process_list_arg(
-            self.options.get("file_pattern") or "*"
-        )
-        self.options["env_replace"] = process_bool_arg(
-            self.options.get("env_replace") or False
+class FindReplace(BaseTask):
+    class Options(FindReplaceOptions):
+        max: int = Field(
+            None,
+            description="The max number of matches to replace.  Defaults to replacing all matches.",
         )
 
+    parsed_options: Options
+
     def _run_task(self):
         kwargs = {}
-        if "max" in self.options:
-            kwargs["max"] = self.options["max"]
-
-        if self.options["env_replace"]:
-            if self.options["replace"] in os.environ.keys():
-                self.options["replace"] = os.environ[self.options["replace"]]
+        if self.parsed_options.max:
+            kwargs["max"] = self.parsed_options.max
+
+        if self.parsed_options["env_replace"]:
+            if self.parsed_options["replace"] in os.environ.keys():
+                self.parsed_options["replace"] = os.environ[
+                    self.parsed_options["replace"]
+                ]
             else:
                 raise TaskOptionsError(
-                    f"The environment variable {self.options['replace']} was not found. Ensure that this value is populated or set env_replace to False."
+                    f"The environment variable {self.parsed_options['replace']} was not found. Ensure that this value is populated or set env_replace to False."
                 )
 
-        for file_pattern in self.options["file_pattern"]:
+        for file_pattern in self.parsed_options["file_pattern"]:
             find_replace(
-                find=self.options["find"],
-                replace=self.options["replace"],
-                directory=self.options["path"],
+                find=self.parsed_options.find,
+                replace=self.parsed_options.replace,
+                directory=self.parsed_options.path,
                 filePattern=file_pattern,
                 logger=self.logger,
                 **kwargs,
             )
 
 
-find_replace_regex_options = FindReplace.task_options.copy()
-del find_replace_regex_options["max"]
-
-
 class FindReplaceRegex(FindReplace):
-    task_options = find_replace_regex_options
+    class Options(FindReplaceOptions):
+        pass
 
     def _run_task(self):
         find_replace_regex(
-            find=self.options["find"],
-            replace=self.options["replace"],
-            directory=self.options["path"],
-            filePattern=self.options["file_pattern"],
+            find=self.parsed_options.find,
+            replace=self.parsed_options.replace,
+            directory=self.parsed_options.path,
+            filePattern=self.parsed_options.file_pattern,
             logger=self.logger,
         )
 
diff --git a/cumulusci/utils/metaprogramming.py b/cumulusci/utils/metaprogramming.py
new file mode 100644
index 000000000..89da3ebc0
--- /dev/null
+++ b/cumulusci/utils/metaprogramming.py
@@ -0,0 +1,10 @@
+class classproperty(object):
+    """Similar to @property, but allows class-level properties.
+
+    That is, a property whose getter is like a classmethod."""
+
+    def __init__(self, fget):
+        self.fget = fget
+
+    def __get__(self, owner_self, owner_cls):
+        return self.fget(owner_cls)
diff --git a/cumulusci/utils/options.py b/cumulusci/utils/options.py
new file mode 100644
index 000000000..3f9fe11d9
--- /dev/null
+++ b/cumulusci/utils/options.py
@@ -0,0 +1,117 @@
+import json
+from inspect import signature
+from typing import Any, Dict, List
+
+from pydantic import DirectoryPath, Field, FilePath, create_model
+
+from cumulusci.utils.yaml.model_parser import CCIDictModel
+
+
+def _describe_field(field):
+    "Convert a Pydantic field into a CCI task_option dict"
+    rc = {
+        "description": field.field_info.description,
+        "required": field.required,
+    }
+    if field.field_info.default != ...:
+        rc["default"] = field.field_info.default
+    return rc
+
+
+class CCIOptions(CCIDictModel):
+    "Base class for all options in tasks"
+
+    @classmethod
+    def as_task_options(cls):
+        return {
+            fieldname: _describe_field(field)
+            for fieldname, field in cls.__fields__.items()
+        }
+
+
+class CCIOptionType:
+    """Base class for custom option types.
+
+    Subclasses must implement 'from_str' and it must have a type-hinted return value like this:
+
+    class ListOfStringsOption(CCIOptionType):
+        @classmethod
+        def from_str(cls, v) -> List[str]:
+            return [s.strip() for s in v.split(",")]
+    """
+
+    name = None
+
+    @classmethod
+    def __get_validators__(cls):
+        "https://pydantic-docs.helpmanual.io/usage/types/#classes-with-__get_validators__"
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, v):
+        """Validate and convert a value.
+
+        If its a string, parse it, else, just validate it.
+        """
+        if isinstance(v, str):
+            if v.startswith("{") or v.startswith("["):
+                v = json.loads(v)
+            else:
+                v = cls.from_str(v)
+
+        # Pydantic can't just parse/validate arbitrary data unless
+        # it has a model. So we create a dummy model for
+        # it to have a parsing/validating context
+        target_type = signature(cls.from_str).return_annotation
+
+        Dummy = create_model(cls.name or cls.__name__, __root__=(target_type, ...))
+
+        return Dummy.parse_obj(v).__root__
+
+
+class ListOfStringsOption(CCIOptionType):
+    """Parses a list of strings from a comma-delimited string"""
+
+    @classmethod
+    def from_str(cls, v) -> List[str]:
+        return [s.strip() for s in v.split(",")]
+
+
+class MappingOption(CCIOptionType):
+    """Parses a Mapping of Str->Any from a string in format a:b,c:d"""
+
+    @classmethod
+    def from_str(cls, v) -> Dict[str, Any]:
+        return parse_list_of_pairs_dict_arg(v)
+
+
+def parse_list_of_pairs_dict_arg(arg):
+    """Process an arg in the format "aa:bb,cc:dd" """
+    if isinstance(arg, dict):
+        return arg
+    elif isinstance(arg, str):
+        rc = {}
+        for key_value in arg.split(","):
+            subparts = key_value.split(":", maxsplit=1)
+            if len(subparts) == 2:
+                key, value = subparts
+                if key in rc:
+                    raise TypeError(f"Var specified twice: {key}")
+                rc[key] = value
+            else:
+                raise TypeError(f"Var is not a name/value pair: {key_value}")
+        return rc
+    else:
+        raise TypeError(f"Arg is not a dict or string ({type(arg)}): {arg}")
+
+
+__all__ = [
+    "Field",
+    "FilePath",
+    "DirectoryPath",
+    "parse_list_of_pairs_dict_arg",
+    "CCIOptions",
+    "CCIOptionType",
+    "ListOfStringsOption",
+    "MappingOption",
+]
diff --git a/docs/config.md b/docs/config.md
index 119ed9482..0435bb405 100644
--- a/docs/config.md
+++ b/docs/config.md
@@ -119,18 +119,44 @@ tasks specific to your project when running `cci task list`.
 
 Congratulations! You created a new custom task in CumulusCI.
 
-If you've built a custom task in Python, you can make it available to
-the project by adding the task to a `tasks` subdirectory in the repo
-and under the `tasks` section of the
-`cumulusci.yml` file. (Let's assume that your task's class is named
-`MyNewTaskClassName` and exists in the file `tasks/task_file.py`.)
+### Custom Tasks in Python
+
+You can use Python to implement entirely new functionality. You do so by
+putting it in the "tasks" subdiretory of your project. For example, you
+might write a file called `tasks/write_file_custom_task.py`.
+
+```python
+from pathlib import Path
+
+from cumulusci.core.tasks import BaseTask, CCIOptions
+from cumulusci.utils.options import Field
+
+
+class WriteFileCustomTask(BaseTask):
+    class Options(CCIOptions):
+        mypath: Path = Field(..., description="A filepath to be used by the task")
+        mystring: str = Field("Hello", description="A string to be used by the task")
+
+    parsed_options: Options
+
+    def _run_task(self):
+        file = self.parsed_options.mypath
+        data = self.parsed_options.mystring
+        file.write_text(data)
+        self.logger.info(f"Wrote {data} to {file}")
+```
+
+You can make it available to
+the project by adding it under the `tasks` section of the
+`cumulusci.yml` file. If one does not exist, you can
+create one.
 
 ```yaml
 tasks:
     my_new_task:
-        description: Description of the task
-        class_path: tasks.task_file.MyNewTaskClassName
-        group: projectName
+        description: My custom task to write data to a file
+            class_path: tasks.write_file_custom_task.WriteFileCustomTask
+            group: projectName
 ```
 
 This task will be accessible directly to this project. It will also
@@ -138,6 +164,81 @@ be accessible to any other project that adds this as a source
 (see [Tasks and Flows from a Different Project](tasks-and-flows-from-a-different-project)) with
 `allow_remote_code: True`.
 
+#### Options for Custom Tasks
+
+Task options are defined by declaring a nested `Options` class. This class must sublass `cumulusci.utils.options.CCIOptions`. These options are validated via the use of [Pydantic models](https://pydantic-docs.helpmanual.io/usage/models/) which are generated dynamically for each `Options` class.
+Each option can define its own type via either a [standard library type](https://pydantic-docs.helpmanual.io/usage/types/) or by utilizing a custom type from [`cumulusci.utils.options`](https://github.com/SFDO-Tooling/CumulusCI/blob/main/cumulusci/utils/options.py).
+Additionally the [`Field()`](https://pydantic-docs.helpmanual.io/usage/schema/#field-customisation`) function is useful for further customizing options. This can be imported from `cumulusci.utils.options` and used when defining individual options.
+It has the same features as the pydantic function.
+
+The task above (`WriteFileCustomTask`) takes two options: (1) A defaulted string (myString),
+and (2) A required file path.
+
+Once the options are defined, they can be accessed via the `parsed_options` property of the task.
+
+Some of the most commonly used types are:
+
+-   `pathlib.Path`: simply uses the type itself for validation by passing the value to Path(v);
+-   `FilePath`: like Path, but the path must exist and be a file
+-   `DirectoryPath`: like Path, but the path must exist and be a directory
+-   `MappingOption`: Parses pairs of values from a string in format "a:b,c:d"
+-   `ListOfStringsOption`: Parses a list of comma-separated strings from an argument in the format "abc,def,ghi"
+-   `JSON`: Parse a JSON string into a structure, including a deeply nested structure
+
+Others can be found in the [pydantic docs](https://pydantic-docs.helpmanual.io/usage/types).
+
+If you are comfortable with Python types, you can do very sophisticated parsing.
+For example:
+
+```python
+from typing import Optional
+
+from pydantic import Json
+
+from cumulusci.core.tasks import BaseTask, CCIOptions
+from cumulusci.utils.options import Field
+
+
+class Person(CCIOptions):
+    name: str
+    children: Optional[dict[str, "Person"]]
+
+
+People = dict[str, Person]
+
+
+class ComplexOptionsCustomTask(BaseTask):
+    class Options(CCIOptions):
+        lineage: Json[People] | People = Field(..., description="Foo")
+
+    parsed_options: Options
+
+    def _run_task(self):
+        self.logger.info(f"Got {self.parsed_options.lineage}")
+```
+
+Which can parse this commmand line:
+
+```sh
+$ cci task run complex_options --lineage '{"Bob": {"name": "Bob Cat Sr", "children": {"Bob": {"name": "Bob Cat Jr"}}}}'
+    Got {'Bob': Person(name='Bob Cat Sr', children={'Bob': Person(name='Bob Cat Jr', children=None)})}
+```
+
+Or this YAML:
+
+```yaml
+complex_options:
+    description: Description of the task
+    class_path: tasks.complex_options_custom_task.ComplexOptionsCustomTask
+    options:
+        lineage:
+            Bob:
+                name: Bob Cat Sr
+                children:
+                    Bob:
+                        name: Bob Cat Jr
+```
+
 (use-variables-for-task-options)=
 
 ### Use Variables for Task Options
diff --git a/docs/cookbook.md b/docs/cookbook.md
index 878f5d1fb..3b175a169 100644
--- a/docs/cookbook.md
+++ b/docs/cookbook.md
@@ -89,18 +89,21 @@ The following shows an example task named `project_default_settings`
 which runs the public static method `initializeProjectDefaults()`
 located in file `scripts.initialize.cls`:
 
-    project_default_settings:
-        description: Configure the default project settings
-        class_path: cumulusci.tasks.apex.anon.AnonymousApexTask
-        group: projectName
-        options:
-            path: scripts/initialize.cls
-            apex: initializeProjectDefaults();
-
-## Flow Recipes
+```yaml
+project_default_settings:
+    description: Configure the default project settings
+    class_path: cumulusci.tasks.apex.anon.AnonymousApexTask
+    group: projectName
+    options:
+        path: scripts/initialize.cls
+        apex: initializeProjectDefaults();
+```
 
-## Robot Recipes
+## Converting from `task_options` to new-style Options
 
-## Metadata ETL Recipes
+If you have custom tasks that you'd like to convert to using the new options API, then you will want to do the following:
 
-## Python Recipes
+1. Create a nested `Options` class within the task class.
+2. For each of the options you have defined in the `task_options` dict you will create a corresponding option property in the `Options` class.
+3. Delete the `task_options` dictionary.
+4. Review the `_init_options()` and `_validate_options()` methods on the task class -- if they exist -- to see whether they are still relevant and correct.


Test patch:
diff --git a/cumulusci/tasks/tests/test_util.py b/cumulusci/tasks/tests/test_util.py
index 6db6ead14..747358328 100644
--- a/cumulusci/tasks/tests/test_util.py
+++ b/cumulusci/tasks/tests/test_util.py
@@ -152,8 +152,16 @@ class TestUtilTasks:
             )
             task = util.FindReplace(self.project_config, task_config, self.org_config)
             task()
-            assert task.options["replace"] == "bars"
+            assert task.parsed_options["replace"] == "bars"
             find_replace.assert_called_once()
+            assert find_replace.mock_calls[0] == mock.call(
+                find="foo",
+                replace="bars",
+                directory=mock.ANY,
+                filePattern="*",
+                logger=mock.ANY,
+                max=1,
+            )
 
     @mock.patch("cumulusci.tasks.util.find_replace_regex")
     def test_FindReplaceRegex(self, find_replace_regex):
diff --git a/cumulusci/utils/tests/test_option_parsing.py b/cumulusci/utils/tests/test_option_parsing.py
new file mode 100644
index 000000000..82fceb962
--- /dev/null
+++ b/cumulusci/utils/tests/test_option_parsing.py
@@ -0,0 +1,156 @@
+from datetime import date
+from pathlib import Path
+
+import pytest
+
+from cumulusci.core.config import (
+    BaseProjectConfig,
+    OrgConfig,
+    TaskConfig,
+    UniversalConfig,
+)
+from cumulusci.core.exceptions import TaskOptionsError
+from cumulusci.core.tasks import BaseTask
+from cumulusci.utils.options import (
+    CCIOptions,
+    Field,
+    ListOfStringsOption,
+    MappingOption,
+)
+
+ORG_ID = "00D000000000001"
+USERNAME = "sample@example"
+
+
+class TaskToTestTypes(BaseTask):
+    class Options(CCIOptions):
+        the_list: ListOfStringsOption = Field(
+            default=[],
+            description="A list of strings",
+        )
+        the_mapping: MappingOption = Field(
+            default=[],
+            description="A list of strings",
+        )
+        the_date: date = Field(default=None, description="A date")
+        the_bool: bool = Field(default=True, description="A bool")
+        the_path: Path = Field(default=None, description="A bool")
+        req: int = Field(..., description="Mandatory int")
+
+    parsed_options: Options
+
+    def _run_task(self):
+        for key, value in vars(self.parsed_options).items():
+            if value:
+                print(key, repr(getattr(self.parsed_options, key)))
+
+
+class TestTaskOptionsParsing:
+    def setup_class(self):
+        self.global_config = UniversalConfig()
+        self.project_config = BaseProjectConfig(
+            self.global_config, config={"noyaml": True}
+        )
+        self.org_config = OrgConfig({"username": USERNAME, "org_id": ORG_ID}, "test")
+        self.task_config = TaskConfig(
+            {
+                "options": {
+                    "the_list": ["a"],
+                    "the_mapping": {"b": "c"},
+                    "the_date": date(2019, 1, 30),
+                    "the_bool": True,
+                    "req": 1,
+                }
+            }
+        )
+
+    def test_noop(self):
+        task = TaskToTestTypes(self.project_config, self.task_config, self.org_config)
+        task._init_options({})
+        assert task.options["the_list"] == ["a"]
+        assert task.options["the_mapping"] == {"b": "c"}
+        assert task.options["the_date"] == date(2019, 1, 30)
+        assert task.options["the_bool"] is True
+        assert task.parsed_options.the_list == task.options["the_list"]
+        assert task.parsed_options.the_mapping == task.options["the_mapping"]
+        assert task.parsed_options.the_date == task.options["the_date"]
+        assert task.parsed_options.the_bool == task.options["the_bool"]
+
+    def test_split_list(self):
+        task_config = TaskConfig({"options": {"the_list": "a,b,c,d,e", "req": 1}})
+        task = TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert task.parsed_options.the_list == ["a", "b", "c", "d", "e"]
+
+    def test_split_dict(self):
+        task_config = TaskConfig(
+            {"options": {"the_mapping": "a:aa,b:bb,c:cc", "req": 1}}
+        )
+        task = TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert task.parsed_options.the_mapping == {"a": "aa", "b": "bb", "c": "cc"}
+
+    def test_parse_date(self):
+        task_config = TaskConfig({"options": {"the_date": "1972-11-03", "req": 1}})
+        task = TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert task.parsed_options.the_date == date(1972, 11, 3)
+
+    def test_parse_bool(self):
+        task_config = TaskConfig({"options": {"the_bool": "False", "req": 1}})
+        task = TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert task.parsed_options.the_bool is False
+
+    def test_parse_path(self):
+        task_config = TaskConfig({"options": {"the_path": __file__, "req": 1}})
+        task = TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert task.parsed_options.the_path.exists()
+
+    def test_missing_option(self):
+        task_config = TaskConfig({"options": {"the_path": __file__}})
+        with pytest.raises(TaskOptionsError) as e:
+            TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert "req" in str(e.value)
+        assert "required" in str(e.value)
+
+    def test_wrong_option_type(self):
+        task_config = TaskConfig({"options": {"the_bool": "Negatory", "req": 1}})
+        with pytest.raises(TaskOptionsError) as e:
+            TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert "the_bool" in str(e.value)
+        assert "boolean" in str(e.value)
+
+    def test_extra_options(self):
+        task_config = TaskConfig({"options": {"foo": "bar", "req": 1}})
+        with pytest.raises(TaskOptionsError) as e:
+            TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert "foo" in str(e.value)
+        assert "extra options" in str(e.value)
+
+    def test_exception_in_parsing(self):
+        task_config = TaskConfig(
+            {"options": {"the_mapping": "aaaaa,bbbb:cccc", "req": 1}}
+        )
+        with pytest.raises(TaskOptionsError) as e:
+            TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert "the_mapping" in str(e.value)
+        assert "name/value" in str(e.value)
+
+    def test_wrong_type(self):
+        task_config = TaskConfig({"options": {"the_path": 5, "req": 1}})
+        with pytest.raises(TaskOptionsError) as e:
+            TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert "the_path" in str(e.value)
+        assert "path" in str(e.value)
+
+    def test_null_for_required(self):
+        task_config = TaskConfig({"options": {"the_path": None, "req": None}})
+        with pytest.raises(TaskOptionsError) as e:
+            TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert "req" in str(e.value)
+        assert "none" in str(e.value)
+
+    def test_multiple_errors(self):
+        task_config = TaskConfig({"options": {"the_bool": "Nope"}})
+        with pytest.raises(TaskOptionsError) as e:
+            TaskToTestTypes(self.project_config, task_config, self.org_config)
+        assert "the_bool" in str(e.value)
+        assert "req" in str(e.value)
+        assert "Errors" in str(e.value)