# Path: google/datalab/contrib/pipeline/composer/_api.py
# class Api(object):
#   """A helper class to issue Composer HTTP requests."""
# 
#   _ENDPOINT = 'https://composer.googleapis.com/v1alpha1'
#   _ENVIRONMENTS_PATH_FORMAT = '/projects/%s/locations/%s/environments/%s'
# 
#   @staticmethod
#   def get_environment_details(zone, environment):
#     """ Issues a request to Composer to get the environment details.
# 
#     Args:
#       zone: GCP zone of the composer environment
#       environment: name of the Composer environment
#     Returns:
#       A parsed result object.
#     Raises:
#       Exception if there is an error performing the operation.
#     """
#     default_context = google.datalab.Context.default()
#     url = (Api._ENDPOINT + (Api._ENVIRONMENTS_PATH_FORMAT % (default_context.project_id, zone,
#                                                              environment)))
# 
#     return google.datalab.utils.Http.request(url, credentials=default_context.credentials)


import google.datalab.storage as storage
import re
from google.datalab.contrib.pipeline.composer._api import Api
  """ Represents a Composer object that encapsulates a set of functionality relating to the
  Cloud Composer service.

  This object can be used to generate the python airflow spec.
  """

  gcs_file_regexp = re.compile('gs://.*')

  def __init__(self, zone, environment):
    """ Initializes an instance of a Composer object.

    Args:
      zone: Zone in which Composer environment has been created.
      environment: Name of the Composer environment.
    """
    self._zone = zone
    self._environment = environment
    self._gcs_dag_location = None

  def deploy(self, name, dag_string):
    bucket_name, file_path = self.gcs_dag_location.split('/', 3)[2:]  # setting maxsplit to 3
    file_name = '{0}{1}.py'.format(file_path, name)

    bucket = storage.Bucket(bucket_name)
    file_object = bucket.object(file_name)
    file_object.write_stream(dag_string, 'text/plain')

  @property
  def gcs_dag_location(self):
    if not self._gcs_dag_location:

# Path: google/datalab/contrib/pipeline/composer/_api.py
# class Api(object):
#   """A helper class to issue Composer HTTP requests."""
# 
#   _ENDPOINT = 'https://composer.googleapis.com/v1alpha1'
#   _ENVIRONMENTS_PATH_FORMAT = '/projects/%s/locations/%s/environments/%s'
# 
#   @staticmethod
#   def get_environment_details(zone, environment):
#     """ Issues a request to Composer to get the environment details.
# 
#     Args:
#       zone: GCP zone of the composer environment
#       environment: name of the Composer environment
#     Returns:
#       A parsed result object.
#     Raises:
#       Exception if there is an error performing the operation.
#     """
#     default_context = google.datalab.Context.default()
#     url = (Api._ENDPOINT + (Api._ENVIRONMENTS_PATH_FORMAT % (default_context.project_id, zone,
#                                                              environment)))
# 
#     return google.datalab.utils.Http.request(url, credentials=default_context.credentials)


# Path: google/datalab/contrib/pipeline/composer/_composer.py
import google.datalab.storage as storage
import re
from google.datalab.contrib.pipeline.composer._api import Api

  """ Represents a Composer object that encapsulates a set of functionality relating to the
  Cloud Composer service.

  This object can be used to generate the python airflow spec.
  """

  gcs_file_regexp = re.compile('gs://.*')

  def __init__(self, zone, environment):
    """ Initializes an instance of a Composer object.

    Args:
      zone: Zone in which Composer environment has been created.
      environment: Name of the Composer environment.
    """
    self._zone = zone
    self._environment = environment
    self._gcs_dag_location = None

  def deploy(self, name, dag_string):
    bucket_name, file_path = self.gcs_dag_location.split('/', 3)[2:]  # setting maxsplit to 3
    file_name = '{0}{1}.py'.format(file_path, name)

    bucket = storage.Bucket(bucket_name)
    file_object = bucket.object(file_name)
    file_object.write_stream(dag_string, 'text/plain')

  @property
  def gcs_dag_location(self):
    if not self._gcs_dag_location:

      environment_details = Api.get_environment_details(self._zone, self._environment)