The template module, written by Madhura Parikh, shows how to build your own module for a new online service.
# Licensed under a 3-clause BSD style license - see LICENSE.rst from __future__ import print_function # put all imports organized as shown below # 1. standard library imports # 2. third party imports import astropy.units as u import astropy.coordinates as coord import astropy.io.votable as votable from astropy.table import Table from astropy.io import fits # 3. local imports - use relative imports # commonly required local imports shown below as example # all Query classes should inherit from BaseQuery. from ..query import BaseQuery # has common functions required by most modules from ..utils import commons # prepend_docstr is a way to copy docstrings between methods from ..utils import prepend_docstr_noreturns # async_to_sync generates the relevant query tools from _async methods from ..utils import async_to_sync # import configurable items declared in __init__.py from . import conf # export all the public classes and methods __all__ = ['Template', 'TemplateClass'] # declare global variables and constants if any # Now begin your main class # should be decorated with the async_to_sync imported previously @async_to_sync class TemplateClass(BaseQuery): """ Not all the methods below are necessary but these cover most of the common cases, new methods may be added if necessary, follow the guidelines at <http://astroquery.readthedocs.io/en/latest/api.html> """ # use the Configuration Items imported from __init__.py to set the URL, # TIMEOUT, etc. URL = conf.server TIMEOUT = conf.timeout # all query methods are implemented with an "async" method that handles # making the actual HTTP request and returns the raw HTTP response, which # should be parsed by a separate _parse_result method. The query_object # method is created by async_to_sync automatically. It would look like # this: """ def query_object(object_name, get_query_payload=False) response = self.query_object_async(object_name, get_query_payload=get_query_payload) if get_query_payload: return response result = self._parse_result(response, verbose=verbose) return result """ def query_object_async(self, object_name, get_query_payload=False, cache=True): """ This method is for services that can parse object names. Otherwise use :meth:`astroquery.template_module.TemplateClass.query_region`. Put a brief description of what the class does here. Parameters ---------- object_name : str name of the identifier to query. get_query_payload : bool, optional This should default to False. When set to `True` the method should return the HTTP request parameters as a dict. verbose : bool, optional This should default to `False`, when set to `True` it displays VOTable warnings. any_other_param : <param_type> similarly list other parameters the method takes Returns ------- response : `requests.Response` The HTTP response returned from the service. All async methods should return the raw HTTP response. Examples -------- While this section is optional you may put in some examples that show how to use the method. The examples are written similar to standard doctests in python. """ # the async method should typically have the following steps: # 1. First construct the dictionary of the HTTP request params. # 2. If get_query_payload is `True` then simply return this dict. # 3. Else make the actual HTTP request and return the corresponding # HTTP response # All HTTP requests are made via the `BaseQuery._request` method. This # use a generic HTTP request method internally, similar to # `requests.Session.request` of the Python Requests library, but # with added caching-related tools. # See below for an example: # first initialize the dictionary of HTTP request parameters request_payload = dict() # Now fill up the dictionary. Here the dictionary key should match # the exact parameter name as expected by the remote server. The # corresponding dict value should also be in the same format as # expected by the server. Additional parsing of the user passed # value may be required to get it in the right units or format. # All this parsing may be done in a separate private `_args_to_payload` # method for cleaner code. request_payload['object_name'] = object_name # similarly fill up the rest of the dict ... if get_query_payload: return request_payload # BaseQuery classes come with a _request method that includes a # built-in caching system response = self._request('GET', self.URL, params=request_payload, timeout=self.TIMEOUT, cache=cache) return response # For services that can query coordinates, use the query_region method. # The pattern is similar to the query_object method. The query_region # method also has a 'radius' keyword for specifying the radius around # the coordinates in which to search. If the region is a box, then # the keywords 'width' and 'height' should be used instead. The coordinates # may be accepted as an `astropy.coordinates` object or as a string, which # may be further parsed. # similarly we write a query_region_async method that makes the # actual HTTP request and returns the HTTP response def query_region_async(self, coordinates, radius, height, width, get_query_payload=False, cache=True): """ Queries a region around the specified coordinates. Parameters ---------- coordinates : str or `astropy.coordinates`. coordinates around which to query radius : str or `astropy.units.Quantity`. the radius of the cone search width : str or `astropy.units.Quantity` the width for a box region height : str or `astropy.units.Quantity` the height for a box region get_query_payload : bool, optional Just return the dict of HTTP request parameters. verbose : bool, optional Display VOTable warnings or not. Returns ------- response : `requests.Response` The HTTP response returned from the service. All async methods should return the raw HTTP response. """ request_payload = self._args_to_payload(coordinates, radius, height, width) if get_query_payload: return request_payload response = self._request('GET', self.URL, params=request_payload, timeout=self.TIMEOUT, cache=cache) return response # as we mentioned earlier use various python regular expressions, etc # to create the dict of HTTP request parameters by parsing the user # entered values. For cleaner code keep this as a separate private method: def _args_to_payload(self, *args, **kwargs): request_payload = dict() # code to parse input and construct the dict # goes here. Then return the dict to the caller return request_payload # the methods above call the private _parse_result method. # This should parse the raw HTTP response and return it as # an `astropy.table.Table`. Below is the skeleton: def _parse_result(self, response, verbose=False): # if verbose is False then suppress any VOTable related warnings if not verbose: commons.suppress_vo_warnings() # try to parse the result into an astropy.Table, else # return the raw result with an informative error message. try: # do something with regex to get the result into # astropy.Table form. return the Table. pass except: # catch common errors here # return raw result/ handle in some way pass return Table() # Image queries do not use the async_to_sync approach: the "synchronous" # version must be defined explicitly. The example below therefore presents # a complete example of how to write your own synchronous query tools if # you prefer to avoid the automatic approach. # # For image queries, the results should be returned as a # list of `astropy.fits.HDUList` objects. Typically image queries # have the following method family: # 1. get_images - this is the high level method that interacts with # the user. It reads in the user input and returns the final # list of fits images to the user. # 2. get_images_async - This is a lazier form of the get_images function, # in that it returns just the list of handles to the image files # instead of actually downloading them. # 3. extract_image_urls - This takes in the raw HTTP response and scrapes # it to get the downloadable list of image URLs. # 4. get_image_list - this is similar to the get_images, but it simply # takes in the list of URLs scrapped by extract_image_urls and # returns this list rather than the actual FITS images # NOTE : in future support may be added to allow the user to save # the downloaded images to a preferred location. Here we look at the # skeleton code for image services def get_images(self, coordinates, radius, get_query_payload): """ A query function that searches for image cut-outs around coordinates Parameters ---------- coordinates : str or `astropy.coordinates`. coordinates around which to query radius : str or `astropy.units.Quantity`. the radius of the cone search get_query_payload : bool, optional If true than returns the dictionary of query parameters, posted to remote server. Defaults to `False`. Returns ------- A list of `astropy.fits.HDUList` objects """ readable_objs = self.get_images_async(coordinates, radius, get_query_payload=get_query_payload) if get_query_payload: return readable_objs # simply return the dict of HTTP request params # otherwise return the images as a list of astropy.fits.HDUList return [obj.get_fits() for obj in readable_objs] @prepend_docstr_noreturns(get_images.__doc__) def get_images_async(self, coordinates, radius, get_query_payload=False): """ Returns ------- A list of context-managers that yield readable file-like objects """ # As described earlier, this function should return just # the handles to the remote image files. Use the utilities # in commons.py for doing this: # first get the links to the remote image files image_urls = self.get_image_list(coordinates, radius, get_query_payload=get_query_payload) if get_query_payload: # if true then return the HTTP request params dict return image_urls # otherwise return just the handles to the image files. return [commons.FileContainer(U) for U in image_urls] # the get_image_list method, simply returns the download # links for the images as a list @prepend_docstr_noreturns(get_images.__doc__) def get_image_list(self, coordinates, radius, get_query_payload=False, cache=True): """ Returns ------- list of image urls """ # This method should implement steps as outlined below: # 1. Construct the actual dict of HTTP request params. # 2. Check if the get_query_payload is True, in which # case it should just return this dict. # 3. Otherwise make the HTTP request and receive the # HTTP response. # 4. Pass this response to the extract_image_urls # which scrapes it to extract the image download links. # 5. Return the download links as a list. request_payload = self._args_to_payload(coordinates, radius) if get_query_payload: return request_payload response = self._request(method="GET", url=self.URL, data=request_payload, timeout=self.TIMEOUT, cache=cache) return self.extract_image_urls(response.text) # the extract_image_urls method takes in the HTML page as a string # and uses regexps, etc to scrape the image urls: def extract_image_urls(self, html_str): """ Helper function that uses regex to extract the image urls from the given HTML. Parameters ---------- html_str : str source from which the urls are to be extracted Returns ------- list of image URLs """ # do something with regex on the HTML # return the list of image URLs pass # the default tool for users to interact with is an instance of the Class Template = TemplateClass() # once your class is done, tests should be written # See ./tests for examples on this # Next you should write the docs in astroquery/docs/module_name # using Sphinx.