Template ModuleΒΆ
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
# 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_nosections
# 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 ValueError:
# catch common errors here, but never use bare excepts
# 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_nosections(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_nosections(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.