docs: def _pipeline_cell(args, cell_body):
      """Implements the pipeline subcommand in the %%bq magic.
      Args:
        args: the arguments following '%%bq pipeline'.
        cell_body: Cell contents.
      """
      name = args.get('name')
      if name is None:
          raise Exception('Pipeline name was not specified.')
  
      import google.datalab.utils as utils
      bq_pipeline_config = utils.commands.parse_config(
        cell_body, utils.commands.notebook_environment())
  
      try:
        airflow_spec = \
          google.datalab.contrib.bigquery.commands.get_airflow_spec_from_config(name,
                                                                                bq_pipeline_config)
      except AttributeError:
        return "Perhaps you're missing: import google.datalab.contrib.bigquery.commands"
  
      # If a gcs_dag_bucket is specified, we deploy to it so that the Airflow VM rsyncs it.
      error_message = ''
      gcs_dag_bucket = args.get('gcs_dag_bucket')
      gcs_dag_file_path = args.get('gcs_dag_file_path')
      if gcs_dag_bucket:
        try:
          airflow = google.datalab.contrib.pipeline.airflow.Airflow(gcs_dag_bucket, gcs_dag_file_path)
          airflow.deploy(name, airflow_spec)
          error_message += ("Airflow pipeline successfully deployed! View dashboard for more "
                            "details.\n")
        except AttributeError:
          return "Perhaps you're missing: import google.datalab.contrib.pipeline.airflow"
  
      location = args.get('location')
      environment = args.get('environment')
  
      if location and environment:
        try:
          composer = google.datalab.contrib.pipeline.composer.Composer(location, environment)
          composer.deploy(name, airflow_spec)
          error_message += ("Composer pipeline successfully deployed! View dashboard for more "
                            "details.\n")
        except AttributeError:
          return "Perhaps you're missing: import google.datalab.contrib.pipeline.composer"
  
      if args.get('debug'):
        error_message += '\n\n' + airflow_spec
  
      return error_message