files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  ======================
  eHST Astroquery Module
  ======================
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  """
  from urllib.parse import urlencode
  
  from astropy import units
  from astropy.coordinates import SkyCoord
  from astropy.coordinates import Angle
  
  from astroquery.utils.tap import TapPlus
  from astroquery.query import BaseQuery
  import json
  import warnings
  from astropy.utils.exceptions import AstropyDeprecationWarning
  from astropy.utils.decorators import deprecated
  
  from . import conf
  from astroquery import log
  
  __all__ = ['ESAHubble', 'ESAHubbleClass']
  
  
  class ESAHubbleClass(BaseQuery):
      """
      Class to init ESA Hubble Module and communicate with eHST TAP
      """
      TIMEOUT = conf.TIMEOUT
      calibration_levels = {"AUXILIARY": 0, "RAW": 1, "CALIBRATED": 2,
                            "PRODUCT": 3}
      product_types = ["SCIENCE", "PREVIEW", "THUMBNAIL" or "AUXILIARY"]
      copying_string = "Copying file to {0}..."
  
      def __init__(self, *, tap_handler=None, show_messages=True):
          if tap_handler is None:
              self._tap = TapPlus(url=conf.EHST_TAP_SERVER,
                                  data_context='data', client_id="ASTROQUERY")
          else:
              self._tap = tap_handler
          if show_messages:
              self.get_status_messages()
  
      def download_product(self, observation_id, *, calibration_level=None,
                           filename=None, verbose=False, product_type=None):
          """
          Download products from EHST based on their observation ID and the
          calibration level or the product type.
  
          Parameters
          ----------
          observation_id : string
              id of the observation to be downloaded, mandatory
              The identifier of the observation we want to retrieve, regardless
              of whether it is simple or composite.
          calibration_level : string
              calibration level, optional
              The identifier of the data reduction/processing applied to the
              data. By default, the most scientifically relevant level will be
              chosen. RAW, CALIBRATED, PRODUCT or AUXILIARY
          filename : string
              file name to be used to store the artifact, optional, default
              None
              File name for the observation.
          verbose : bool
              optional, default 'False'
              flag to display information about the process
          product_type : string
              type of product retrieval, optional
              SCIENCE, PREVIEW, THUMBNAIL or AUXILIARY
              ------------
              Deprecation Warning: PRODUCT, SCIENCE_PRODUCT or POSTCARD
              are no longer supported.
              ------------
  
          Returns
          -------
          None. It downloads the observation indicated
          """
          if product_type and product_type in ['PRODUCT', 'SCIENCE_PRODUCT', 'POSTCARD']:
              warnings.warn(
                  "PRODUCT, SCIENCE_PRODUCT or POSTCARD product types are no longer supported. "
                  "Please use SCIENCE, PREVIEW, THUMBNAIL or AUXILIARY instead.",
                  AstropyDeprecationWarning)
  
          params = {"OBSERVATIONID": observation_id,
                    "TAPCLIENT": "ASTROQUERY",
                    "RETRIEVAL_TYPE": "OBSERVATION"}
  
          if filename is None:
              filename = observation_id + ".tar"
  
          if calibration_level:
              params["CALIBRATIONLEVEL"] = calibration_level
  
          # Product type check to ensure backwards compatibility
          product_type = self.__set_product_type(product_type)
          if product_type:
              self.__validate_product_type(product_type)
              params["PRODUCTTYPE"] = product_type
  
          filename = self._get_product_filename(product_type, filename)
          self._tap.load_data(params, filename, verbose=verbose)
  
          return filename
  
      def __set_product_type(self, product_type):
          if product_type:
              if 'SCIENCE_PRODUCT' in product_type:
                  return 'SCIENCE'
              elif 'PRODUCT' in product_type:
                  return None
              elif 'POSTCARD' in product_type:
                  return 'PREVIEW'
          return product_type
  
      def get_member_observations(self, observation_id):
          """
          Returns the related members of simple and composite observations
  
          Parameters
          ----------
          observation_id : str
              Observation identifier.
  
          Returns
          -------
          A list of strings with the observation_id of the associated
          observations
          """
          observation_type = self.get_observation_type(observation_id)
  
          if 'Composite' in observation_type:
              oids = self._select_related_members(observation_id)
          elif 'Simple' in observation_type:
              oids = self._select_related_composite(observation_id)
          else:
              raise ValueError("Invalid observation id")
          return oids
  
      def get_hap_hst_link(self, observation_id):
          """
          Returns the related members of hap and hst observations
  
          Parameters
          ----------
          observation_id : string
             id of the observation to be downloaded, mandatory
             The identifier of the observation we want to retrieve, regardless
             of whether it is simple or composite.
  
          Returns
          -------
          A list of strings with the observation_id of the associated
          observations
          """
          observation_type = self.get_observation_type(observation_id)
          if 'Composite' in observation_type:
              raise ValueError("HAP-HST link is only available for simple observations. Input observation is Composite.")
          elif 'HAP' in observation_type:
              oids = self._select_related_members(observation_id)
          elif 'HST' in observation_type:
              query = ("select observation_id from ehst.observation where obs_type='HAP Simple' "
                       f"and members like '%{observation_id}%'")
              job = self.query_tap(query=query)
              oids = job["observation_id"].pformat(show_name=False)
          else:
              raise ValueError("Invalid observation id")
          return oids
  
      def get_observation_type(self, observation_id):
          """
          Returns the type of an observation
  
          Parameters
          ----------
          observation_id : string
             id of the observation to be downloaded, mandatory
             The identifier of the observation we want to retrieve, regardless
             of whether it is simple or composite.
  
          Returns
          -------
          String with the observation type
          """
          if observation_id is None:
              raise ValueError("Please input an observation id")
  
          query = f"select obs_type from ehst.observation where observation_id='{observation_id}'"
          job = self.query_tap(query=query)
          if any(job["obs_type"]):
              obs_type = self._get_decoded_string(string=job["obs_type"][0])
          else:
              raise ValueError("Invalid Observation ID")
          return obs_type
  
      def _select_related_members(self, observation_id):
          query = f"select members from ehst.observation where observation_id='{observation_id}'"
          job = self.query_tap(query=query)
          oids = self._get_decoded_string(string=job["members"][0]).replace("caom:HST/", "").split(" ")
          return oids
  
      def _select_related_composite(self, observation_id):
          query = f"select observation_id from ehst.observation where members like '%{observation_id}%'"
          job = self.query_tap(query=query)
          oids = job["observation_id"].pformat(show_name=False)
          return oids
  
      def __validate_product_type(self, product_type):
          if (product_type not in self.product_types):
              raise ValueError("This product_type is not allowed")
  
      def _get_product_filename(self, product_type, filename):
          if (product_type == "PRODUCT"):
              return filename
          elif (product_type == "SCIENCE"):
              log.info("This is a SCIENCE_PRODUCT, the filename will be "
                       f"renamed to {filename}.fits.gz")
              return f"{filename}.fits.gz"
          elif (product_type == "THUMBNAIL" or product_type == "PREVIEW"):
              log.info("This is a POSTCARD, the filename will be "
                       f"renamed to {filename}.jpg")
              return f"{filename}.jpg"
  
          return filename
  
      def get_artifact(self, artifact_id, *, filename=None, verbose=False):
          """
          Download artifacts from EHST. Artifact is a single Hubble product file.
  
          Parameters
          ----------
          artifact_id : string
              id of the artifact to be downloaded, mandatory
              The identifier of the physical product (file) we want to retrieve.
          filename : string
              file name to be used to store the artifact, optional, default None
              File name for the artifact
          verbose : bool
              optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          None. It downloads the artifact indicated
          """
  
          params = {"RETRIEVAL_TYPE": "PRODUCT", "ARTIFACTID": artifact_id, "TAPCLIENT": "ASTROQUERY"}
          if
  
  [... the rest is omitted ...]
files: astroquery/esa/hubble/core.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  ======================
  eHST Astroquery Module
  ======================
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  """
  
  from astropy import units
  from astropy.coordinates import SkyCoord
  from astropy.coordinates import Angle
  
  from astroquery.utils.tap import TapPlus
  from astroquery.query import BaseQuery
  import json
  import warnings
  from astropy.utils.exceptions import AstropyDeprecationWarning
  from astropy.utils.decorators import deprecated
  
  from . import conf
  from astroquery import log
  
  __all__ = ['ESAHubble', 'ESAHubbleClass']
  
  
  class ESAHubbleClass(BaseQuery):
      """
      Class to init ESA Hubble Module and communicate with eHST TAP
      """
      TIMEOUT = conf.TIMEOUT
      calibration_levels = {"AUXILIARY": 0, "RAW": 1, "CALIBRATED": 2,
                            "PRODUCT": 3}
      product_types = ["SCIENCE", "PREVIEW", "THUMBNAIL" or "AUXILIARY"]
      copying_string = "Copying file to {0}..."
  
      def __init__(self, *, tap_handler=None, show_messages=True):
          if tap_handler is None:
              self._tap = TapPlus(url=conf.EHST_TAP_SERVER,
                                  data_context='data', client_id="ASTROQUERY")
          else:
              self._tap = tap_handler
          if show_messages:
              self.get_status_messages()
  
      def download_product(self, observation_id, *, calibration_level=None,
                           filename=None, verbose=False, product_type=None):
          """
          Download products from EHST based on their observation ID and the
          calibration level or the product type.
  
          Parameters
          ----------
          observation_id : string
              id of the observation to be downloaded, mandatory
              The identifier of the observation we want to retrieve, regardless
              of whether it is simple or composite.
          calibration_level : string
              calibration level, optional
              The identifier of the data reduction/processing applied to the
              data. By default, the most scientifically relevant level will be
              chosen. RAW, CALIBRATED, PRODUCT or AUXILIARY
          filename : string
              file name to be used to store the artifact, optional, default
              None
              File name for the observation.
          verbose : bool
              optional, default 'False'
              flag to display information about the process
          product_type : string
              type of product retrieval, optional
              SCIENCE, PREVIEW, THUMBNAIL or AUXILIARY
              ------------
              Deprecation Warning: PRODUCT, SCIENCE_PRODUCT or POSTCARD
              are no longer supported.
              ------------
  
          Returns
          -------
          None. It downloads the observation indicated
          """
          if product_type and product_type in ['PRODUCT', 'SCIENCE_PRODUCT', 'POSTCARD']:
              warnings.warn(
                  "PRODUCT, SCIENCE_PRODUCT or POSTCARD product types are no longer supported. "
                  "Please use SCIENCE, PREVIEW, THUMBNAIL or AUXILIARY instead.",
                  AstropyDeprecationWarning)
  
          params = {"OBSERVATIONID": observation_id,
                    "TAPCLIENT": "ASTROQUERY",
                    "RETRIEVAL_TYPE": "OBSERVATION"}
  
          if filename is None:
              filename = observation_id + ".tar"
  
          if calibration_level:
              params["CALIBRATIONLEVEL"] = calibration_level
  
          # Product type check to ensure backwards compatibility
          product_type = self.__set_product_type(product_type)
          if product_type:
              self.__validate_product_type(product_type)
              params["PRODUCTTYPE"] = product_type
  
          filename = self._get_product_filename(product_type, filename)
          self._tap.load_data(params, filename, verbose=verbose)
  
          return filename
  
      def __set_product_type(self, product_type):
          if product_type:
              if 'SCIENCE_PRODUCT' in product_type:
                  return 'SCIENCE'
              elif 'PRODUCT' in product_type:
                  return None
              elif 'POSTCARD' in product_type:
                  return 'PREVIEW'
          return product_type
  
      def get_member_observations(self, observation_id):
          """
          Returns the related members of simple and composite observations
  
          Parameters
          ----------
          observation_id : str
              Observation identifier.
  
          Returns
          -------
          A list of strings with the observation_id of the associated
          observations
          """
          observation_type = self.get_observation_type(observation_id)
  
          if 'Composite' in observation_type:
              oids = self._select_related_members(observation_id)
          elif 'Simple' in observation_type:
              oids = self._select_related_composite(observation_id)
          else:
              raise ValueError("Invalid observation id")
          return oids
  
      def get_hap_hst_link(self, observation_id):
          """
          Returns the related members of hap and hst observations
  
          Parameters
          ----------
          observation_id : string
             id of the observation to be downloaded, mandatory
             The identifier of the observation we want to retrieve, regardless
             of whether it is simple or composite.
  
          Returns
          -------
          A list of strings with the observation_id of the associated
          observations
          """
          observation_type = self.get_observation_type(observation_id)
          if 'Composite' in observation_type:
              raise ValueError("HAP-HST link is only available for simple observations. Input observation is Composite.")
          elif 'HAP' in observation_type:
              oids = self._select_related_members(observation_id)
          elif 'HST' in observation_type:
              query = ("select observation_id from ehst.observation where obs_type='HAP Simple' "
                       f"and members like '%{observation_id}%'")
              job = self.query_tap(query=query)
              oids = job["observation_id"].pformat(show_name=False)
          else:
              raise ValueError("Invalid observation id")
          return oids
  
      def get_observation_type(self, observation_id):
          """
          Returns the type of an observation
  
          Parameters
          ----------
          observation_id : string
             id of the observation to be downloaded, mandatory
             The identifier of the observation we want to retrieve, regardless
             of whether it is simple or composite.
  
          Returns
          -------
          String with the observation type
          """
          if observation_id is None:
              raise ValueError("Please input an observation id")
  
          query = f"select obs_type from ehst.observation where observation_id='{observation_id}'"
          job = self.query_tap(query=query)
          if any(job["obs_type"]):
              obs_type = self._get_decoded_string(string=job["obs_type"][0])
          else:
              raise ValueError("Invalid Observation ID")
          return obs_type
  
      def _select_related_members(self, observation_id):
          query = f"select members from ehst.observation where observation_id='{observation_id}'"
          job = self.query_tap(query=query)
          oids = self._get_decoded_string(string=job["members"][0]).replace("caom:HST/", "").split(" ")
          return oids
  
      def _select_related_composite(self, observation_id):
          query = f"select observation_id from ehst.observation where members like '%{observation_id}%'"
          job = self.query_tap(query=query)
          oids = job["observation_id"].pformat(show_name=False)
          return oids
  
      def __validate_product_type(self, product_type):
          if (product_type not in self.product_types):
              raise ValueError("This product_type is not allowed")
  
      def _get_product_filename(self, product_type, filename):
          if (product_type == "PRODUCT"):
              return filename
          elif (product_type == "SCIENCE"):
              log.info("This is a SCIENCE_PRODUCT, the filename will be "
                       f"renamed to {filename}.fits.gz")
              return f"{filename}.fits.gz"
          elif (product_type == "THUMBNAIL" or product_type == "PREVIEW"):
              log.info("This is a POSTCARD, the filename will be "
                       f"renamed to {filename}.jpg")
              return f"{filename}.jpg"
  
          return filename
  
      def get_artifact(self, artifact_id, *, filename=None, verbose=False):
          """
          Download artifacts from EHST. Artifact is a single Hubble product file.
  
          Parameters
          ----------
          artifact_id : string
              id of the artifact to be downloaded, mandatory
              The identifier of the physical product (file) we want to retrieve.
          filename : string
              file name to be used to store the artifact, optional, default None
              File name for the artifact
          verbose : bool
              optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          None. It downloads the artifact indicated
          """
  
          params = {"RETRIEVAL_TYPE": "PRODUCT", "ARTIFACTID": artifact_id, "TAPCLIENT": "ASTROQUERY"}
          if filename is None:
              filename = artifact
  
  [... the rest is omitted ...]
files: astroquery/esa/hubble/core.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =======================
  eJWST Astroquery Module
  =======================
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  """
  
  import binascii
  import gzip
  import os
  import shutil
  import tarfile
  import zipfile
  from datetime import datetime
  from urllib.parse import urlencode
  
  from astropy import log
  from astropy import units
  from astropy.coordinates import SkyCoord
  from astropy.table import vstack
  from astropy.units import Quantity
  from requests.exceptions import ConnectionError
  
  from astroquery.exceptions import RemoteServiceError
  from astroquery.ipac.ned import Ned
  from astroquery.query import BaseQuery
  from astroquery.simbad import Simbad
  from astroquery.utils import commons
  from astroquery.utils.tap import TapPlus
  from astroquery.vizier import Vizier
  from . import conf
  from .data_access import JwstDataHandler
  
  
  __all__ = ['Jwst', 'JwstClass']
  
  
  class JwstClass(BaseQuery):
  
      """
      Proxy class to default TapPlus object (pointing to JWST Archive)
      """
  
      JWST_DEFAULT_COLUMNS = ['observationid', 'calibrationlevel', 'public',
                              'dataproducttype', 'instrument_name',
                              'energy_bandpassname', 'target_name', 'target_ra',
                              'target_dec', 'position_bounds_center',
                              'position_bounds_spoly']
  
      PLANE_DATAPRODUCT_TYPES = ['image', 'cube', 'measurements', 'spectrum']
      ARTIFACT_PRODUCT_TYPES = ['info', 'thumbnail', 'auxiliary', 'science',
                                'preview']
      INSTRUMENT_NAMES = ['NIRISS', 'NIRSPEC', 'NIRCAM', 'MIRI', 'FGS']
      TARGET_RESOLVERS = ['ALL', 'SIMBAD', 'NED', 'VIZIER']
      CAL_LEVELS = ['ALL', 1, 2, 3, -1]
      REQUESTED_OBSERVATION_ID = "Missing required argument: 'observation_id'"
  
      def __init__(self, *, tap_plus_handler=None, data_handler=None, show_messages=True):
          if tap_plus_handler is None:
              self.__jwsttap = TapPlus(url=conf.JWST_TAP_SERVER,
                                       data_context='data', client_id='ASTROQUERY')
          else:
              self.__jwsttap = tap_plus_handler
  
          if data_handler is None:
              self.__jwstdata = JwstDataHandler(
                  base_url=conf.JWST_DATA_SERVER)
          else:
              self.__jwstdata = data_handler
  
          if show_messages:
              self.get_status_messages()
  
      def load_tables(self, *, only_names=False, include_shared_tables=False,
                      verbose=False):
          """Loads all public tables
          TAP & TAP+
  
          Parameters
          ----------
          only_names : bool, TAP+ only, optional, default 'False'
              True to load table names only
          include_shared_tables : bool, TAP+, optional, default 'False'
              True to include shared tables
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of table objects
          """
          return self.__jwsttap.load_tables(only_names,
                                            include_shared_tables,
                                            verbose)
  
      def load_table(self, table, *, verbose=False):
          """Loads the specified table
          TAP+ only
  
          Parameters
          ----------
          table : str, mandatory
              full qualified table name (i.e. schema name + table name)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A table object
          """
          return self.__jwsttap.load_table(table, verbose)
  
      def launch_job(self, query, *, name=None, output_file=None,
                     output_format="votable", verbose=False, dump_to_file=False,
                     background=False, upload_resource=None, upload_table_name=None,
                     async_job=False):
          """Launches a synchronous or asynchronous job
          TAP & TAP+
  
          Parameters
          ----------
          query : str, mandatory
              query to be executed
          name : str, optional, default None
              name of the job to be executed
          output_file : str, optional, default None
              file name where the results are saved if dumpToFile is True.
              If this parameter is not provided, the jobid is used instead
          output_format : str, optional, default 'votable'
              results format. Options are:
              'votable': str, binary VOTable format
              'csv': str, comma-separated values format
              'fits': str, FITS format
          verbose : bool, optional, default 'False'
              flag to display information about the process
          dump_to_file : bool, optional, default 'False'
              if True, the results are saved in a file instead of using memory
          background : bool, optional, default 'False'
              when the job is executed in asynchronous mode, this flag specifies
              whether the execution will wait until results are available
          upload_resource: str, optional, default None
              resource to be uploaded to UPLOAD_SCHEMA
          upload_table_name: str, required if uploadResource is provided
              Default None
              resource temporary table name associated to the uploaded resource
          async_job: bool, optional, default 'False'
              tag to execute the job in sync or async mode
  
          Returns
          -------
          A Job object
          """
          if async_job:
              return (self.__jwsttap.launch_job_async(query,
                                                      name=name,
                                                      output_file=output_file,
                                                      output_format=output_format,
                                                      verbose=verbose,
                                                      dump_to_file=dump_to_file,
                                                      background=background,
                                                      upload_resource=upload_resource,
                                                      upload_table_name=upload_table_name))
          else:
              return self.__jwsttap.launch_job(query,
                                               name=name,
                                               output_file=output_file,
                                               output_format=output_format,
                                               verbose=verbose,
                                               dump_to_file=dump_to_file,
                                               upload_resource=upload_resource,
                                               upload_table_name=upload_table_name)
  
      def load_async_job(self, *, jobid=None, name=None, verbose=False):
          """Loads an asynchronous job
          TAP & TAP+
  
          Parameters
          ----------
          jobid : str, mandatory if no name is provided, default None
              job identifier
          name : str, mandatory if no jobid is provided, default None
              job name
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A Job object
          """
          return self.__jwsttap.load_async_job(jobid, name, verbose)
  
      def search_async_jobs(self, *, jobfilter=None, verbose=False):
          """Searches for jobs applying the specified filter
          TAP+ only
  
          Parameters
          ----------
          jobfilter : JobFilter, optional, default None
              job filter
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of Job objects
          """
          return self.__jwsttap.search_async_jobs(jobfilter, verbose)
  
      def list_async_jobs(self, *, verbose=False):
          """Returns all the asynchronous jobs
          TAP & TAP+
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of Job objects
          """
          return self.__jwsttap.list_async_jobs(verbose)
  
      def query_region(self, coordinate, *,
                       radius=None,
                       width=None,
                       height=None,
                       observation_id=None,
                       cal_level="Top",
                       prod_type=None,
                       instrument_name=None,
                       filter_name=None,
                       proposal_id=None,
                       only_public=False,
                       show_all_columns=False,
                       async_job=False, verbose=False):
          """Launches a query region job in sync/async mode
          TAP & TAP+
  
          Parameters
          ----------
          coordinate : astropy.coordinate, mandatory
              coordinates center point
          radius : astropy.units, required if no 'width' nor 'height'
              are provided
              radius (deg)
          width : astropy.units, required if no 'radius' is provided
              box width
          height : astropy.units, required if no 'radius' is provided
              box height
          observation_id : str, optional, default None
              get the observation given by its ID.
          cal_level : object, optional, default 'Top'
              get the planes with the given calibration level. Options are:
              'Top': str, only the planes with the highest calibration level
              1,2,3: int, the given calibration level
          prod_type : str, optional, default None
  
  [... the rest is omitted ...]
files: astroquery/esa/jwst/core.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =======================
  eJWST Astroquery Module
  =======================
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  """
  
  import binascii
  import gzip
  import os
  import shutil
  import tarfile
  import zipfile
  from datetime import datetime
  
  from astropy import log
  from astropy import units
  from astropy.coordinates import SkyCoord
  from astropy.table import vstack
  from astropy.units import Quantity
  from requests.exceptions import ConnectionError
  
  from astroquery.exceptions import RemoteServiceError
  from astroquery.ipac.ned import Ned
  from astroquery.query import BaseQuery
  from astroquery.simbad import Simbad
  from astroquery.utils import commons
  from astroquery.utils.tap import TapPlus
  from astroquery.vizier import Vizier
  from . import conf
  from .data_access import JwstDataHandler
  
  
  __all__ = ['Jwst', 'JwstClass']
  
  
  class JwstClass(BaseQuery):
  
      """
      Proxy class to default TapPlus object (pointing to JWST Archive)
      """
  
      JWST_DEFAULT_COLUMNS = ['observationid', 'calibrationlevel', 'public',
                              'dataproducttype', 'instrument_name',
                              'energy_bandpassname', 'target_name', 'target_ra',
                              'target_dec', 'position_bounds_center',
                              'position_bounds_spoly']
  
      PLANE_DATAPRODUCT_TYPES = ['image', 'cube', 'measurements', 'spectrum']
      ARTIFACT_PRODUCT_TYPES = ['info', 'thumbnail', 'auxiliary', 'science',
                                'preview']
      INSTRUMENT_NAMES = ['NIRISS', 'NIRSPEC', 'NIRCAM', 'MIRI', 'FGS']
      TARGET_RESOLVERS = ['ALL', 'SIMBAD', 'NED', 'VIZIER']
      CAL_LEVELS = ['ALL', 1, 2, 3, -1]
      REQUESTED_OBSERVATION_ID = "Missing required argument: 'observation_id'"
  
      def __init__(self, *, tap_plus_handler=None, data_handler=None, show_messages=True):
          if tap_plus_handler is None:
              self.__jwsttap = TapPlus(url=conf.JWST_TAP_SERVER,
                                       data_context='data', client_id='ASTROQUERY')
          else:
              self.__jwsttap = tap_plus_handler
  
          if data_handler is None:
              self.__jwstdata = JwstDataHandler(
                  base_url=conf.JWST_DATA_SERVER)
          else:
              self.__jwstdata = data_handler
  
          if show_messages:
              self.get_status_messages()
  
      def load_tables(self, *, only_names=False, include_shared_tables=False,
                      verbose=False):
          """Loads all public tables
          TAP & TAP+
  
          Parameters
          ----------
          only_names : bool, TAP+ only, optional, default 'False'
              True to load table names only
          include_shared_tables : bool, TAP+, optional, default 'False'
              True to include shared tables
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of table objects
          """
          return self.__jwsttap.load_tables(only_names,
                                            include_shared_tables,
                                            verbose)
  
      def load_table(self, table, *, verbose=False):
          """Loads the specified table
          TAP+ only
  
          Parameters
          ----------
          table : str, mandatory
              full qualified table name (i.e. schema name + table name)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A table object
          """
          return self.__jwsttap.load_table(table, verbose)
  
      def launch_job(self, query, *, name=None, output_file=None,
                     output_format="votable", verbose=False, dump_to_file=False,
                     background=False, upload_resource=None, upload_table_name=None,
                     async_job=False):
          """Launches a synchronous or asynchronous job
          TAP & TAP+
  
          Parameters
          ----------
          query : str, mandatory
              query to be executed
          name : str, optional, default None
              name of the job to be executed
          output_file : str, optional, default None
              file name where the results are saved if dumpToFile is True.
              If this parameter is not provided, the jobid is used instead
          output_format : str, optional, default 'votable'
              results format. Options are:
              'votable': str, binary VOTable format
              'csv': str, comma-separated values format
              'fits': str, FITS format
          verbose : bool, optional, default 'False'
              flag to display information about the process
          dump_to_file : bool, optional, default 'False'
              if True, the results are saved in a file instead of using memory
          background : bool, optional, default 'False'
              when the job is executed in asynchronous mode, this flag specifies
              whether the execution will wait until results are available
          upload_resource: str, optional, default None
              resource to be uploaded to UPLOAD_SCHEMA
          upload_table_name: str, required if uploadResource is provided
              Default None
              resource temporary table name associated to the uploaded resource
          async_job: bool, optional, default 'False'
              tag to execute the job in sync or async mode
  
          Returns
          -------
          A Job object
          """
          if async_job:
              return (self.__jwsttap.launch_job_async(query,
                                                      name=name,
                                                      output_file=output_file,
                                                      output_format=output_format,
                                                      verbose=verbose,
                                                      dump_to_file=dump_to_file,
                                                      background=background,
                                                      upload_resource=upload_resource,
                                                      upload_table_name=upload_table_name))
          else:
              return self.__jwsttap.launch_job(query,
                                               name=name,
                                               output_file=output_file,
                                               output_format=output_format,
                                               verbose=verbose,
                                               dump_to_file=dump_to_file,
                                               upload_resource=upload_resource,
                                               upload_table_name=upload_table_name)
  
      def load_async_job(self, *, jobid=None, name=None, verbose=False):
          """Loads an asynchronous job
          TAP & TAP+
  
          Parameters
          ----------
          jobid : str, mandatory if no name is provided, default None
              job identifier
          name : str, mandatory if no jobid is provided, default None
              job name
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A Job object
          """
          return self.__jwsttap.load_async_job(jobid, name, verbose)
  
      def search_async_jobs(self, *, jobfilter=None, verbose=False):
          """Searches for jobs applying the specified filter
          TAP+ only
  
          Parameters
          ----------
          jobfilter : JobFilter, optional, default None
              job filter
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of Job objects
          """
          return self.__jwsttap.search_async_jobs(jobfilter, verbose)
  
      def list_async_jobs(self, *, verbose=False):
          """Returns all the asynchronous jobs
          TAP & TAP+
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of Job objects
          """
          return self.__jwsttap.list_async_jobs(verbose)
  
      def query_region(self, coordinate, *,
                       radius=None,
                       width=None,
                       height=None,
                       observation_id=None,
                       cal_level="Top",
                       prod_type=None,
                       instrument_name=None,
                       filter_name=None,
                       proposal_id=None,
                       only_public=False,
                       show_all_columns=False,
                       async_job=False, verbose=False):
          """Launches a query region job in sync/async mode
          TAP & TAP+
  
          Parameters
          ----------
          coordinate : astropy.coordinate, mandatory
              coordinates center point
          radius : astropy.units, required if no 'width' nor 'height'
              are provided
              radius (deg)
          width : astropy.units, required if no 'radius' is provided
              box width
          height : astropy.units, required if no 'radius' is provided
              box height
          observation_id : str, optional, default None
              get the observation given by its ID.
          cal_level : object, optional, default 'Top'
              get the planes with the given calibration level. Options are:
              'Top': str, only the planes with the highest calibration level
              1,2,3: int, the given calibration level
          prod_type : str, optional, default None
              get the observations providing the given
  
  [... the rest is omitted ...]
files: astroquery/esa/jwst/core.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  Gaia TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  CONTENT_TYPE_POST_DEFAULT = "application/x-www-form-urlencoded"
  
  
  class DummyTapHandler:
  
      def __init__(self):
          self.__invokedMethod = None
          self.__parameters = {}
  
      def reset(self):
          self.__parameters = {}
          self.__invokedMethod = None
  
      def check_call(self, method_name, parameters):
          self.check_method(method_name)
          self.check_parameters(parameters, method_name)
  
      def execute_upload(self, data,
                         content_type=CONTENT_TYPE_POST_DEFAULT,
                         verbose=False):
          pass
  
      def check_method(self, method):
          if method == self.__invokedMethod:
              return
          else:
              raise Exception(f"Method '{method}' not invoked. (Invoked method is "
                              f"{self.__invokedMethod}')")
  
      def check_parameters(self, parameters, method_name):
          if parameters is None:
              return len(self.__parameters) == 0
          num_params_in_call = len(self.__parameters)
          num_params_expected = len(parameters)
          print("Num parameters found in call: {}, expected: {}".
                format(num_params_in_call, num_params_expected))
          if num_params_expected != num_params_in_call:
              raise Exception("Wrong number of parameters for method '%s'. \
              Found: %d. Expected %d",
                              (method_name,
                               len(self.__parameters),
                               len(parameters)))
          for key in parameters:
              if key in self.__parameters:
                  # check value
                  if self.__parameters[key] != parameters[key]:
                      k = self.__parameters[key]
                      raise Exception(f"Wrong '{method_name}' parameter value for "
                                      f"method '{key}'. Found: '{k}'. "
                                      f"Expected: '{parameters[key]}'")
              else:
                  raise Exception(f"Parameter '{key}' not found for method '{method_name}'")
          return False
  
      def load_tables(self, only_names=False, include_shared_tables=False,
                      verbose=False):
          self.__invokedMethod = 'load_tables'
          self.__parameters['only_names'] = only_names
          self.__parameters['include_shared_tables'] = include_shared_tables
          self.__parameters['verbose'] = verbose
  
      def load_table(self, table, verbose=False):
          self.__invokedMethod = 'load_table'
          self.__parameters['table'] = table
          self.__parameters['verbose'] = verbose
  
      def launch_job(self, query, name=None, output_file=None,
                     output_format="votable", verbose=False, dump_to_file=False,
                     upload_resource=None, upload_table_name=None):
          self.__invokedMethod = 'launch_job'
          self.__parameters['query'] = query
          self.__parameters['name'] = name
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['upload_resource'] = upload_resource
          self.__parameters['upload_table_name'] = upload_table_name
  
      def launch_job_async(self, query, name=None, output_file=None,
                           output_format="votable", verbose=False,
                           dump_to_file=False, background=False,
                           upload_resource=None, upload_table_name=None):
          self.__invokedMethod = 'launch_job_async'
          self.__parameters['query'] = query
          self.__parameters['name'] = name
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['background'] = background
          self.__parameters['upload_resource'] = upload_resource
          self.__parameters['upload_table_name'] = upload_table_name
  
      def load_async_job(self, jobid=None, name=None, verbose=False):
          self.__invokedMethod = 'load_async_job'
          self.__parameters['jobid'] = jobid
          self.__parameters['name'] = name
          self.__parameters['verbose'] = verbose
  
      def search_async_jobs(self, jobfilter=None, verbose=False):
          self.__invokedMethod = 'search_async_jobs'
          self.__parameters['jobfilter'] = jobfilter
          self.__parameters['verbose'] = verbose
  
      def list_async_jobs(self, verbose=False):
          self.__invokedMethod = 'list_async_jobs'
          self.__parameters['verbose'] = verbose
  
      def query_object(self, coordinate, radius=None, width=None, height=None,
                       verbose=False):
          self.__invokedMethod = 'query_object'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
          self.__parameters['height'] = height
          self.__parameters['verbose'] = verbose
  
      def query_object_async(self, coordinate, radius=None, width=None,
                             height=None, verbose=False):
          self.__invokedMethod = 'query_object_async'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
          self.__parameters['height'] = height
          self.__parameters['verbose'] = verbose
  
      def query_region(self, coordinate, radius=None, width=None):
          self.__invokedMethod = 'query_region'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
  
      def query_region_async(self, coordinate, radius=None, width=None):
          self.__invokedMethod = 'query_region_async'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
  
      def get_images(self, coordinate):
          self.__invokedMethod = 'get_images'
          self.__parameters['coordinate'] = coordinate
  
      def get_images_async(self, coordinate):
          self.__invokedMethod = 'get_images_sync'
          self.__parameters['coordinate'] = coordinate
  
      def cone_search(self, coordinate, radius, output_file=None,
                      output_format="votable", verbose=False,
                      dump_to_file=False, columns=[]):
          self.__invokedMethod = 'cone_search'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['columns'] = columns
  
      def cone_search_async(self, coordinate, radius, background=False,
                            output_file=None, output_format="votable",
                            verbose=False, dump_to_file=False, columns=[]):
          self.__invokedMethod = 'cone_search_async'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['background'] = background
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['columns'] = columns
  
      def remove_jobs(self, jobs_list, verbose=False):
          self.__invokedMethod = 'remove_jobs'
          self.__parameters['jobs_list'] = jobs_list
          self.__parameters['verbose'] = verbose
  
      def save_results(self, job, verbose=False):
          self.__invokedMethod = 'save_results'
          self.__parameters['job'] = job
          self.__parameters['verbose'] = verbose
  
      def login(self, user=None, password=None, credentials_file=None,
                verbose=False):
          self.__invokedMethod = 'login'
          self.__parameters['user'] = verbose
          self.__parameters['password'] = verbose
          self.__parameters['credentials_file'] = verbose
          self.__parameters['verbose'] = verbose
  
      def login_gui(self, verbose=False):
          self.__invokedMethod = 'login_gui'
          self.__parameters['verbose'] = verbose
  
      def logout(self, verbose=False):
          self.__invokedMethod = 'logout'
          self.__parameters['verbose'] = verbose
  
      def load_data(self, params_dict, output_file=None, verbose=False):
          self.__invokedMethod = 'load_data'
          self.__parameters['params_dict'] = params_dict
          self.__parameters['output_file'] = output_file
          self.__parameters['verbose'] = verbose
  
      def get_datalinks(self, ids, verbose=False):
          self.__invokedMethod = 'get_datalinks'
          self
  
  [... the rest is omitted ...]
files: astroquery/gaia/tests/DummyTapHandler.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  Gaia TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  from urllib.parse import urlencode
  
  CONTENT_TYPE_POST_DEFAULT = "application/x-www-form-urlencoded"
  
  
  class DummyTapHandler:
  
      def __init__(self):
          self.__invokedMethod = None
          self.__parameters = {}
  
      def reset(self):
          self.__parameters = {}
          self.__invokedMethod = None
  
      def check_call(self, method_name, parameters):
          self.check_method(method_name)
          self.check_parameters(parameters, method_name)
  
      def execute_upload(self, data,
                         content_type=CONTENT_TYPE_POST_DEFAULT,
                         verbose=False):
          pass
  
      def check_method(self, method):
          if method == self.__invokedMethod:
              return
          else:
              raise Exception(f"Method '{method}' not invoked. (Invoked method is "
                              f"{self.__invokedMethod}')")
  
      def check_parameters(self, parameters, method_name):
          if parameters is None:
              return len(self.__parameters) == 0
          num_params_in_call = len(self.__parameters)
          num_params_expected = len(parameters)
          print("Num parameters found in call: {}, expected: {}".
                format(num_params_in_call, num_params_expected))
          if num_params_expected != num_params_in_call:
              raise Exception("Wrong number of parameters for method '%s'. \
              Found: %d. Expected %d",
                              (method_name,
                               len(self.__parameters),
                               len(parameters)))
          for key in parameters:
              if key in self.__parameters:
                  # check value
                  if self.__parameters[key] != parameters[key]:
                      k = self.__parameters[key]
                      raise Exception(f"Wrong '{method_name}' parameter value for "
                                      f"method '{key}'. Found: '{k}'. "
                                      f"Expected: '{parameters[key]}'")
              else:
                  raise Exception(f"Parameter '{key}' not found for method '{method_name}'")
          return False
  
      def load_tables(self, only_names=False, include_shared_tables=False,
                      verbose=False):
          self.__invokedMethod = 'load_tables'
          self.__parameters['only_names'] = only_names
          self.__parameters['include_shared_tables'] = include_shared_tables
          self.__parameters['verbose'] = verbose
  
      def load_table(self, table, verbose=False):
          self.__invokedMethod = 'load_table'
          self.__parameters['table'] = table
          self.__parameters['verbose'] = verbose
  
      def launch_job(self, query, name=None, output_file=None,
                     output_format="votable", verbose=False, dump_to_file=False,
                     upload_resource=None, upload_table_name=None):
          self.__invokedMethod = 'launch_job'
          self.__parameters['query'] = query
          self.__parameters['name'] = name
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['upload_resource'] = upload_resource
          self.__parameters['upload_table_name'] = upload_table_name
  
      def launch_job_async(self, query, name=None, output_file=None,
                           output_format="votable", verbose=False,
                           dump_to_file=False, background=False,
                           upload_resource=None, upload_table_name=None):
          self.__invokedMethod = 'launch_job_async'
          self.__parameters['query'] = query
          self.__parameters['name'] = name
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['background'] = background
          self.__parameters['upload_resource'] = upload_resource
          self.__parameters['upload_table_name'] = upload_table_name
  
      def load_async_job(self, jobid=None, name=None, verbose=False):
          self.__invokedMethod = 'load_async_job'
          self.__parameters['jobid'] = jobid
          self.__parameters['name'] = name
          self.__parameters['verbose'] = verbose
  
      def search_async_jobs(self, jobfilter=None, verbose=False):
          self.__invokedMethod = 'search_async_jobs'
          self.__parameters['jobfilter'] = jobfilter
          self.__parameters['verbose'] = verbose
  
      def list_async_jobs(self, verbose=False):
          self.__invokedMethod = 'list_async_jobs'
          self.__parameters['verbose'] = verbose
  
      def query_object(self, coordinate, radius=None, width=None, height=None,
                       verbose=False):
          self.__invokedMethod = 'query_object'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
          self.__parameters['height'] = height
          self.__parameters['verbose'] = verbose
  
      def query_object_async(self, coordinate, radius=None, width=None,
                             height=None, verbose=False):
          self.__invokedMethod = 'query_object_async'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
          self.__parameters['height'] = height
          self.__parameters['verbose'] = verbose
  
      def query_region(self, coordinate, radius=None, width=None):
          self.__invokedMethod = 'query_region'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
  
      def query_region_async(self, coordinate, radius=None, width=None):
          self.__invokedMethod = 'query_region_async'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['width'] = width
  
      def get_images(self, coordinate):
          self.__invokedMethod = 'get_images'
          self.__parameters['coordinate'] = coordinate
  
      def get_images_async(self, coordinate):
          self.__invokedMethod = 'get_images_sync'
          self.__parameters['coordinate'] = coordinate
  
      def cone_search(self, coordinate, radius, output_file=None,
                      output_format="votable", verbose=False,
                      dump_to_file=False, columns=[]):
          self.__invokedMethod = 'cone_search'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['columns'] = columns
  
      def cone_search_async(self, coordinate, radius, background=False,
                            output_file=None, output_format="votable",
                            verbose=False, dump_to_file=False, columns=[]):
          self.__invokedMethod = 'cone_search_async'
          self.__parameters['coordinate'] = coordinate
          self.__parameters['radius'] = radius
          self.__parameters['background'] = background
          self.__parameters['output_file'] = output_file
          self.__parameters['output_format'] = output_format
          self.__parameters['verbose'] = verbose
          self.__parameters['dump_to_file'] = dump_to_file
          self.__parameters['columns'] = columns
  
      def remove_jobs(self, jobs_list, verbose=False):
          self.__invokedMethod = 'remove_jobs'
          self.__parameters['jobs_list'] = jobs_list
          self.__parameters['verbose'] = verbose
  
      def save_results(self, job, verbose=False):
          self.__invokedMethod = 'save_results'
          self.__parameters['job'] = job
          self.__parameters['verbose'] = verbose
  
      def login(self, user=None, password=None, credentials_file=None,
                verbose=False):
          self.__invokedMethod = 'login'
          self.__parameters['user'] = verbose
          self.__parameters['password'] = verbose
          self.__parameters['credentials_file'] = verbose
          self.__parameters['verbose'] = verbose
  
      def login_gui(self, verbose=False):
          self.__invokedMethod = 'login_gui'
          self.__parameters['verbose'] = verbose
  
      def logout(self, verbose=False):
          self.__invokedMethod = 'logout'
          self.__parameters['verbose'] = verbose
  
      def load_data(self, params_dict, output_file=None, verbose=False):
          self.__invokedMethod = 'load_data'
          self.__parameters['params_dict'] = params_dict
          self.__parameters['output_file'] = output_file
          self.__parameters['verbose'] = verbose
  
      def get_datalinks(self, ids, verbose=False):
          self.__invokedMethod =
  
  [... the rest is omitted ...]
files: astroquery/gaia/tests/DummyTapHandler.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  
  try:
      # python 3
      import http.client as httplib
  except ImportError:
      # python 2
      import httplib
  import mimetypes
  import time
  
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap import taputils
  from astroquery import version
  
  import requests
  
  __all__ = ['TapConn']
  
  CONTENT_TYPE_POST_DEFAULT = "application/x-www-form-urlencoded"
  
  
  class TapConn:
      """TAP plus connection class
      Provides low level HTTP connection capabilities
      """
  
      def __init__(self, ishttps,
                   host,
                   server_context=None,
                   port=80,
                   sslport=443,
                   connhandler=None,
                   tap_context=None,
                   upload_context=None,
                   table_edit_context=None,
                   data_context=None,
                   datalink_context=None):
          """Constructor
  
          Parameters
          ----------
          ishttps: bool, mandatory
              'True' is the protocol to use is HTTPS
          host : str, mandatory
              host name
          server_context : str, mandatory
              server context
          tap_context : str, optional
              tap context
          upload_context : str, optional
              upload context
          table_edit_context : str, optional
              table edit context
          data_context : str, optional
              data context
          datalink_context : str, optional
              datalink context
          port : int, optional, default 80
              HTTP port
          sslport : int, optional, default 443
              HTTPS port
          connhandler connection handler object, optional, default None
              HTTP(s) connection hander (creator). If no handler is provided, a
              new one is created.
          """
          self.__interna_init()
          self.__isHttps = ishttps
          self.__connHost = host
          self.__connPort = port
          self.__connPortSsl = sslport
          if server_context is not None:
              if server_context.startswith("/"):
                  self.__serverContext = server_context
              else:
                  self.__serverContext = f"/{server_context}"
          else:
              self.__serverContext = ""
          self.__tapContext = self.__create_context(tap_context)
          self.__dataContext = self.__create_context(data_context)
          self.__datalinkContext = self.__create_context(datalink_context)
          self.__uploadContext = self.__create_context(upload_context)
          self.__tableEditContext = self.__create_context(table_edit_context)
          if connhandler is None:
              self.__connectionHandler = ConnectionHandler(self.__connHost,
                                                           self.__connPort,
                                                           self.__connPortSsl)
          else:
              self.__connectionHandler = connhandler
  
      def __create_context(self, context):
          if context is not None and context != "":
              if str(context).startswith("/"):
                  return f"{self.__serverContext}{context}"
              else:
                  return f"{self.__serverContext}/{context}"
          else:
              return self.__serverContext
  
      def __interna_init(self):
          self.__connectionHandler = None
          self.__isHttps = False
          self.__connHost = ""
          self.__connPort = 80
          self.__connPortSsl = 443
          self.__serverContext = None
          self.__tapContext = None
          self.__postHeaders = {
              "Content-type": CONTENT_TYPE_POST_DEFAULT,
              "Accept": "text/plain",
              "User-Agent": "astroquery/{vers}".format(vers=version.version),
          }
          self.__getHeaders = {}
          self.__cookie = None
          self.__currentStatus = 0
          self.__currentReason = ""
  
      def __get_tap_context(self, subContext):
          return f"{self.__tapContext}/{subContext}"
  
      def __get_data_context(self, encodedData=None):
          if self.__dataContext is None:
              raise ValueError("data_context must be specified at TAP object "
                               + "creation for this action to be performed")
          if encodedData is not None:
              return f"{self.__dataContext}?{encodedData}"
          else:
              return self.__dataContext
  
      def __get_datalink_context(self, subContext, encodedData=None):
          if self.__datalinkContext is None:
              raise ValueError("datalink_context must be specified at TAP "
                               + "object creation for this action to be "
                               + "performed")
          if encodedData is not None:
              return f"{self.__datalinkContext}/{subContext}?{encodedData}"
  
          else:
              return f"{self.__datalinkContext}/{subContext}"
  
      def __get_upload_context(self):
          if self.__uploadContext is None:
              raise ValueError("upload_context must be specified at TAP "
                               + "object creation for this action to be "
                               + "performed")
          return self.__uploadContext
  
      def __get_table_edit_context(self):
          if self.__tableEditContext is None:
              raise ValueError("table_edit_context must be specified at TAP "
                               + "object creation for this action to be "
                               + "performed")
          return self.__tableEditContext
  
      def __get_server_context(self, subContext):
          return f"{self.__serverContext}/{subContext}"
  
      def execute_tapget(self, subcontext, verbose=False):
          """Executes a TAP GET request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          subcontext : str, mandatory
              context to be added to host+serverContext+tapContext, usually the
              TAP list name
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          if subcontext.startswith("http"):
              # absolute url
              return self.__execute_get(subcontext, verbose)
          else:
              context = self.__get_tap_context(subcontext)
              return self.__execute_get(context, verbose)
  
      def execute_dataget(self, query, verbose=False):
          """Executes a data GET request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          query : str, mandatory
              URL encoded data (query string)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          context = self.__get_data_context(query)
          return self.__execute_get(context, verbose)
  
      def execute_datalinkget(self, subcontext, query, verbose=False):
          """Executes a datalink GET request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          subcontext : str, mandatory
              datalink subcontext
          query : str, mandatory
              URL encoded data (query string)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          context = self.__get_datalink_context(subcontext, query)
          return self.__execute_get(context, verbose)
  
      def __execute_get(self, context, verbose=False):
          conn = self.__get_connection(verbose)
          if verbose:
              print(f"host = {conn.host}:{conn.port}")
              print(f"context = {context}")
          conn.request("GET", context, None, self.__getHeaders)
          response = conn.getresponse()
          self.__currentReason = response.reason
          self.__currentStatus = response.status
          return response
  
      def execute_tappost(self, subcontext, data,
                          content_type=CONTENT_TYPE_POST_DEFAULT,
                          verbose=False):
          """Executes a POST request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          subcontext : str, mandatory
              context to be added to host+serverContext+tapContext, usually the
              TAP list name
          data : str, mandatory
              POST data
          content_type: str, optional, default: application/x-www-form-urlencoded
              HTTP(s) content-type header value
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          context = self.__get_tap_context(subcontext)
          return self.__execute_post(context, data, content_type, verbose)
  
      def execute_datapost(self, data,
                           content_type=CONTENT_TYPE_POST_DEFAULT,
                           verbose=False):
          """Executes a POST request
          The connection is done through HTTP or HTTPS depending on
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tapconn.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  
  try:
      # python 3
      import http.client as httplib
  except ImportError:
      # python 2
      import httplib
  import mimetypes
  import time
  
  from urllib.parse import urlencode
  
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap import taputils
  from astroquery import version
  
  import requests
  
  __all__ = ['TapConn']
  
  CONTENT_TYPE_POST_DEFAULT = "application/x-www-form-urlencoded"
  
  
  class TapConn:
      """TAP plus connection class
      Provides low level HTTP connection capabilities
      """
  
      def __init__(self, ishttps,
                   host,
                   server_context=None,
                   port=80,
                   sslport=443,
                   connhandler=None,
                   tap_context=None,
                   upload_context=None,
                   table_edit_context=None,
                   data_context=None,
                   datalink_context=None):
          """Constructor
  
          Parameters
          ----------
          ishttps: bool, mandatory
              'True' is the protocol to use is HTTPS
          host : str, mandatory
              host name
          server_context : str, mandatory
              server context
          tap_context : str, optional
              tap context
          upload_context : str, optional
              upload context
          table_edit_context : str, optional
              table edit context
          data_context : str, optional
              data context
          datalink_context : str, optional
              datalink context
          port : int, optional, default 80
              HTTP port
          sslport : int, optional, default 443
              HTTPS port
          connhandler connection handler object, optional, default None
              HTTP(s) connection hander (creator). If no handler is provided, a
              new one is created.
          """
          self.__interna_init()
          self.__isHttps = ishttps
          self.__connHost = host
          self.__connPort = port
          self.__connPortSsl = sslport
          if server_context is not None:
              if server_context.startswith("/"):
                  self.__serverContext = server_context
              else:
                  self.__serverContext = f"/{server_context}"
          else:
              self.__serverContext = ""
          self.__tapContext = self.__create_context(tap_context)
          self.__dataContext = self.__create_context(data_context)
          self.__datalinkContext = self.__create_context(datalink_context)
          self.__uploadContext = self.__create_context(upload_context)
          self.__tableEditContext = self.__create_context(table_edit_context)
          if connhandler is None:
              self.__connectionHandler = ConnectionHandler(self.__connHost,
                                                           self.__connPort,
                                                           self.__connPortSsl)
          else:
              self.__connectionHandler = connhandler
  
      def __create_context(self, context):
          if context is not None and context != "":
              if str(context).startswith("/"):
                  return f"{self.__serverContext}{context}"
              else:
                  return f"{self.__serverContext}/{context}"
          else:
              return self.__serverContext
  
      def __interna_init(self):
          self.__connectionHandler = None
          self.__isHttps = False
          self.__connHost = ""
          self.__connPort = 80
          self.__connPortSsl = 443
          self.__serverContext = None
          self.__tapContext = None
          self.__postHeaders = {
              "Content-type": CONTENT_TYPE_POST_DEFAULT,
              "Accept": "text/plain",
              "User-Agent": "astroquery/{vers}".format(vers=version.version),
          }
          self.__getHeaders = {}
          self.__cookie = None
          self.__currentStatus = 0
          self.__currentReason = ""
  
      def __get_tap_context(self, subContext):
          return f"{self.__tapContext}/{subContext}"
  
      def __get_data_context(self, encodedData=None):
          if self.__dataContext is None:
              raise ValueError("data_context must be specified at TAP object "
                               + "creation for this action to be performed")
          if encodedData is not None:
              return f"{self.__dataContext}?{encodedData}"
          else:
              return self.__dataContext
  
      def __get_datalink_context(self, subContext, encodedData=None):
          if self.__datalinkContext is None:
              raise ValueError("datalink_context must be specified at TAP "
                               + "object creation for this action to be "
                               + "performed")
          if encodedData is not None:
              return f"{self.__datalinkContext}/{subContext}?{encodedData}"
  
          else:
              return f"{self.__datalinkContext}/{subContext}"
  
      def __get_upload_context(self):
          if self.__uploadContext is None:
              raise ValueError("upload_context must be specified at TAP "
                               + "object creation for this action to be "
                               + "performed")
          return self.__uploadContext
  
      def __get_table_edit_context(self):
          if self.__tableEditContext is None:
              raise ValueError("table_edit_context must be specified at TAP "
                               + "object creation for this action to be "
                               + "performed")
          return self.__tableEditContext
  
      def __get_server_context(self, subContext):
          return f"{self.__serverContext}/{subContext}"
  
      def execute_tapget(self, subcontext, verbose=False):
          """Executes a TAP GET request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          subcontext : str, mandatory
              context to be added to host+serverContext+tapContext, usually the
              TAP list name
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          if subcontext.startswith("http"):
              # absolute url
              return self.__execute_get(subcontext, verbose)
          else:
              context = self.__get_tap_context(subcontext)
              return self.__execute_get(context, verbose)
  
      def execute_dataget(self, query, verbose=False):
          """Executes a data GET request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          query : str, mandatory
              URL encoded data (query string)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          context = self.__get_data_context(query)
          return self.__execute_get(context, verbose)
  
      def execute_datalinkget(self, subcontext, query, verbose=False):
          """Executes a datalink GET request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          subcontext : str, mandatory
              datalink subcontext
          query : str, mandatory
              URL encoded data (query string)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          context = self.__get_datalink_context(subcontext, query)
          return self.__execute_get(context, verbose)
  
      def __execute_get(self, context, verbose=False):
          conn = self.__get_connection(verbose)
          if verbose:
              print(f"host = {conn.host}:{conn.port}")
              print(f"context = {context}")
          conn.request("GET", context, None, self.__getHeaders)
          response = conn.getresponse()
          self.__currentReason = response.reason
          self.__currentStatus = response.status
          return response
  
      def execute_tappost(self, subcontext, data,
                          content_type=CONTENT_TYPE_POST_DEFAULT,
                          verbose=False):
          """Executes a POST request
          The connection is done through HTTP or HTTPS depending on the login
          status (logged in -> HTTPS)
  
          Parameters
          ----------
          subcontext : str, mandatory
              context to be added to host+serverContext+tapContext, usually the
              TAP list name
          data : str, mandatory
              POST data
          content_type: str, optional, default: application/x-www-form-urlencoded
              HTTP(s) content-type header value
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          An HTTP(s) response object
          """
          context = self.__get_tap_context(subcontext)
          return self.__execute_post(context, data, content_type, verbose)
  
      def execute_datapost(self, data,
                           content_type=CONTENT_TYPE_POST_DEFAULT,
                           verbose=False):
          """Executes a POST request
          The connection
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tapconn.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  from astroquery.utils.tap import taputils
  
  import requests
  
  
  class DummyConnHandler:
  
      def __init__(self):
          self.request = None
          self.data = None
          self.fileExt = ".ext"
          self.defaultResponse = None
          self.responses = {}
          self.errorFileOutput = None
          self.errorReceivedResponse = None
          self.contentType = None
          self.verbose = None
          self.query = None
          self.fileOutput = None
  
      def set_default_response(self, defaultResponse):
          self.defaultResponse = defaultResponse
  
      def get_default_response(self):
          return self.defaultResponse
  
      def get_last_request(self):
          return self.request
  
      def get_last_data(self):
          return self.data
  
      def get_last_query(self):
          return self.query
  
      def get_error_file_output(self):
          return self.errorFileOutput
  
      def get_error_received_response(self):
          return self.errorReceivedResponse
  
      def set_response(self, request, response):
          self.responses[str(request)] = response
  
      def execute_tapget(self, request=None, verbose=False):
          return self.__execute_get(request, verbose)
  
      def execute_dataget(self, query, verbose=False):
          return self.__execute_get(query)
  
      def execute_datalinkget(self, subcontext, query, verbose=False):
          self.query = query
          return self.__execute_get(subcontext, verbose)
  
      def __execute_get(self, request, verbose):
          self.request = request
          self.verbose = verbose
          return self.__get_response(request)
  
      def execute_tappost(self, subcontext=None, data=None,
                          content_type=None, verbose=False):
          return self.__execute_post(subcontext, data, content_type, verbose)
  
      def execute_datapost(self, data=None, content_type=None, verbose=False):
          return self.__execute_post("", data, content_type, verbose)
  
      def execute_datalinkpost(self, subcontext=None, data=None,
                               content_type=None, verbose=False):
          return self.__execute_post(subcontext, data, content_type, verbose)
  
      def __execute_post(self, subcontext=None, data=None,
                         content_type=None, verbose=False):
          self.data = data
          self.contentType = content_type
          self.verbose = verbose
          sortedKey = self.__create_sorted_dict_key(data)
          if subcontext.find('?') == -1:
              self.request = f"{subcontext}?{sortedKey}"
          else:
              if subcontext.endswith('?'):
                  self.request = f"{subcontext}{sortedKey}"
              else:
                  self.request = f"{subcontext}&{sortedKey}"
          return self.__get_response(self.request)
  
      def dump_to_file(self, fileOutput, response):
          self.errorFileOutput = fileOutput
          self.errorReceivedResponse = response
          print(f"DummyConnHandler - dump to file: file: '{fileOutput}', \
          response status: {response.status}, response msg: {response.reason}")
  
      def __get_response(self, responseid):
          try:
              return self.responses[str(responseid)]
          except KeyError as e:
              if self.defaultResponse is not None:
                  return self.defaultResponse
              else:
                  print(f"\nNot found response for key\n\t'{responseid}'")
                  print("Available keys: ")
                  if self.responses is None:
                      print("\tNone available")
                  else:
                      for k in self.responses.keys():
                          print(f"\t'{k}'")
                  raise (e)
  
      def __create_sorted_dict_key(self, data):
          dictTmp = {}
          items = data.split('&')
          for i in (items):
              subItems = i.split('=')
              dictTmp[subItems[0]] = subItems[1]
          # sort dict
          return taputils.taputil_create_sorted_dict_key(dictTmp)
  
      def check_launch_response_status(self, response, debug,
                                       expected_response_status,
                                       raise_exception=True):
          isError = False
          if response.status != expected_response_status:
              if debug:
                  print(f"ERROR: {response.status}: {response.reason}")
              isError = True
          if isError and raise_exception:
              errMsg = taputils.get_http_response_error(response)
              print(response.status, errMsg)
              raise requests.exceptions.HTTPError(errMsg)
          else:
              return isError
  
      def get_suitable_extension(self, headers):
          return self.fileExt
  
      def set_suitable_extension(self, ext):
          self.fileExt = ext
  
      def get_suitable_extension_by_format(self, output_format):
          return self.fileExt
  
      def get_file_from_header(self, headers):
          return self.fileOutput
  
      def find_header(self, headers, key):
          return taputils.taputil_find_header(headers, key)
  
      def execute_table_edit(self, data,
                             content_type="application/x-www-form-urlencoded",
                             verbose=False):
          return self.__execute_post(subcontext="tableEdit", data=data,
                                     content_type=content_type, verbose=verbose)
  
      def execute_table_tool(self, data,
                             content_type="application/x-www-form-urlencoded",
                             verbose=False):
          return self.__execute_post(subcontext="TableTool", data=data,
                                     content_type=content_type, verbose=verbose)
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tests/DummyConnHandler.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  from astroquery.utils.tap import taputils
  
  from urllib.parse import urlencode
  
  import requests
  
  
  class DummyConnHandler:
  
      def __init__(self):
          self.request = None
          self.data = None
          self.fileExt = ".ext"
          self.defaultResponse = None
          self.responses = {}
          self.errorFileOutput = None
          self.errorReceivedResponse = None
          self.contentType = None
          self.verbose = None
          self.query = None
          self.fileOutput = None
  
      def set_default_response(self, defaultResponse):
          self.defaultResponse = defaultResponse
  
      def get_default_response(self):
          return self.defaultResponse
  
      def get_last_request(self):
          return self.request
  
      def get_last_data(self):
          return self.data
  
      def get_last_query(self):
          return self.query
  
      def get_error_file_output(self):
          return self.errorFileOutput
  
      def get_error_received_response(self):
          return self.errorReceivedResponse
  
      def set_response(self, request, response):
          self.responses[str(request)] = response
  
      def execute_tapget(self, request=None, verbose=False):
          return self.__execute_get(request, verbose)
  
      def execute_dataget(self, query, verbose=False):
          return self.__execute_get(query)
  
      def execute_datalinkget(self, subcontext, query, verbose=False):
          self.query = query
          return self.__execute_get(subcontext, verbose)
  
      def __execute_get(self, request, verbose):
          self.request = request
          self.verbose = verbose
          return self.__get_response(request)
  
      def execute_tappost(self, subcontext=None, data=None,
                          content_type=None, verbose=False):
          return self.__execute_post(subcontext, data, content_type, verbose)
  
      def execute_datapost(self, data=None, content_type=None, verbose=False):
          return self.__execute_post("", data, content_type, verbose)
  
      def execute_datalinkpost(self, subcontext=None, data=None,
                               content_type=None, verbose=False):
          return self.__execute_post(subcontext, data, content_type, verbose)
  
      def __execute_post(self, subcontext=None, data=None,
                         content_type=None, verbose=False):
          self.data = data
          self.contentType = content_type
          self.verbose = verbose
          sortedKey = self.__create_sorted_dict_key(data)
          if subcontext.find('?') == -1:
              self.request = f"{subcontext}?{sortedKey}"
          else:
              if subcontext.endswith('?'):
                  self.request = f"{subcontext}{sortedKey}"
              else:
                  self.request = f"{subcontext}&{sortedKey}"
          return self.__get_response(self.request)
  
      def dump_to_file(self, fileOutput, response):
          self.errorFileOutput = fileOutput
          self.errorReceivedResponse = response
          print(f"DummyConnHandler - dump to file: file: '{fileOutput}', \
          response status: {response.status}, response msg: {response.reason}")
  
      def __get_response(self, responseid):
          try:
              return self.responses[str(responseid)]
          except KeyError as e:
              if self.defaultResponse is not None:
                  return self.defaultResponse
              else:
                  print(f"\nNot found response for key\n\t'{responseid}'")
                  print("Available keys: ")
                  if self.responses is None:
                      print("\tNone available")
                  else:
                      for k in self.responses.keys():
                          print(f"\t'{k}'")
                  raise (e)
  
      def __create_sorted_dict_key(self, data):
          dictTmp = {}
          items = data.split('&')
          for i in (items):
              subItems = i.split('=')
              dictTmp[subItems[0]] = subItems[1]
          # sort dict
          return taputils.taputil_create_sorted_dict_key(dictTmp)
  
      def check_launch_response_status(self, response, debug,
                                       expected_response_status,
                                       raise_exception=True):
          isError = False
          if response.status != expected_response_status:
              if debug:
                  print(f"ERROR: {response.status}: {response.reason}")
              isError = True
          if isError and raise_exception:
              errMsg = taputils.get_http_response_error(response)
              print(response.status, errMsg)
              raise requests.exceptions.HTTPError(errMsg)
          else:
              return isError
  
      def url_encode(self, data):
          return urlencode(data)
  
      def get_suitable_extension(self, headers):
          return self.fileExt
  
      def set_suitable_extension(self, ext):
          self.fileExt = ext
  
      def get_suitable_extension_by_format(self, output_format):
          return self.fileExt
  
      def get_file_from_header(self, headers):
          return self.fileOutput
  
      def find_header(self, headers, key):
          return taputils.taputil_find_header(headers, key)
  
      def execute_table_edit(self, data,
                             content_type="application/x-www-form-urlencoded",
                             verbose=False):
          return self.__execute_post(subcontext="tableEdit", data=data,
                                     content_type=content_type, verbose=verbose)
  
      def execute_table_tool(self, data,
                             content_type="application/x-www-form-urlencoded",
                             verbose=False):
          return self.__execute_post(subcontext="TableTool", data=data,
                                     content_type=content_type, verbose=verbose)
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tests/DummyConnHandler.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  Modified on 1 jun. 2021 by mhsarmiento
  """
  from urllib.parse import urlencode
  
  from astroquery.utils.tap import taputils
  from astroquery.utils.tap.conn.tapconn import TapConn
  from astroquery.utils.tap.xmlparser.tableSaxParser import TableSaxParser
  from astroquery.utils.tap.model.job import Job
  from astroquery.utils.tap.gui.login import LoginDialog
  from astroquery.utils.tap.xmlparser.jobSaxParser import JobSaxParser
  from astroquery.utils.tap.xmlparser.jobListSaxParser import JobListSaxParser
  from astroquery.utils.tap.xmlparser.groupSaxParser import GroupSaxParser
  from astroquery.utils.tap.xmlparser.sharedItemsSaxParser import SharedItemsSaxParser  # noqa
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap.model.filter import Filter
  import requests
  from astroquery import log
  import getpass
  import os
  from astropy.table.table import Table
  import tempfile
  
  __all__ = ['Tap', 'TapPlus']
  
  VERSION = "20200428.1"
  
  
  class Tap:
      """TAP class
      Provides TAP capabilities
      """
  
      def __init__(self, url=None,
                   host=None,
                   server_context=None,
                   tap_context=None,
                   port=80, sslport=443,
                   default_protocol_is_https=False,
                   connhandler=None,
                   upload_context=None,
                   table_edit_context=None,
                   data_context=None,
                   datalink_context=None,
                   verbose=False):
          """Constructor
  
          Parameters
          ----------
          url : str, mandatory if no host is specified, default None
              TAP URL
          host : str, optional, default None
              host name
          server_context : str, mandatory, default None
              server context
          tap_context : str, mandatory, default None
              tap context
          upload_context : str, optional, default None
              upload context
          table_edit_context : str, mandatory, default None
              context for all actions to be performed over a existing table
          data_context : str, optional, default None
              data context
          datalink_context : str, optional, default None
              datalink context
          port : int, optional, default '80'
              HTTP port
          sslport : int, optional, default '443'
              HTTPS port
          default_protocol_is_https : bool, optional, default False
              Specifies whether the default protocol to be used is HTTPS
          connhandler : connection handler object, optional, default None
              HTTP(s) connection hander (creator). If no handler is provided, a
              new one is created.
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          self.__internalInit()
          if url is not None:
              protocol, host, port, server, tap = self.__parseUrl(url)
              if server_context is None:
                  server_context = server
              if tap_context is None:
                  tap_context = tap
              if protocol == "http":
                  tap = TapConn(ishttps=False,
                                host=host,
                                server_context=server_context,
                                tap_context=tap_context,
                                upload_context=upload_context,
                                table_edit_context=table_edit_context,
                                data_context=data_context,
                                datalink_context=datalink_context,
                                port=port,
                                sslport=sslport)
                  self.__connHandler = tap
              else:
                  # https port -> sslPort
                  tap = TapConn(ishttps=True,
                                host=host,
                                server_context=server_context,
                                tap_context=tap_context,
                                upload_context=upload_context,
                                table_edit_context=table_edit_context,
                                data_context=data_context,
                                datalink_context=datalink_context,
                                port=port,
                                sslport=port)
                  self.__connHandler = tap
          else:
              tap = TapConn(ishttps=default_protocol_is_https,
                            host=host,
                            server_context=server_context,
                            tap_context=tap_context,
                            upload_context=upload_context,
                            table_edit_context=table_edit_context,
                            data_context=data_context,
                            datalink_context=datalink_context,
                            port=port,
                            sslport=sslport)
              self.__connHandler = tap
          # if connectionHandler is set, use it (useful for testing)
          if connhandler is not None:
              self.__connHandler = connhandler
          if verbose:
              print(f"Created TAP+ (v{VERSION}) - Connection:\n{self.__connHandler}")
  
      def __internalInit(self):
          self.__connHandler = None
          self.tap_client_id = f"aqtappy1-{VERSION}"
  
      def load_tables(self, verbose=False):
          """Loads all public tables
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of table objects
          """
          return self.__load_tables(verbose=verbose)
  
      def load_table(self, table, verbose=False):
          """Loads the specified table
  
          Parameters
          ----------
          table : str, mandatory
              full qualified table name (i.e. schema name + table name)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A table object
          """
          if table is None:
              raise ValueError("Table name is required")
          if verbose:
              print(f"Retrieving table '{table}'")
          response = self.__connHandler.execute_tapget(f"tables?tables={table}",
                                                       verbose=verbose)
          if verbose:
              print(response.status, response.reason)
          self.__connHandler.check_launch_response_status(response,
                                                          verbose,
                                                          200)
          if verbose:
              print(f"Parsing table '{table}'...")
          tsp = TableSaxParser()
          tsp.parseData(response)
          if verbose:
              print("Done.")
          return tsp.get_table()
  
      def __load_tables(self, only_names=False, include_shared_tables=False,
                        verbose=False):
          """Loads all public tables
  
          Parameters
          ----------
          only_names : bool, TAP+ only, optional, default 'False'
              True to load table names only
          include_shared_tables : bool, TAP+, optional, default 'False'
              True to include shared tables
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of table objects
          """
          # share_info=true&share_accessible=true&only_tables=true
          flags = ""
          addedItem = False
          if only_names:
              flags = "only_tables=true"
              addedItem = True
          if include_shared_tables:
              if addedItem:
                  flags += "&"
              flags += "share_accessible=true"
              addedItem = True
          log.info("Retrieving tables...")
          if flags != "":
              response = self.__connHandler.execute_tapget(f"tables?{flags}", verbose=verbose)
          else:
              response = self.__connHandler.execute_tapget("tables",
                                                           verbose=verbose)
          if verbose:
              print(response.status, response.reason)
          isError = self.__connHandler.check_launch_response_status(response,
                                                                    verbose,
                                                                    200)
          if isError:
              log.info(f"{response.status} {response.reason}")
              raise requests.exceptions.HTTPError(response.reason)
              return None
          log.info("Parsing tables...")
          tsp = TableSaxParser()
          tsp.parseData(response)
          log.info("Done.")
          return tsp.get_tables()
  
      def launch_job(self, query, name=None, output_file=None,
                     output_format="votable", verbose=False,
                     dump_to_file=False, upload_resource=None,
                     upload_table_name=None, maxrec=None):
          """Launches a synchronous job
  
          Parameters
          ----------
          query : str, mandatory
              query to be executed
          name : str, optional, default None
              custom name defined by the user for the job that is going to be created
          output_file : str, optional, default None
              file name where the results are saved if dumpToFile is True.
              If this parameter is not provided, the jobid is used instead
          output_format : str, optional, default 'votable'
              results format
          verbose : bool, optional, default 'False'
              flag to display information about the process
          dump_to_file : bool, optional, default 'False'
              if True, the results are saved in a file instead of using memory
          upload_resource : str, optional, default None
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/core.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  Modified on 1 jun. 2021 by mhsarmiento
  """
  from astroquery.utils.tap import taputils
  from astroquery.utils.tap.conn.tapconn import TapConn
  from astroquery.utils.tap.xmlparser.tableSaxParser import TableSaxParser
  from astroquery.utils.tap.model.job import Job
  from astroquery.utils.tap.gui.login import LoginDialog
  from astroquery.utils.tap.xmlparser.jobSaxParser import JobSaxParser
  from astroquery.utils.tap.xmlparser.jobListSaxParser import JobListSaxParser
  from astroquery.utils.tap.xmlparser.groupSaxParser import GroupSaxParser
  from astroquery.utils.tap.xmlparser.sharedItemsSaxParser import SharedItemsSaxParser  # noqa
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap.model.filter import Filter
  import requests
  from astroquery import log
  import getpass
  import os
  from astropy.table.table import Table
  import tempfile
  
  __all__ = ['Tap', 'TapPlus']
  
  VERSION = "20200428.1"
  
  
  class Tap:
      """TAP class
      Provides TAP capabilities
      """
  
      def __init__(self, url=None,
                   host=None,
                   server_context=None,
                   tap_context=None,
                   port=80, sslport=443,
                   default_protocol_is_https=False,
                   connhandler=None,
                   upload_context=None,
                   table_edit_context=None,
                   data_context=None,
                   datalink_context=None,
                   verbose=False):
          """Constructor
  
          Parameters
          ----------
          url : str, mandatory if no host is specified, default None
              TAP URL
          host : str, optional, default None
              host name
          server_context : str, mandatory, default None
              server context
          tap_context : str, mandatory, default None
              tap context
          upload_context : str, optional, default None
              upload context
          table_edit_context : str, mandatory, default None
              context for all actions to be performed over a existing table
          data_context : str, optional, default None
              data context
          datalink_context : str, optional, default None
              datalink context
          port : int, optional, default '80'
              HTTP port
          sslport : int, optional, default '443'
              HTTPS port
          default_protocol_is_https : bool, optional, default False
              Specifies whether the default protocol to be used is HTTPS
          connhandler : connection handler object, optional, default None
              HTTP(s) connection hander (creator). If no handler is provided, a
              new one is created.
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          self.__internalInit()
          if url is not None:
              protocol, host, port, server, tap = self.__parseUrl(url)
              if server_context is None:
                  server_context = server
              if tap_context is None:
                  tap_context = tap
              if protocol == "http":
                  tap = TapConn(ishttps=False,
                                host=host,
                                server_context=server_context,
                                tap_context=tap_context,
                                upload_context=upload_context,
                                table_edit_context=table_edit_context,
                                data_context=data_context,
                                datalink_context=datalink_context,
                                port=port,
                                sslport=sslport)
                  self.__connHandler = tap
              else:
                  # https port -> sslPort
                  tap = TapConn(ishttps=True,
                                host=host,
                                server_context=server_context,
                                tap_context=tap_context,
                                upload_context=upload_context,
                                table_edit_context=table_edit_context,
                                data_context=data_context,
                                datalink_context=datalink_context,
                                port=port,
                                sslport=port)
                  self.__connHandler = tap
          else:
              tap = TapConn(ishttps=default_protocol_is_https,
                            host=host,
                            server_context=server_context,
                            tap_context=tap_context,
                            upload_context=upload_context,
                            table_edit_context=table_edit_context,
                            data_context=data_context,
                            datalink_context=datalink_context,
                            port=port,
                            sslport=sslport)
              self.__connHandler = tap
          # if connectionHandler is set, use it (useful for testing)
          if connhandler is not None:
              self.__connHandler = connhandler
          if verbose:
              print(f"Created TAP+ (v{VERSION}) - Connection:\n{self.__connHandler}")
  
      def __internalInit(self):
          self.__connHandler = None
          self.tap_client_id = f"aqtappy1-{VERSION}"
  
      def load_tables(self, verbose=False):
          """Loads all public tables
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of table objects
          """
          return self.__load_tables(verbose=verbose)
  
      def load_table(self, table, verbose=False):
          """Loads the specified table
  
          Parameters
          ----------
          table : str, mandatory
              full qualified table name (i.e. schema name + table name)
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A table object
          """
          if table is None:
              raise ValueError("Table name is required")
          if verbose:
              print(f"Retrieving table '{table}'")
          response = self.__connHandler.execute_tapget(f"tables?tables={table}",
                                                       verbose=verbose)
          if verbose:
              print(response.status, response.reason)
          self.__connHandler.check_launch_response_status(response,
                                                          verbose,
                                                          200)
          if verbose:
              print(f"Parsing table '{table}'...")
          tsp = TableSaxParser()
          tsp.parseData(response)
          if verbose:
              print("Done.")
          return tsp.get_table()
  
      def __load_tables(self, only_names=False, include_shared_tables=False,
                        verbose=False):
          """Loads all public tables
  
          Parameters
          ----------
          only_names : bool, TAP+ only, optional, default 'False'
              True to load table names only
          include_shared_tables : bool, TAP+, optional, default 'False'
              True to include shared tables
          verbose : bool, optional, default 'False'
              flag to display information about the process
  
          Returns
          -------
          A list of table objects
          """
          # share_info=true&share_accessible=true&only_tables=true
          flags = ""
          addedItem = False
          if only_names:
              flags = "only_tables=true"
              addedItem = True
          if include_shared_tables:
              if addedItem:
                  flags += "&"
              flags += "share_accessible=true"
              addedItem = True
          log.info("Retrieving tables...")
          if flags != "":
              response = self.__connHandler.execute_tapget(f"tables?{flags}", verbose=verbose)
          else:
              response = self.__connHandler.execute_tapget("tables",
                                                           verbose=verbose)
          if verbose:
              print(response.status, response.reason)
          isError = self.__connHandler.check_launch_response_status(response,
                                                                    verbose,
                                                                    200)
          if isError:
              log.info(f"{response.status} {response.reason}")
              raise requests.exceptions.HTTPError(response.reason)
              return None
          log.info("Parsing tables...")
          tsp = TableSaxParser()
          tsp.parseData(response)
          log.info("Done.")
          return tsp.get_tables()
  
      def launch_job(self, query, name=None, output_file=None,
                     output_format="votable", verbose=False,
                     dump_to_file=False, upload_resource=None,
                     upload_table_name=None, maxrec=None):
          """Launches a synchronous job
  
          Parameters
          ----------
          query : str, mandatory
              query to be executed
          name : str, optional, default None
              custom name defined by the user for the job that is going to be created
          output_file : str, optional, default None
              file name where the results are saved if dumpToFile is True.
              If this parameter is not provided, the jobid is used instead
          output_format : str, optional, default 'votable'
              results format
          verbose : bool, optional, default 'False'
              flag to display information about the process
          dump_to_file : bool, optional, default 'False'
              if True, the results are saved in a file instead of using memory
          upload_resource : str, optional, default None
              resource to be uploaded to U
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/core.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  
  import time
  from urllib.parse import urlencode
  
  from astroquery.utils.tap.model import modelutils
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap import taputils
  from astropy.logger import log
  import requests
  
  
  __all__ = ['Job']
  
  
  class Job:
      """Job class
      """
  
      def __init__(self, async_job, query=None, connhandler=None):
          """Constructor
  
          Parameters
          ----------
          async_job : bool, mandatory
              'True' if the job is asynchronous
          query : str, optional, default None
              Query
          connhandler : TapConn, optional, default None
              Connection handler
          """
          # async is a reserved keyword starting python 3.7
          self.async_ = async_job
          self.connHandler = None
          self.isFinished = None
          self.jobid = None
          self.remoteLocation = None
          # phase is actually indended to be private as get_phase is non-trivial
          self._phase = None
          self.outputFile = None
          self.outputFileUser = None
          self.responseStatus = 0
          self.responseMsg = None
          self.results = None
          self.__resultInMemory = False    # only used within class
          self.failed = False
          self.runid = None
          self.ownerid = None
          self.startTime = None
          self.endTime = None
          self.creationTime = None
          self.executionDuration = None
          self.destruction = None
          self.locationId = None
          self.name = None
          self.quote = None
  
          self.connHandler = connhandler
          self.parameters = {}
          self.parameters['query'] = query
          # default output format
          self.parameters['format'] = 'votable'
  
      def set_phase(self, phase):
          """Sets the job phase
  
          Parameters
          ----------
          phase : str, mandatory
              job phase
          """
          if self.is_finished():
              raise ValueError("Cannot assign a phase when a job is finished")
          self._phase = phase
  
      def start(self, verbose=False):
          """Starts the job (allowed in PENDING phase only)
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          self.__change_phase(phase="RUN", verbose=verbose)
  
      def abort(self, verbose=False):
          """Aborts the job (allowed in PENDING phase only)
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          self.__change_phase(phase="ABORT", verbose=verbose)
  
      def __change_phase(self, phase, verbose=False):
          if self._phase == 'PENDING':
              context = f"async/{self.jobid}/phase"
              response = self.connHandler.execute_tappost(
                  subcontext=context, data=urlencode({"PHASE": phase}), verbose=verbose
              )
              if verbose:
                  print(response.status, response.reason)
                  print(response.getheaders())
              self.__last_phase_response_status = response.status
              if phase == 'RUN':
                  # a request for RUN does not mean the server executes the job
                  phase = 'QUEUED'
                  if response.status != 200 and response.status != 303:
                      errMsg = taputils.get_http_response_error(response)
                      print(response.status, errMsg)
                      raise requests.exceptions.HTTPError(errMsg)
              else:
                  if response.status != 200:
                      errMsg = taputils.get_http_response_error(response)
                      print(response.status, errMsg)
                      raise requests.exceptions.HTTPError(errMsg)
              self._phase = phase
              return response
          else:
              raise ValueError(f"Cannot start a job in phase: {self._phase}")
  
      def send_parameter(self, name=None, value=None, verbose=False):
          """Sends a job parameter (allowed in PENDING phase only).
  
          Parameters
          ----------
          name : string
              Parameter name.
          value : string
              Parameter value.
          """
          if self._phase == 'PENDING':
              # send post parameter/value
              context = f"async/{self.jobid}"
              response = self.connHandler.execute_tappost(subcontext=context,
                                                          data=urlencode({name: value}),
                                                          verbose=verbose)
              if verbose:
                  print(response.status, response.reason)
                  print(response.getheaders())
              self.__last_phase_response_status = response.status
              if response.status != 200:
                  errMsg = taputils.get_http_response_error(response)
                  print(response.status, errMsg)
                  raise requests.exceptions.HTTPError(errMsg)
              return response
          else:
              raise ValueError(f"Cannot start a job in phase: {self._phase}")
  
      def get_phase(self, update=False):
          """Returns the job phase. May optionally update the job's phase.
  
          Parameters
          ----------
          update : bool
              if True, the phase will by updated by querying the server before
              returning.
  
          Returns
          -------
          The job phase
          """
          if update:
              phase_request = f"async/{self.jobid}/phase"
              response = self.connHandler.execute_tapget(phase_request)
  
              self.__last_phase_response_status = response.status
              if response.status != 200:
                  errMsg = taputils.get_http_response_error(response)
                  print(response.status, errMsg)
                  raise requests.exceptions.HTTPError(errMsg)
  
              self._phase = str(response.read().decode('utf-8'))
          return self._phase
  
      def set_response_status(self, status, msg):
          """Sets the HTTP(s) connection status
  
          Parameters
          ----------
          status : int, mandatory
              HTTP(s) response status
          msg : str, mandatory
              HTTP(s) response message
          """
          self.__responseStatus = status
          self.__responseMsg = msg
  
      def get_data(self):
          """Returns the job results (Astroquery API specification)
          This method will block if the job is asynchronous and the job has not
          finished yet.
  
          Returns
          -------
          The job results (astropy.table).
          """
          return self.get_results()
  
      def get_results(self):
          """Returns the job results
          This method will block if the job is asynchronous and the job has not
          finished yet.
  
          Returns
          -------
          The job results (astropy.table).
          """
          if self.results is not None:
              return self.results
          # try load results from file
          # read_results_table_from_file checks whether
          # the file already exists or not
          outputFormat = self.parameters['format']
          results = modelutils.read_results_table_from_file(self.outputFile,
                                                            outputFormat)
          if results is not None:
              self.results = results
              return results
          # Try to load from server: only async
          if not self.async_:
              # sync: result is in a file
              return None
          else:
              # async: result is in the server once the job is finished
              self.__load_async_job_results()
              return self.results
  
      def set_results(self, results):
          """Sets the job results
  
          Parameters
          ----------
          results : Table object, mandatory
              job results
          """
          self.results = results
          self.__resultInMemory = True
  
      def save_results(self, verbose=False):
          """Saves job results
          If the job is asynchronous, this method will block until the results
          are available.
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          if self.__resultInMemory:
              if verbose:
                  print(f"Saving results to: {self.outputFile}")
              self.results.to_xml(self.outputFile)
          else:
              if not self.async_:
                  # sync: cannot access server again
                  log.info("No results to save")
              else:
                  # Async
                  self.wait_for_job_end(verbose)
                  response = self.connHandler.execute_tapget(
                      f"async/{self.jobid}/results/result")
                  if verbose:
                      print(response.status, response.reason)
                      print(response.getheaders())
                  isError = self.connHandler.\
                      check_launch_response_status(response,
                                                   verbose,
                                                   200)
                  if isError:
                      print(response.reason)
                      raise Exception(response.reason)
                  if self.outputFileUser is None:
                      # User did not provide an output
                      # The output is a temporary one, analyse header
                      self.outputFile = taputils.get_suitable_output_file(
                          self.
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/model/job.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  
  import time
  
  from astroquery.utils.tap.model import modelutils
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap import taputils
  from astropy.logger import log
  import requests
  
  
  __all__ = ['Job']
  
  
  class Job:
      """Job class
      """
  
      def __init__(self, async_job, query=None, connhandler=None):
          """Constructor
  
          Parameters
          ----------
          async_job : bool, mandatory
              'True' if the job is asynchronous
          query : str, optional, default None
              Query
          connhandler : TapConn, optional, default None
              Connection handler
          """
          # async is a reserved keyword starting python 3.7
          self.async_ = async_job
          self.connHandler = None
          self.isFinished = None
          self.jobid = None
          self.remoteLocation = None
          # phase is actually indended to be private as get_phase is non-trivial
          self._phase = None
          self.outputFile = None
          self.outputFileUser = None
          self.responseStatus = 0
          self.responseMsg = None
          self.results = None
          self.__resultInMemory = False    # only used within class
          self.failed = False
          self.runid = None
          self.ownerid = None
          self.startTime = None
          self.endTime = None
          self.creationTime = None
          self.executionDuration = None
          self.destruction = None
          self.locationId = None
          self.name = None
          self.quote = None
  
          self.connHandler = connhandler
          self.parameters = {}
          self.parameters['query'] = query
          # default output format
          self.parameters['format'] = 'votable'
  
      def set_phase(self, phase):
          """Sets the job phase
  
          Parameters
          ----------
          phase : str, mandatory
              job phase
          """
          if self.is_finished():
              raise ValueError("Cannot assign a phase when a job is finished")
          self._phase = phase
  
      def start(self, verbose=False):
          """Starts the job (allowed in PENDING phase only)
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          self.__change_phase(phase="RUN", verbose=verbose)
  
      def abort(self, verbose=False):
          """Aborts the job (allowed in PENDING phase only)
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          self.__change_phase(phase="ABORT", verbose=verbose)
  
      def __change_phase(self, phase, verbose=False):
          if self._phase == 'PENDING':
              context = f"async/{self.jobid}/phase"
              args = {
                  "PHASE": str(phase)}
              data = self.connHandler.url_encode(args)
              response = self.connHandler.execute_tappost(subcontext=context,
                                                          data=data,
                                                          verbose=verbose)
              if verbose:
                  print(response.status, response.reason)
                  print(response.getheaders())
              self.__last_phase_response_status = response.status
              if phase == 'RUN':
                  # a request for RUN does not mean the server executes the job
                  phase = 'QUEUED'
                  if response.status != 200 and response.status != 303:
                      errMsg = taputils.get_http_response_error(response)
                      print(response.status, errMsg)
                      raise requests.exceptions.HTTPError(errMsg)
              else:
                  if response.status != 200:
                      errMsg = taputils.get_http_response_error(response)
                      print(response.status, errMsg)
                      raise requests.exceptions.HTTPError(errMsg)
              self._phase = phase
              return response
          else:
              raise ValueError(f"Cannot start a job in phase: {self._phase}")
  
      def send_parameter(self, name=None, value=None, verbose=False):
          """Sends a job parameter (allowed in PENDING phase only).
  
          Parameters
          ----------
          name : string
              Parameter name.
          value : string
              Parameter value.
          """
          if self._phase == 'PENDING':
              # send post parameter/value
              context = f"async/{self.jobid}"
              args = {
                  name: str(value)}
              data = self.connHandler.url_encode(args)
              response = self.connHandler.execute_tappost(subcontext=context,
                                                          data=data,
                                                          verbose=verbose)
              if verbose:
                  print(response.status, response.reason)
                  print(response.getheaders())
              self.__last_phase_response_status = response.status
              if response.status != 200:
                  errMsg = taputils.get_http_response_error(response)
                  print(response.status, errMsg)
                  raise requests.exceptions.HTTPError(errMsg)
              return response
          else:
              raise ValueError(f"Cannot start a job in phase: {self._phase}")
  
      def get_phase(self, update=False):
          """Returns the job phase. May optionally update the job's phase.
  
          Parameters
          ----------
          update : bool
              if True, the phase will by updated by querying the server before
              returning.
  
          Returns
          -------
          The job phase
          """
          if update:
              phase_request = f"async/{self.jobid}/phase"
              response = self.connHandler.execute_tapget(phase_request)
  
              self.__last_phase_response_status = response.status
              if response.status != 200:
                  errMsg = taputils.get_http_response_error(response)
                  print(response.status, errMsg)
                  raise requests.exceptions.HTTPError(errMsg)
  
              self._phase = str(response.read().decode('utf-8'))
          return self._phase
  
      def set_response_status(self, status, msg):
          """Sets the HTTP(s) connection status
  
          Parameters
          ----------
          status : int, mandatory
              HTTP(s) response status
          msg : str, mandatory
              HTTP(s) response message
          """
          self.__responseStatus = status
          self.__responseMsg = msg
  
      def get_data(self):
          """Returns the job results (Astroquery API specification)
          This method will block if the job is asynchronous and the job has not
          finished yet.
  
          Returns
          -------
          The job results (astropy.table).
          """
          return self.get_results()
  
      def get_results(self):
          """Returns the job results
          This method will block if the job is asynchronous and the job has not
          finished yet.
  
          Returns
          -------
          The job results (astropy.table).
          """
          if self.results is not None:
              return self.results
          # try load results from file
          # read_results_table_from_file checks whether
          # the file already exists or not
          outputFormat = self.parameters['format']
          results = modelutils.read_results_table_from_file(self.outputFile,
                                                            outputFormat)
          if results is not None:
              self.results = results
              return results
          # Try to load from server: only async
          if not self.async_:
              # sync: result is in a file
              return None
          else:
              # async: result is in the server once the job is finished
              self.__load_async_job_results()
              return self.results
  
      def set_results(self, results):
          """Sets the job results
  
          Parameters
          ----------
          results : Table object, mandatory
              job results
          """
          self.results = results
          self.__resultInMemory = True
  
      def save_results(self, verbose=False):
          """Saves job results
          If the job is asynchronous, this method will block until the results
          are available.
  
          Parameters
          ----------
          verbose : bool, optional, default 'False'
              flag to display information about the process
          """
          if self.__resultInMemory:
              if verbose:
                  print(f"Saving results to: {self.outputFile}")
              self.results.to_xml(self.outputFile)
          else:
              if not self.async_:
                  # sync: cannot access server again
                  log.info("No results to save")
              else:
                  # Async
                  self.wait_for_job_end(verbose)
                  response = self.connHandler.execute_tapget(
                      f"async/{self.jobid}/results/result")
                  if verbose:
                      print(response.status, response.reason)
                      print(response.getheaders())
                  isError = self.connHandler.\
                      check_launch_response_status(response,
                                                   verbose,
                                                   200)
                  if isError:
                      print(response.reason)
                      raise Exception(response.reason)
                  if self.outputFileUser is None:
                      # User did not provide an output
                      # The output is a temporary
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/model/job.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  """
  import os
  from unittest.mock import patch
  from urllib.parse import quote_plus, urlencode
  
  import numpy as np
  import pytest
  from requests import HTTPError
  
  from astroquery.utils.tap.model.tapcolumn import TapColumn
  
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  from astroquery.utils.tap.core import TapPlus
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap import taputils
  
  
  def data_path(filename):
      data_dir = os.path.join(os.path.dirname(__file__), 'data')
      return os.path.join(data_dir, filename)
  
  
  def test_load_tables():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(500)
      tableDataFile = data_path('test_tables.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
      with pytest.raises(Exception):
          tap.load_tables()
  
      responseLoadTable.set_status_code(200)
      res = tap.load_tables()
      assert len(res) == 2
  
      # Table 1
      table = __find_table('public', 'table1', res)
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
      # Table 2
      table = __find_table('public', 'table2', res)
      assert table.description == 'Table2 desc'
      columns = table.columns
      assert len(columns) == 3
      col = __find_column('table2_col1', columns)
      __check_column(col, 'Table2 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table2_col2', columns)
      __check_column(col, 'Table2 Column2 desc', '', 'INTEGER', None)
      col = __find_column('table2_col3', columns)
      __check_column(col, 'Table2 Column3 desc', '', 'INTEGER', None)
  
  
  def test_load_tables_parameters():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(200)
      tableDataFile = data_path('test_tables.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      # empty request
      tap.load_tables()
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=false & share_accessible=false: equals to
      # empty request
      tap.load_tables(only_names=False, include_shared_tables=False)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names
      tableRequest = "tables?only_tables=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag share_accessible=true
      tableRequest = "tables?share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=true & share_accessible=true
      tableRequest = "tables?only_tables=true&share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True, include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
  
  def test_load_table():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
  
      # No arguments
      with pytest.raises(Exception):
          tap.load_table()
  
      responseLoadTable = DummyResponse(500)
      tableDataFile = data_path('test_table1.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableSchema = "public"
      tableName = "table1"
      fullQualifiedTableName = f"{tableSchema}.{tableName}"
      tableRequest = f"tables?tables={fullQualifiedTableName}"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      with pytest.raises(Exception):
          tap.load_table(fullQualifiedTableName)
  
      responseLoadTable.set_status_code(200)
      table = tap.load_table(fullQualifiedTableName)
      assert table is not None
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
  
  def test_launch_sync_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseLaunchJob.set_data(method='POST', body=jobData)
      query = 'select top 5 * from table'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": quote_plus(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      jobRequest = f"sync?{sortedKey}"
      connHandler.set_response(jobRequest, responseLaunchJob)
  
      with pytest.raises(Exception):
          tap.launch_job(query, maxrec=10)
  
      responseLaunchJob.set_status_code(200)
  
      job = tap.launch_job(query)
  
      assert job is not None
      assert job.async_ is False
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_launch_sync_job_redirect():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      jobid = '12345'
      resultsReq = f'sync/{jobid}'
      resultsLocation = f'http://test:1111/tap/{resultsReq}'
      launchResponseHeaders = [
          ['location', resultsLocation]
      ]
      responseLaunchJob.set_data(method='POST')
      query = 'select top 5 * from table'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": quote_plus(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      jobRequest = f"sync?{sortedKey}"
      connHandler.set_response(jobRequest, responseLaunchJob)
     
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/tests/test_tap.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  """
  import os
  from unittest.mock import patch
  
  import numpy as np
  import pytest
  from requests import HTTPError
  
  from astroquery.utils.tap.model.tapcolumn import TapColumn
  
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  from astroquery.utils.tap.core import TapPlus
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap import taputils
  
  
  def data_path(filename):
      data_dir = os.path.join(os.path.dirname(__file__), 'data')
      return os.path.join(data_dir, filename)
  
  
  def test_load_tables():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(500)
      tableDataFile = data_path('test_tables.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
      with pytest.raises(Exception):
          tap.load_tables()
  
      responseLoadTable.set_status_code(200)
      res = tap.load_tables()
      assert len(res) == 2
  
      # Table 1
      table = __find_table('public', 'table1', res)
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
      # Table 2
      table = __find_table('public', 'table2', res)
      assert table.description == 'Table2 desc'
      columns = table.columns
      assert len(columns) == 3
      col = __find_column('table2_col1', columns)
      __check_column(col, 'Table2 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table2_col2', columns)
      __check_column(col, 'Table2 Column2 desc', '', 'INTEGER', None)
      col = __find_column('table2_col3', columns)
      __check_column(col, 'Table2 Column3 desc', '', 'INTEGER', None)
  
  
  def test_load_tables_parameters():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(200)
      tableDataFile = data_path('test_tables.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      # empty request
      tap.load_tables()
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=false & share_accessible=false: equals to
      # empty request
      tap.load_tables(only_names=False, include_shared_tables=False)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names
      tableRequest = "tables?only_tables=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag share_accessible=true
      tableRequest = "tables?share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=true & share_accessible=true
      tableRequest = "tables?only_tables=true&share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True, include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
  
  def test_load_table():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
  
      # No arguments
      with pytest.raises(Exception):
          tap.load_table()
  
      responseLoadTable = DummyResponse(500)
      tableDataFile = data_path('test_table1.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableSchema = "public"
      tableName = "table1"
      fullQualifiedTableName = f"{tableSchema}.{tableName}"
      tableRequest = f"tables?tables={fullQualifiedTableName}"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      with pytest.raises(Exception):
          tap.load_table(fullQualifiedTableName)
  
      responseLoadTable.set_status_code(200)
      table = tap.load_table(fullQualifiedTableName)
      assert table is not None
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
  
  def test_launch_sync_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseLaunchJob.set_data(method='POST', body=jobData)
      query = 'select top 5 * from table'
      dTmp = {"q": query}
      dTmpEncoded = connHandler.url_encode(dTmp)
      p = dTmpEncoded.find("=")
      q = dTmpEncoded[p + 1:]
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": str(q)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      jobRequest = f"sync?{sortedKey}"
      connHandler.set_response(jobRequest, responseLaunchJob)
  
      with pytest.raises(Exception):
          tap.launch_job(query, maxrec=10)
  
      responseLaunchJob.set_status_code(200)
  
      job = tap.launch_job(query)
  
      assert job is not None
      assert job.async_ is False
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_launch_sync_job_redirect():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      jobid = '12345'
      resultsReq = f'sync/{jobid}'
      resultsLocation = f'http://test:1111/tap/{resultsReq}'
      launchResponseHeaders = [
          ['location', resultsLocation]
      ]
      responseLaunchJob.set_data(method='POST')
      query = 'select top 5 * from table'
      dTmp = {"q": query}
      dTmpEncoded = connHandler.url_encode(dTmp)
      p = dTmpEncoded.find("=")
      q = dTmpEncoded[p + 1:]
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/tests/test_tap.py