Usage

Basic Usage

To use API Wrapper in a project:

from apiwrapper import APIWrapper

Use it as a helper:

my_api = APIWrapper()
url = 'https://api.github.com/users/ardydedase/repos'
resp = my_api.make_request(url, method='get', headers=None, data=None, callback=None).parsed
print(resp)

Use it as a parent class:

class GithubAPI(APIWrapper):
    def get_repos(self, username):
        """
        Uses 'make_request' method
        """
        url = "https://api.github.com/users/{username}/repos".format(username=username)
        return self.make_request(url, method='get', headers=None, data=None, callback=None).parsed

Parameter reference for make_request():

def make_request(self, url, method='get', headers=None, data=None,
                 callback=None, errors=STRICT, verify=False, **params):
    """
    Reusable method for performing requests.
    :param url - URL to request
    :param method - request method, default is 'get'
    :param headers - request headers
    :param data - post data
    :param callback - callback to be applied to response,
                      default callback will parse response as json object.
    :param errors - specifies communication errors handling mode, possible
                    values are:
                     * strict (default) - throw an error as soon as one
                        occurred
                     * graceful - ignore certain errors, e.g. EmptyResponse
                     * ignore - ignore all errors and return a result in
                                any case.
                                NOTE that it DOES NOT mean that no
                                exceptions can be
                                raised from this method, it mostly ignores
                                communication
                                related errors.
                     * None or empty string equals to default
    :param params - additional query parameters for request
    """

Polling

APIWrapper’s built-in polling method makes it convenient to declare polling methods and calls. Its flexibility allows a number of options including switching between JSON and XML response types.

In this poll method example, let’s use Skyscanner’s API.

Let’s start by importing APIWrapper class and all the error modes available in the apiwrapper package:

from apiwrapper import (
    APIWrapper,
    STRICT,
    GRACEFUL,
    IGNORE)

Next will be to declare the Flights class that will inherit our APIWrapper parent class. The parent APIWrapper class is initizialized with response_format=’json’. The api_key is a private property so we don’t have to pass it as an argument every time we call make_request:

class Flights(APIWrapper):
    """
    Skyscanner Flights Live Pricing
    http://business.skyscanner.net/
    portal/en-GB/Documentation/FlightsLivePricingList
    """
    API_HOST = 'http://partners.api.skyscanner.net'
    PRICING_SESSION_URL = '{api_host}/apiservices/pricing/v1.0'.format(
        api_host=API_HOST)

    def __init__(self, api_key):
        self.api_key = api_key
        super(Flights, self).__init__(response_format='json')

Wrap the make_request method from APIWrapper and inject the apikey only if it is not available in the request url:

def make_request(self, url, method='get', headers=None,
                 data=None, callback=None, errors=STRICT,
                 verify=False, **params):
    """
    Call the `make_request` method from apiwrapper.
    So we can inject the apikey when it is not available.
    """
    if 'apikey' not in url.lower():
        params.update({
            'apiKey': self.api_key
        })
    return super(Flights, self).make_request(url, method, headers,
                                             data, callback, errors,
                                             verify, **params)

The create_session method prepares the API’s polling session and returns the polling url poll_url. This method uses the make_request method declared above. It also makes use of the _headers() method from APIWrapper:

def create_session(self, **params):
    """
    Create the session
    date format: YYYY-mm-dd
    location: ISO code.
    After creating the session,
    this method will return the poll_url.
    """
    service_url = self.PRICING_SESSION_URL
    return self.make_request(service_url,
                             method='post',
                             headers=self._headers(),
                             callback=lambda resp: resp.headers[
                                 'location'],
                             data=params)

This boolean method is_poll_complete_callback will be passed as a callback parameter in the APIWrapper.poll method call. is_poll_complete_callback will receive the poll response from poll method as a parameter. This method will then use the poll_resp value to check whether the polling is complete or not and returns a boolean:

def _is_poll_complete_callback(self, poll_resp):
    """
    Checks the condition in poll response to determine if it is complete
    and no subsequent poll requests should be done.
    """
    if poll_resp.parsed is None:
        return False
    success_list = ['UpdatesComplete', True, 'COMPLETE']
    status = None
    if self.response_format == 'xml':
        status = poll_resp.parsed.find('./Status').text
    elif self.response_format == 'json':
        status = poll_resp.parsed.get(
            'Status', poll_resp.parsed.get('status'))
    if status is None:
        raise RuntimeError('Unable to get poll response status.')
    return status in success_list

And lastly, the get_result method polls the API using the URL that was returned from create_session. Notice that we are passing _is_poll_complete_callback as an argument to the is_poll_complete_callback parameter in the poll method. After the poll is complete, the get_result method will return the flight search result:

def get_result(self, errors=STRICT, **params):
    """
    Get all results, no filtering,
    etc. by creating and polling the session.
    """
    service_url = self.create_session(**params)
    return self.poll(service_url, errors=errors, is_poll_complete_callback=self._is_poll_complete_callback)

Now that the Flights class is ready. The get_result method can be called as follows:

from datetime import datetime, timedelta

datetime_format = '%Y-%m-%d'
outbound_datetime = datetime.now() + timedelta(days=7)
inbound_datetime = outbound_datetime + timedelta(days=3)
outbound_date = outbound_datetime.strftime(datetime_format)
inbound_date = inbound_datetime.strftime(datetime_format)

flights_service = Flights(<skyscanner_api_key>)
result = flights_service.get_result(
    errors=GRACEFUL,
    country='UK',
    currency='GBP',
    locale='en-GB',
    originplace='SIN-sky',
    destinationplace='KUL-sky',
    outbounddate=outbound_date,
    inbounddate=inbound_date,
    adults=1).parsed

Parameter reference for poll():

def poll(self, url, initial_delay=2, delay=1, tries=20, errors=STRICT, is_complete_callback=None, **params):
    """
    Poll the URL
    :param url - URL to poll, should be returned by 'create_session' call
    :param initial_delay - specifies how many seconds to wait before the first poll
    :param delay - specifies how many seconds to wait between the polls
    :param tries - number of polls to perform
    :param errors - errors handling mode, see corresponding parameter in 'make_request' method
    :param params - additional query params for each poll request
    """

Response callbacks

callback parameter in make_request method. It passes the Response object as an argument:

class GithubAPI(APIWrapper):
    def _my_callback(self, resp):
        """
        'resp' is a Response object returned from `requests` library
        """
        return resp.json()


    def get_repos(self, username):
        """
        Uses 'make_request' method
        """
        url = "https://api.github.com/users/{username}/repos".format(username=username)
        return self.make_request(url, method='get', headers=None, data=None, callback=self._my_callback)