text: import collections
  import copy
  import inspect
  import uuid
  import warnings
  from datetime import timedelta
  from typing import (
      TYPE_CHECKING,
      Any,
      Callable,
      Dict,
      Iterable,
      List,
      Mapping,
      Set,
      Tuple,
      Union,
  )
  
  import prefect
  import prefect.engine.cache_validators
  import prefect.engine.signals
  import prefect.triggers
  from prefect.utilities import logging
  from prefect.utilities.notifications import callback_factory
  
  if TYPE_CHECKING:
      from prefect.core.flow import Flow  # pylint: disable=W0611
      from prefect.engine.result_handlers import ResultHandler
      from prefect.engine.state import State
  
  VAR_KEYWORD = inspect.Parameter.VAR_KEYWORD
  
  
  def _validate_run_signature(run: Callable) -> None:
      func = getattr(run, "__wrapped__", run)
      try:
          run_sig = inspect.getfullargspec(func)
      except TypeError as exc:
          if str(exc) == "unsupported callable":
              raise ValueError(
                  "This function can not be inspected (this is common "
                  "with `builtin` and `numpy` functions). In order to "
                  "use it as a task, please wrap it in a standard "
                  "Python function. For more detail, see "
                  "https://docs.prefect.io/core/tutorials/task-guide.html#the-task-decorator"
              )
          raise
  
      if run_sig.varargs:
          raise ValueError(
              "Tasks with variable positional arguments (*args) are not "
              "supported, because all Prefect arguments are stored as "
              "keywords. As a workaround, consider modifying the run() "
              "method to accept **kwargs and feeding the values "
              "to *args."
          )
  
      reserved_kwargs = ["upstream_tasks", "mapped", "task_args", "flow"]
      violations = [kw for kw in reserved_kwargs if kw in run_sig.args]
      if violations:
          msg = "Tasks cannot have the following argument names: {}.".format(
              ", ".join(violations)
          )
          msg += " These are reserved keyword arguments."
          raise ValueError(msg)
  
  
  class SignatureValidator(type):
      def __new__(cls, name: str, parents: tuple, methods: dict) -> "SignatureValidator":
          run = methods.get("run", lambda: None)
          _validate_run_signature(run)
  
          # necessary to ensure classes that inherit from parent class
          # also get passed through __new__
          return type.__new__(cls, name, parents, methods)  # type: ignore
  
  
  class Task(metaclass=SignatureValidator):
      """
      The Task class which is used as the full representation of a unit of work.
  
      This Task class can be used directly as a first class object where it must
      be inherited from by a class that implements the `run` method.  For a more
      functional way of generating Tasks, see [the task decorator](../utilities/tasks.html).
  
      Inheritance example:
      ```python
      class AddTask(Task):
          def run(self, x, y):
              return x + y
      ```
  
      *Note:* The implemented `run` method cannot have `*args` in its signature. In addition,
      the following keywords are reserved: `upstream_tasks`, `task_args` and `mapped`.
  
      An instance of a `Task` can be used functionally to generate other task instances
      with the same attributes but with different values bound to their `run` methods.
  
      Example:
      ```python
      class AddTask(Task):
          def run(self, x, y):
              return x + y
  
      a = AddTask()
  
      with Flow("My Flow") as f:
          t1 = a(1, 2) # t1 != a
          t2 = a(5, 7) # t2 != a
      ```
  
      To bind values to a Task's run method imperatively (and without making a copy), see `Task.bind`.
  
      Args:
          - name (str, optional): The name of this task
          - slug (str, optional): The slug for this task. Slugs are required and must be unique
              within any flow; if not provided a random UUID will be generated.
          - tags ([str], optional): A list of tags for this task
          - max_retries (int, optional): The maximum amount of times this task can be retried
          - retry_delay (timedelta, optional): The amount of time to wait until task is retried
          - timeout (int, optional): The amount of time (in seconds) to wait while
              running this task before a timeout occurs; note that sub-second resolution is not supported
          - trigger (callable, optional): a function that determines whether the task should run, based
                  on the states of any upstream tasks.
          - skip_on_upstream_skip (bool, optional): if `True`, if any immediately
                  upstream tasks are skipped, this task will automatically be skipped as well,
                  regardless of trigger. By default, this prevents tasks from attempting to use either state or data
                  from tasks that didn't run. If `False`, the task's trigger will be called as normal,
                  with skips considered successes. Defaults to `True`.
          - cache_for (timedelta, optional): The amount of time to maintain a cache
              of the outputs of this task.  Useful for situations where the containing Flow
              will be rerun multiple times, but this task doesn't need to be.
          - cache_validator (Callable, optional): Validator that will determine
              whether the cache for this task is still valid (only required if `cache_for`
              is provided; defaults to `prefect.engine.cache_validators.duration_only`)
          - cache_key (str, optional): if provided, a `cache_key` serves as a unique identifier for this Task's cache, and can
              be shared across both Tasks _and_ Flows; if not provided, the Task's _name_ will be used if running locally, or the
              Task's database ID if running in Cloud
          - checkpoint (bool, optional): if this Task is successful, whether to
              store its result using the `result_handler` available during the run; Also note that
              checkpointing will only occur locally if `prefect.config.flows.checkpointing` is set to `True`
          - result_handler (ResultHandler, optional): the handler to use for
              retrieving and storing state results during execution; if not provided, will default to the
              one attached to the Flow
          - state_handlers (Iterable[Callable], optional): A list of state change handlers
              that will be called whenever the task changes state, providing an
              opportunity to inspect or modify the new state. The handler
              will be passed the task instance, the old (prior) state, and the new
              (current) state, with the following signature:
                  `state_handler(task: Task, old_state: State, new_state: State) -> Optional[State]`
              If multiple functions are passed, then the `new_state` argument will be the
              result of the previous handler.
          - on_failure (Callable, optional): A function with signature `fn(task: Task, state: State) -> None`
              with will be called anytime this Task enters a failure state
  
      Raises:
          - TypeError: if `tags` is of type `str`
          - TypeError: if `timeout` is not of type `int`
      """
  
      # Tasks are not iterable, though they do have a __getitem__ method
      __iter__ = None
  
      def __init__(
          self,
          name: str = None,
          slug: str = None,
          tags: Iterable[str] = None,
          max_retries: int = None,
          retry_delay: timedelta = None,
          timeout: int = None,
          trigger: Callable[[Set["State"]], bool] = None,
          skip_on_upstream_skip: bool = True,
          cache_for: timedelta = None,
          cache_validator: Callable = None,
          cache_key: str = None,
          checkpoint: bool = None,
          result_handler: "ResultHandler" = None,
          state_handlers: List[Callable] = None,
          on_failure: Callable = None,
      ):
  
          self.name = name or type(self).__name__
          self.slug = slug or str(uuid.uuid4())
  
          self.logger = logging.get_logger("Task: {}".format(self.name))
  
          # avoid silently iterating over a string
          if isinstance(tags, str):
              raise TypeError("Tags should be a set of tags, not a string.")
          current_tags = set(prefect.context.get("tags", set()))
          self.tags = (set(tags) if tags is not None else set()) | current_tags
  
          max_retries = (
              max_retries
              if max_retries is not None
              else prefect.config.tasks.defaults.max_retries
          )
          retry_delay = (
              retry_delay
              if retry_delay is not None
              else prefect.config.tasks.defaults.retry_delay
          )
          timeout = (
              timeout if timeout is not None else prefect.config.tasks.defaults.timeout
          )
  
          if max_retries > 0 and retry_delay is None:
              raise ValueError(
                  "A datetime.timedelta `retry_delay` must be provided if max_retries > 0"
              )
          # specify not max retries because the default is false
          if retry_delay is not None and not max_retries:
              raise ValueError(
                  "A `max_retries` argument greater than 0 must be provided if specifying a retry delay."
              )
          if timeout is not None and not isinstance(timeout, int):
              raise TypeError(
                  "Only integer timeouts (representing seconds) are supported."
              )
          self.max_retries = max_retries
          self.retry_delay = retry_delay
          self.timeout = timeout
  
          self.trigger = trigger or prefect.triggers.all_successful
          self.skip_on_upstream_skip = skip_on_upstream_skip
  
          if cache_for is None and (
              cache_validator is not None
              and cache_validator is not prefect.engine.cache_validators.never_use
          ):
              warnings.warn(
                  "cache_validator provided without specifying cache expiration (cache_for); this Task will not be cached."
              )
  
          self.cache_for = cache_for
          self.cache_key = cache_key
          default_validator = (
              prefect.engine.cache_validators.never_use
              if cache_for is None
              else prefect.engine.cache_validators.duration_only
          )
          self.cache_validator = cache_validator or default_validator
          self.checkpoint = checkpoint
          self.result_handler = result_handler
  
          if state_handlers and not isinstance(state_handlers, collections.Sequence):
              raise TypeError("state_handlers should be iterable.")
          self.state_handlers = state_handlers or []
          if on_failure is not None:
              self.state_handlers.append(
                  callback_factory(on_failure, check=lambda s: s.is_failed())
              )
          self.auto_generated = False
  
      def __repr__(self) -> str:
          return "<Task: {self.name}>".format(self=self)
  
      # reimplement __hash__ because we override __eq__
      def __hash__(self) -> int:
          return id(self)
  
      # Run  --------------------------------------------------------------------
  
      def run(self) -> None:
          """
          The `run()` method is called (with arguments, if appropriate) to run a task.
  
          *Note:* The implemented `run` method cannot have `*args` in its signature. In addition,
          the following keywords are reserved: `upstream_tasks`, `task_args` and `mapped`.
  
          If a task has arguments in its `run()` method, these can be bound either by using the functional
          API and _calling_ the task instance, or by using `self.bind` directly.
  
          In addition to running arbitrary functions, tasks can interact with Prefect in a few ways:
          <ul><li> Return an optional result. When this function runs successfully,
              the task is considered successful and the result (if any) can be
              made available to downstream tasks. </li>
          <li> Raise an error. Errors are interpreted as failure. </li>
          <li> Raise a [signal](../engine/signals.html). Signals can include `FAIL`, `SUCCESS`, `RETRY`, `SKIP`, etc.
              and indicate that the task should be put in the indicated state.
                  <ul>
                  <li> `FAIL` will lead to retries if appropriate </li>
                  <li> `SUCCESS` will cause the task to be marked successful </li>
                  <li> `RETRY` will cause the task to be marked for retry, even if `max_retries`
                      has been exceeded </li>
                  <li> `SKIP` will skip the task and possibly propogate the skip state through the
                      flow, depending on whether downstream tasks have `skip_on_upstream_skip=True`. </li></ul>
          </li></ul>
          """
          pass
  
      # Dependencies -------------------------------------------------------------
  
      def copy(self, **task_args: Any) -> "Task":
          """
          Creates and returns a copy of the current Task.
  
          Args:
              - **task_args (dict, optional): a dictionary of task attribute keyword arguments, these attributes
                  will be set on the new copy
  
          Raises:
              - AttributeError: if any passed `task_args` are not attributes of the original
  
          Returns:
              - Task: a copy of the current Task, with any attributes updated from `task_args`
          """
  
          flow = prefect.context.get("flow", None)
          if (
              flow
              and self in flow.tasks
              and (flow.edges_to(self) or flow.edges_from(self))
          ):
              warnings.warn(
                  "You are making a copy of a task that has dependencies on or to other tasks "
                  "in the active flow context. The copy will not retain those dependencies."
              )
  
          new = copy.copy(self)
  
          # ensure new slug is provided
          if "slug" not in task_args:
              task_args["slug"] = str(uuid.uuid4())
  
          # check task_args
          for attr, val in task_args.items():
              if not hasattr(new, attr):
                  raise AttributeError(
                      "{0} does not have {1} as an attribute".format(self, attr)
                  )
              else:
                  setattr(new, attr, val)
  
          new.tags = copy.deepcopy(self.tags).union(set(new.tags))
          tags = set(prefect.context.get("tags", set()))
          new.tags.update(tags)
  
          return new
  
      def __call__(
          self,
          *args: Any,
          mapped: bool = False,
          task_args: dict = None,
          upstream_tasks: Iterable[Any] = None,
          flow: "Flow" = None,
          **kwargs: Any
      ) -> "Task":
          """
          Calling a Task instance will first create a _copy_ of the instance, and then
          bind any passed `args` / `kwargs` to the run method of the copy. This new task
          is then returned.
  
          Args:
              - *args: arguments to bind to the new Task's `run` method
              - **kwargs: keyword arguments to bind to the new Task's `run` method
              - mapped (bool, optional): Whether the results of these tasks should be mapped over
                  with the specified keyword arguments; defaults to `False`.
                  If `True`, any arguments contained within a `prefect.utilities.tasks.unmapped`
                  container will _not_ be mapped over.
              - task_args (dict, optional): a dictionary of task attribute keyword arguments, these attributes
                  will be set on the new copy
              - upstream_tasks ([Task], optional): a list of upstream dependencies
                  for the new task.  This kwarg can be used to functionally specify
                  dependencies without binding their result to `run()`
              - flow (Flow, optional): The flow to set dependencies on, defaults to the current
                  flow in context if no flow is specified
  
          Returns:
              - Task: a new Task instance
          """
          new = self.copy(**(task_args or {}))
          new.bind(
              *args, mapped=mapped, upstream_tasks=upstream_tasks, flow=flow, **kwargs
          )
          return new
  
      def bind(
          self,
          *args: Any,
          mapped: bool = False,
          upstream_tasks: Iterable[Any] = None,
          flow: "Flow" = None,
          **kwargs: Any
      ) -> "Task":
          """
          Binding a task to (keyword) arguments creates a _keyed_ edge in the active Flow
          that will pass data from the arguments (whether Tasks or constants) to the
          Task's `run` method under the appropriate key. Once a Task is bound in this
          manner, the same task instance cannot be bound a second time in the same Flow.
  
          To bind arguments to a _copy_ of this Task instance, see `__call__`.
          Additionally, non-keyed edges can be created by passing any upstream
          dependencies through `upstream_tasks`.
  
          Args:
              - *args: arguments to bind to the current Task's `run` method
              - mapped (bool, optional): Whether the results of these tasks should be mapped over
                  with the specified keyword arguments; defaults to `False`.
                  If `True`, any arguments contained within a `prefect.utilities.tasks.unmapped`
                  container will _not_ be mapped over.
              - upstream_tasks ([Task], optional): a list of upstream dependencies for the
                  current task.
              - flow (Flow, optional): The flow to set dependencies on, defaults to the current
                  flow in context if no flow is specified
              - **kwargs: keyword arguments to bind to the current Task's `run` method
  
          Returns:
              - Task: the current Task instance
          """
  
          # this will raise an error if callargs weren't all provided
          signature = inspect.signature(self.run)
          callargs = dict(signature.bind(*args, **kwargs).arguments)  # type: Dict
  
          # bind() compresses all variable keyword arguments under the ** argument name,
          # so we expand them explicitly
          var_kw_arg = next(
              (p for p in signature.parameters.values() if p.kind == VAR_KEYWORD), None
          )
          if var_kw_arg:
              callargs.update(callargs.pop(var_kw_arg.name, {}))
  
          flow = flow or prefect.context.get("flow", None)
          if not flow:
              raise ValueError("Could not infer an active Flow context.")
  
          self.set_dependencies(
              flow=flow,
              upstream_tasks=upstream_tasks,
              keyword_tasks=callargs,
              mapped=mapped,
          )
  
          tags = set(prefect.context.get("tags", set()))
          self.tags.update(tags)
  
          return self
  
      def map(
          self,
          *args: Any,
          upstream_tasks: Iterable[Any] = None,
          flow: "Flow" = None,
          task_args: dict = None,
          **kwargs: Any
      ) -> "Task":
          """
          Map the Task elementwise across one or more Tasks. Arguments that should _not_ be mapped over
          should be placed in the `prefect.utilities.tasks.unmapped` container.
  
          For example:
              ```
              task.map(x=X, y=unmapped(Y))
              ```
          will map over the values of `X`, but not over the values of `Y`
  
  
          Args:
              - *args: arguments to map over, which will elementwise be bound to the Task's `run` method
              - upstream_tasks ([Task], optional): a list of upstream dependencies
                  to map over
              - flow (Flow, optional): The flow to set dependencies on, defaults to the current
                  flow in context if no flow is specified
              - task_args (dict, optional): a dictionary of task attribute keyword arguments,
                  these attributes will be set on the new copy
              - **kwargs: keyword arguments to map over, which will elementwise be bound to the Task's `run` method
  
          Raises:
              - AttributeError: if any passed `task_args` are not attributes of the original
  
          Returns:
              - Task: a new Task instance
          """
          new = self.copy(**(task_args or {}))
          return new.bind(
              *args, mapped=True, upstream_tasks=upstream_tasks, flow=flow, **kwargs
          )
  
      def set_dependencies(
          self,
          flow: "Flow" = None,
          upstream_tasks: Iterable[object] = None,
          downstream_tasks: Iterable[object] = None,
          keyword_tasks: Mapping[str, object] = None,
          mapped: bool = False,
          validate: bool = None,
      ) -> None:
          """
          Set dependencies for a flow either specified or in the current context using this task
  
          Args:
              - flow (Flow, optional): The flow to set dependencies on, defaults to the current
              flow in context if no flow is specified
              - upstream_tasks ([object], optional): A list of upstream tasks for this task
              - downstream_tasks ([object], optional): A list of downtream tasks for this task
              - keyword_tasks ({str, object}}, optional): The results of these tasks will be provided
              to this task under the specified keyword arguments.
              - mapped (bool, optional): Whether the results of the _upstream_ tasks should be mapped over
                  with the specified keyword arguments
              - validate (bool, optional): Whether or not to check the validity of the flow. If not
                  provided, defaults to the value of `eager_edge_validation` in your Prefect
                  configuration file.
  
          Returns:
              - None
  
          Raises:
              - ValueError: if no flow is specified and no flow can be found in the current context
          """
          flow = flow or prefect.context.get("flow", None)
          if not flow:
              raise ValueError(
                  "No Flow was passed, and could not infer an active Flow context."
              )
  
          flow.set_dependencies(
              task=self,
              upstream_tasks=upstream_tasks,
              downstream_tasks=downstream_tasks,
              keyword_tasks=keyword_tasks,
              validate=validate,
              mapped=mapped,
          )
  
      def set_upstream(
          self, task: object, flow: "Flow" = None, key: str = None, mapped: bool = False
      ) -> None:
          """
          Sets the provided task as an upstream dependency of this task.
  
          Args:
              - task (object): A task or object that will be converted to a task that will be set
                  as a upstream dependency of this task.
              - flow (Flow, optional): The flow to set dependencies on, defaults to the current
                  flow in context if no flow is specified
              - key (str, optional): The key to be set for the new edge; the result of the upstream task
                  will be passed to this task's `run()` method under this keyword argument.
              - mapped (bool, optional): Whether this dependency is mapped; defaults to `False`
  
          Raises:
              - ValueError: if no flow is specified and no flow can be found in the current context
          """
          if key is not None:
              keyword_tasks = {key: task}
              self.set_dependencies(flow=flow, keyword_tasks=keyword_tasks, mapped=mapped)
          else:
              self.set_dependencies(flow=flow, upstream_tasks=[task], mapped=mapped)
  
      def set_downstream(
          self, task: "Task", flow: "Flow" = None, key: str = None, mapped: bool = False
      ) -> None:
          """
          Sets the provided task as a downstream dependency of this task.
  
          Args:
              - task (Task): A task that will be set as a downstream dependency of this task.
              - flow (Flow, optional): The flow to set dependencies on, defaults to the current
                  flow in context if no flow is specified
              - key (str, optional): The key to be set for the new edge; the result of this task
                  will be passed to the downstream task's `run()` method under this keyword argument.
              - mapped (bool, optional): Whether this dependency is mapped; defaults to `False`
  
          Raises:
              - ValueError: if no flow is specified and no flow can be found in the current context
          """
          if key is not None:
              keyword_tasks = {key: self}
              task.set_dependencies(  # type: ignore
                  flow=flow, keyword_tasks=keyword_tasks, mapped=mapped
              )  # type: ignore
          else:
              task.set_dependencies(flow=flow, upstream_tasks=[self], mapped=mapped)
  
      def inputs(self) -> Dict[str, Dict]:
          """
          Describe the inputs for this task. The result is a dictionary that maps each input to
          a `type`, `required`, and `default`. All values are inferred from the `run()`
          signature; this method can be overloaded for more precise control.
  
          Returns:
              - dict
          """
          inputs = {}
          for name, parameter in inspect.signature(self.run).parameters.items():
              input_type = parameter.annotation
              if input_type is inspect._empty:  # type: ignore
                  input_type = Any
  
              input_default = parameter.default
              input_required = False
              if input_default is inspect._empty:  # type: ignore
                  input_required = True
                  input_default = None
  
              inputs[name] = dict(
                  type=input_type, default=input_default, required=input_required
              )
  
          return inputs
  
      def outputs(self) -> Any:
          """
          Get the output types for this task.
  
          Returns:
              - Any
          """
          return_annotation = inspect.signature(self.run).return_annotation
          if return_annotation is inspect._empty:  # type: ignore
              return_annotation = Any
          return return_annotation
  
      # Serialization ------------------------------------------------------------
  
      def serialize(self) -> Dict[str, Any]:
          """
          Creates a serialized representation of this task
  
          Returns:
              - dict representing this task
          """
          return prefect.serialization.task.TaskSchema().dump(self)
  
      # Operators  ----------------------------------------------------------------
  
      def is_equal(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self == other`
  
          This can't be implemented as the __eq__() magic method because of Task
          comparisons.
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Equal().bind(self, other)
  
      def is_not_equal(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self != other`
  
          This can't be implemented as the __neq__() magic method because of Task
          comparisons.
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.NotEqual().bind(self, other)
  
      def not_(self) -> "Task":
          """
          Produces a Task that evaluates `not self`
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Not().bind(self)
  
      def or_(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self or other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Or().bind(self, other)
  
      # Magic Method Interactions  ----------------------------------------------------
  
      def __getitem__(self, key: Any) -> "Task":
          """
          Produces a Task that evaluates `self[key]`
  
          Args:
              - key (object): the object to use an an index for this task. It will be converted
                  to a Task if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.GetItem().bind(self, key)
  
      def __or__(self, other: object) -> object:
          """
          Creates a state dependency between `self` and `other`
              `self | other --> self.set_dependencies(downstream_tasks=[other])`
  
          Args:
              - other (object): An object that will be converted to a Task (if it isn't one already)
                  and set as a downstream dependency of this Task.
  
          Returns:
              - Task
          """
          self.set_dependencies(downstream_tasks=[other])
          return other
  
      def __mifflin__(self) -> None:  # coverage: ignore
          "Calls Dunder Mifflin"
          import webbrowser
  
          webbrowser.open("https://cicdw.github.io/welcome.html")
  
      def __ror__(self, other: object) -> "Task":
          """
          Creates a state dependency between `self` and `other`:
              `other | self --> self.set_dependencies(upstream_tasks=[other])`
  
          Args:
              - other (object): An object that will be converted to a Task and set as an
                  upstream dependency of this Task.
  
          Returns:
              - Task
          """
          self.set_dependencies(upstream_tasks=[other])
          return self
  
      # Maginc Method Operators  -----------------------------------------------------
  
      def __add__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self + other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Add().bind(self, other)
  
      def __sub__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self - other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Sub().bind(self, other)
  
      def __mul__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self * other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Mul().bind(self, other)
  
      def __truediv__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self / other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Div().bind(self, other)
  
      def __floordiv__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self // other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.FloorDiv().bind(self, other)
  
      def __mod__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self % other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Mod().bind(self, other)
  
      def __pow__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self ** other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Pow().bind(self, other)
  
      def __and__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self & other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.And().bind(self, other)
  
      def __radd__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other + self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Add().bind(other, self)
  
      def __rsub__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other - self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Sub().bind(other, self)
  
      def __rmul__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other * self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Mul().bind(other, self)
  
      def __rtruediv__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other / self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Div().bind(other, self)
  
      def __rfloordiv__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other // self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.FloorDiv().bind(other, self)
  
      def __rmod__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other % self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Mod().bind(other, self)
  
      def __rpow__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other ** self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.Pow().bind(other, self)
  
      def __rand__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `other & self`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.And().bind(other, self)
  
      def __gt__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self > other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.GreaterThan().bind(self, other)
  
      def __ge__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self >= other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.GreaterThanOrEqual().bind(self, other)
  
      def __lt__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self < other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.LessThan().bind(self, other)
  
      def __le__(self, other: object) -> "Task":
          """
          Produces a Task that evaluates `self <= other`
  
          Args:
              - other (object): the other operand of the operator. It will be converted to a Task
                  if it isn't one already.
  
          Returns:
              - Task
          """
          return prefect.tasks.core.operators.LessThanOrEqual().bind(self, other)
  
  
  class Parameter(Task):
      """
      A Parameter is a special task that defines a required flow input.
  
      A parameter's "slug" is automatically -- and immutably -- set to the parameter name.
      Flows enforce slug uniqueness across all tasks, so this ensures that the flow has
      no other parameters by the same name.
  
      Args:
          - name (str): the Parameter name.
          - required (bool, optional): If True, the Parameter is required and the default
              value is ignored.
          - default (any, optional): A default value for the parameter. If the default
              is not None, the Parameter will not be required.
          - tags ([str], optional): A list of tags for this parameter
  
      """
  
      def __init__(
          self,
          name: str,
          default: Any = None,
          required: bool = True,
          tags: Iterable[str] = None,
      ):
          if default is not None:
              required = False
  
          self.required = required
          self.default = default
  
          from prefect.engine.result_handlers import JSONResultHandler
  
          super().__init__(
              name=name,
              slug=name,
              tags=tags,
              result_handler=JSONResultHandler(),
              checkpoint=True,
          )
  
      def __repr__(self) -> str:
          return "<Parameter: {self.name}>".format(self=self)
  
      def __call__(self, flow: "Flow" = None) -> "Parameter":  # type: ignore
          """
          Calling a Parameter adds it to a flow.
  
          Args:
              - flow (Flow, optional): The flow to set dependencies on, defaults to the current
                  flow in context if no flow is specified
  
          Returns:
              - Task: a new Task instance
  
          """
          result = super().bind(flow=flow)
          assert isinstance(result, Parameter)  # mypy assert
          return result
  
      def copy(self, name: str, **task_args: Any) -> "Task":  # type: ignore
          """
          Creates a copy of the Parameter with a new name.
  
          Args:
              - name (str): the new Parameter name
              - **task_args (dict, optional): a dictionary of task attribute keyword arguments,
                  these attributes will be set on the new copy
  
          Raises:
              - AttributeError: if any passed `task_args` are not attributes of the original
  
          Returns:
              - Parameter: a copy of the current Parameter, with a new name and any attributes
                  updated from `task_args`
          """
          return super().copy(name=name, slug=name, **task_args)
  
      def run(self) -> Any:
          params = prefect.context.get("parameters") or {}
          if self.required and self.name not in params:
              self.logger.debug(
                  'Parameter "{}" was required but not provided.'.format(self.name)
              )
              raise prefect.engine.signals.FAIL(
                  'Parameter "{}" was required but not provided.'.format(self.name)
              )
          return params.get(self.name, self.default)
  
      # Serialization ------------------------------------------------------------
  
      def serialize(self) -> Dict[str, Any]:
          """
          Creates a serialized representation of this parameter
  
          Returns:
              - dict representing this parameter
          """
          return prefect.serialization.task.ParameterSchema().dump(self)
  