class_docstr: Create a copy of the provided flow id, as well as all contained recipes.
  
  :param dataprep_conn_id: The Dataprep connection ID
  :param flow_id: ID of the flow to be copied
  :param name: Name for the copy of the flow
  :param description: Description of the copy of the flow
  :param copy_datasources: Bool value to define should the copy of data inputs be made or not.
code: def execute(self, context: Context) -> dict:
      self.log.info("Copying flow with id %d...", self.flow_id)
      hook = GoogleDataprepHook(dataprep_conn_id=self.dataprep_conn_id)
      response = hook.copy_flow(
          flow_id=int(self.flow_id),
          name=self.name,
          description=self.description,
          copy_datasources=self.copy_datasources,
      )
  
      copied_flow_id = response.get("id")
      if self.project_id and copied_flow_id:
          DataprepFlowLink.persist(
              context=context,
              task_instance=self,
              project_id=self.project_id,
              flow_id=int(copied_flow_id),
          )
      return response
  
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Sequence
  from airflow.providers.google.cloud.hooks.dataprep import GoogleDataprepHook
  from airflow.providers.google.cloud.links.dataprep import DataprepFlowLink, DataprepJobGroupLink
  from airflow.providers.google.cloud.operators.cloud_base import GoogleCloudBaseOperator
label_desc: Use image node_id 2 for calling the DataprepCopyFlowOperator obj's underlying member method code with example usage: obj.execute(context) and returns: response