VATify.eu API

Python example

The python module below implements a very simple VATify.eu REST API client. It de­pends on the popular external library re­quests to provide the low-level HTTPS client func­ti­o­na­lity. Note that this is example code without input sanitization or proper error handling.

Download the source file here. It can be executed directly, by pro­vi­ding the relevant parameters (client ID, access key, coun­try code and com­pany ID) on the command line. For example:

See this guide for more in­for­ma­ti­on about how to use the VATify.eu REST APIs.

#!/usr/bin/env python3
#
#  VATify.eu REST API: python client module (example code)
#  Copyright (C) 2022 KoMnA d.o.o. <code@komna.com>
#
#  Permission to use, copy, modify, and distribute this software for any
#  purpose with or without fee is hereby granted, provided that the above
#  copyright notice and this permission notice appear in all copies.
#
#  THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
#  WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
#  MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
#  ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
#  WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
#  ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
#  OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#
###############################################################################

import sys
import time

import requests



def main():
    """
    Function to run if this script is being executed directly.

    This function reads client credentials, country and company ID from
    command line arguments, authenticates to the VATify.eu API server,
    preforms a real-time query, and prints the resulting company data to
    standard output.
    """

    # Parse command line arguments.
    try:
        client_id, access_key, country, company_id = sys.argv[1:]
    except ValueError:
        sys.stderr.write(("Usage: %s <client-id> <access-key> <country> "
          "<company-id>\n") % (sys.argv[0], ))
        sys.exit(1)

    # Authenticate by exchanging the long-term login credentials for a
    # time-limited bearer token.
    bearer_token = get_bearer_token(client_id, access_key)

    # Start a VATify.eu real-time query.
    poll_url = run_query(bearer_token, country, company_id)

    # Wait until the results are ready.
    delay = 0.5
    while True:
        results = get_query_results(bearer_token, poll_url)
        if results is not None:
            break
        time.sleep(delay)
        if delay < 10:
            delay += 0.5

    # Print results.
    for i, result in enumerate(results):
        sys.stdout.write("\nResult #%d:\n" % (i + 1, ))
        for key, value in result["data"].items():
            sys.stdout.write("  %s: %s\n" % (key, repr(value)))



def get_bearer_token(client_id, access_key):
    """
    Authenticates to the VATify.eu REST API by exchanging a (long-term)
    client ID and access key for a (time-limited) bearer token.  The
    return value is a string representing the newly issued bearer token.
    """

    resp = requests.post("https://api.vatify.eu/v1/oauth2/token",
                         auth=(client_id, access_key),
                         json={ "grant_type": "client_credentials" })

    if resp.status_code != 200:
        raise RuntimeError(_make_exception_text(resp))

    bearer_token = resp.json()["access_token"]
    return bearer_token



def run_query(bearer_token, country, company_id):
    """
    Run a VATify.eu real-time query for company information.  The return
    value is a URL address of the API endpoint to poll for query results.
    """

    resp = requests.post("https://api.vatify.eu/v1/query",
                         headers={ "Authorization": "Bearer " + bearer_token },
                         json={ "country": country, "identifier": company_id })

    if resp.status_code != 202:
        raise RuntimeError(_make_exception_text(resp))

    poll_url = resp.headers["Location"]
    return poll_url


def get_query_results(bearer_token, poll_url):
    """
    Check the status of a VATify.eu real-time query.  Return the query results
    (company data), if already available.
    """

    resp = requests.get(poll_url,
                        headers={ "Authorization": "Bearer " + bearer_token })

    if resp.status_code == 202:
        return None  # query still in progress, results not yet available

    if resp.status_code != 200:
        raise RuntimeError(_make_exception_text(resp))

    query_results = resp.json()["result"]["items"]
    return query_results



def _make_exception_text(http_response):
    """
    Get a resonable exception description in case of an invalid/unexpected
    API call response status code.  Argument ``http_response`` should be
    the corresponding response instance.
    """

    if http_response.status_code < 400:
        return ("VATify.eu REST API has responded with an unexpected status "
                "code: %d") % (http_response.status_code, )

    error_list = []
    for err in http_response.json()["errors"]:
        error_list.append("[%s] %s" % (err["title"], err["detail"]))

    return "\n".join(error_list)



if __name__ == "__main__":
    main()