Add support for async

dev-requirements.txt

@@ -7,3 +7,4 @@ sphinx
 sphinx_rtd_theme
 unittest2
 enum34
+aiohttp>=3.7.4

pyvat/aio/__init__.py

@@ -0,0 +1,102 @@
+from ..item_type import ItemType
+from ..party import Party
+from .registries import ViesRegistry
+from ..result import VatNumberCheckResult
+from ..vat_charge import VatCharge, VatChargeAction
+
+from .. import __version__
+
+VIES_REGISTRY = ViesRegistry()
+"""VIES registry instance.
+"""
+
+
+VAT_REGISTRIES = {
+    'AT': VIES_REGISTRY,
+    'BE': VIES_REGISTRY,
+    'BG': VIES_REGISTRY,
+    'CY': VIES_REGISTRY,
+    'CZ': VIES_REGISTRY,
+    'DE': VIES_REGISTRY,
+    'DK': VIES_REGISTRY,
+    'EE': VIES_REGISTRY,
+    'ES': VIES_REGISTRY,
+    'FI': VIES_REGISTRY,
+    'FR': VIES_REGISTRY,
+    'GR': VIES_REGISTRY,
+    'HU': VIES_REGISTRY,
+    'HR': VIES_REGISTRY,
+    'IE': VIES_REGISTRY,
+    'IT': VIES_REGISTRY,
+    'LT': VIES_REGISTRY,
+    'LU': VIES_REGISTRY,
+    'LV': VIES_REGISTRY,
+    'MT': VIES_REGISTRY,
+    'NL': VIES_REGISTRY,
+    'PL': VIES_REGISTRY,
+    'PT': VIES_REGISTRY,
+    'RO': VIES_REGISTRY,
+    'SE': VIES_REGISTRY,
+    'SK': VIES_REGISTRY,
+    'SI': VIES_REGISTRY,
+}
+"""VAT registries.
+
+Mapping from ISO 3166-1-alpha-2 country codes to the VAT registry capable of
+validating the VAT number.
+"""
+
+from .. import decompose_vat_number
+from .. import is_vat_number_format_valid
+
+
+def check_vat_number(vat_number, country_code=None):
+    """Check if a VAT number is valid.
+
+    If possible, the VAT number will be checked against available registries.
+
+    :param vat_number: VAT number to validate.
+    :param country_code:
+        Optional country code. Should be supplied if known, as there is no
+        guarantee that naively entered VAT numbers contain the correct alpha-2
+        country code prefix for EU countries just as not all non-EU countries
+        have a reliable country code prefix. Default ``None`` prompting
+        detection.
+    :returns:
+        a :class:`VatNumberCheckResult` instance containing the result for
+        the full VAT number check.
+    """
+
+    # Decompose the VAT number.
+    vat_number, country_code = decompose_vat_number(vat_number, country_code)
+    if not vat_number or not country_code:
+        return VatNumberCheckResult(False, [
+            '> Unable to decompose VAT number, resulted in %r and %r' %
+            (vat_number, country_code)
+        ])
+
+    # Test the VAT number format.
+    format_result = is_vat_number_format_valid(vat_number, country_code)
+    if format_result is not True:
+        return VatNumberCheckResult(format_result, [
+            '> VAT number validation failed: %r' % (format_result)
+        ])
+
+    # Attempt to check the VAT number against a registry.
+    if country_code not in VAT_REGISTRIES:
+        return VatNumberCheckResult()
+
+    return VAT_REGISTRIES[country_code].check_vat_number(vat_number,
+                                                         country_code)
+
+from .. import get_sale_vat_charge
+
+__all__ = (
+    'check_vat_number',
+    'get_sale_vat_charge',
+    'is_vat_number_format_valid',
+    ItemType.__name__,
+    Party.__name__,
+    VatCharge.__name__,
+    VatChargeAction.__name__,
+)

pyvat/aio/registries.py

@@ -0,0 +1,61 @@
+# import requests
+import aiohttp
+from asyncio import TimeoutError
+
+import xml.dom.minidom
+
+from ..result import VatNumberCheckResult
+from ..xml_utils import get_first_child_element, get_text, NodeNotFoundError
+from ..exceptions import ServerError
+from ..registries import BaseViesRegistry
+
+class ViesRegistry(BaseViesRegistry):
+    """VIES registry."""
+
+
+    async def check_vat_number(self, vat_number, country_code):
+        # Non-ISO code used for Greece.
+        if country_code == 'GR':
+            country_code = 'EL'
+
+        # Request information about the VAT number.
+        result = VatNumberCheckResult()
+
+        request_data = self.generate_request_data(vat_number, country_code)
+
+        result.log_lines += [
+            u'> POST %s with payload of content type text/xml, charset UTF-8:',
+            request_data,
+        ]
+
+        try:
+            async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(self.DEFAULT_TIMEOUT)) as session:
+                async with session.post(
+                    self.CHECK_VAT_SERVICE_URL, 
+                    data=request_data.encode('utf-8'), 
+                    headers={
+                        'Content-Type': 'text/xml; charset=utf-8',
+                }) as request:
+                    response_text = await request.text()
+                    result.log_lines += [
+                        u'< Response with status %d of content type %s:' %
+                        (request.status, request.headers['Content-Type']),
+                        response_text,
+                    ]
+
+                    # Do not completely fail problematic requests.
+                    if request.status != 200 or \
+                        not request.headers['Content-Type'].startswith('text/xml'):
+                            result.log_lines.append(u'< Response is nondeterministic due to '
+                                                    u'invalid response status code or MIME '
+                                                    u'type')
+                            return result
+        except TimeoutError as e:
+            result.log_lines.append(u'< Request to EU VIEW registry timed out:'
+                                    u' {}'.format(e))
+            return result
+
+        return self.parse_response_to_result(response_text, result)
+
+
+__all__ = ('Registry', 'ViesRegistry', )

pyvat/registries.py

@@ -24,13 +24,7 @@ def check_vat_number(self, vat_number, country_code):
 
         raise NotImplementedError()
 
-
-class ViesRegistry(Registry):
-    """VIES registry.
-
-    Uses the European Commision's VIES registry for validating VAT numbers.
-    """
-
+class BaseViesRegistry(Registry):
     CHECK_VAT_SERVICE_URL = 'http://ec.europa.eu/taxation_customs/vies/' \
         'services/checkVatService'
     """URL for the VAT checking service.
@@ -39,65 +33,20 @@ class ViesRegistry(Registry):
     DEFAULT_TIMEOUT = 8
     """Timeout for the requests."""
 
-    def check_vat_number(self, vat_number, country_code):
-        # Non-ISO code used for Greece.
-        if country_code == 'GR':
-            country_code = 'EL'
-
-        # Request information about the VAT number.
-        result = VatNumberCheckResult()
-
-        request_data = (
-                u'<?xml version="1.0" encoding="UTF-8"?><SOAP-ENV:Envelope'
-                u' xmlns:ns0="urn:ec.europa.eu:taxud:vies:services:checkVa'
-                u't:types" xmlns:ns1="http://schemas.xmlsoap.org/soap/enve'
-                u'lope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-insta'
-                u'nce" xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/env'
-                u'elope/"><SOAP-ENV:Header/><ns1:Body><ns0:checkVat><ns0:c'
-                u'ountryCode>%s</ns0:countryCode><ns0:vatNumber>%s</ns0:va'
-                u'tNumber></ns0:checkVat></ns1:Body></SOAP-ENV:Envelope>' %
-                (country_code, vat_number)
+    def generate_request_data(self, vat_number, country_code):
+        return (
+            u'<?xml version="1.0" encoding="UTF-8"?><SOAP-ENV:Envelope'
+            u' xmlns:ns0="urn:ec.europa.eu:taxud:vies:services:checkVa'
+            u't:types" xmlns:ns1="http://schemas.xmlsoap.org/soap/enve'
+            u'lope/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-insta'
+            u'nce" xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/env'
+            u'elope/"><SOAP-ENV:Header/><ns1:Body><ns0:checkVat><ns0:c'
+            u'ountryCode>%s</ns0:countryCode><ns0:vatNumber>%s</ns0:va'
+            u'tNumber></ns0:checkVat></ns1:Body></SOAP-ENV:Envelope>' %
+            (country_code, vat_number)
         )
 
-        result.log_lines += [
-            u'> POST %s with payload of content type text/xml, charset UTF-8:',
-            request_data,
-        ]
-
-        try:
-            response = requests.post(
-                self.CHECK_VAT_SERVICE_URL,
-                data=request_data.encode('utf-8'),
-                headers={
-                    'Content-Type': 'text/xml; charset=utf-8',
-                },
-                timeout=self.DEFAULT_TIMEOUT
-            )
-        except Timeout as e:
-            result.log_lines.append(u'< Request to EU VIEW registry timed out:'
-                                    u' {}'.format(e))
-            return result
-        except Exception as exception:
-            # Do not completely fail problematic requests.
-            result.log_lines.append(u'< Request failed with exception: %r' %
-                                    (exception))
-            return result
-
-        # Log response information.
-        result.log_lines += [
-            u'< Response with status %d of content type %s:' %
-            (response.status_code, response.headers['Content-Type']),
-            response.text,
-            ]
-
-        # Do not completely fail problematic requests.
-        if response.status_code != 200 or \
-                not response.headers['Content-Type'].startswith('text/xml'):
-            result.log_lines.append(u'< Response is nondeterministic due to '
-                                    u'invalid response status code or MIME '
-                                    u'type')
-            return result
-
+    def parse_response_to_result(self, response_text, result):
         # Parse the DOM and validate as much as we can.
         #
         # We basically expect the result structure to be as follows,
@@ -115,7 +64,7 @@ def check_vat_number(self, vat_number, country_code):
         #     </checkVatResponse>
         #   </soap:Body>
         # </soap:Envelope>
-        result_dom = xml.dom.minidom.parseString(response.text.encode('utf-8'))
+        result_dom = xml.dom.minidom.parseString(response_text.encode('utf-8'))
 
         envelope_node = result_dom.documentElement
         if envelope_node.tagName != 'soap:Envelope':
@@ -179,5 +128,61 @@ def check_vat_number(self, vat_number, country_code):
 
         return result
 
+class ViesRegistry(BaseViesRegistry):
+    """VIES registry.
+
+    Uses the European Commision's VIES registry for validating VAT numbers.
+    """
+
+    def check_vat_number(self, vat_number, country_code):
+        # Non-ISO code used for Greece.
+        if country_code == 'GR':
+            country_code = 'EL'
+
+        # Request information about the VAT number.
+        result = VatNumberCheckResult()
+
+        request_data = self.generate_request_data(vat_number, country_code)
+
+        result.log_lines += [
+            u'> POST %s with payload of content type text/xml, charset UTF-8:',
+            request_data,
+        ]
+
+        try:
+            response = requests.post(
+                self.CHECK_VAT_SERVICE_URL,
+                data=request_data.encode('utf-8'),
+                headers={
+                    'Content-Type': 'text/xml; charset=utf-8',
+                },
+                timeout=self.DEFAULT_TIMEOUT
+            )
+        except Timeout as e:
+            result.log_lines.append(u'< Request to EU VIEW registry timed out:'
+                                    u' {}'.format(e))
+            return result
+        except Exception as exception:
+            # Do not completely fail problematic requests.
+            result.log_lines.append(u'< Request failed with exception: %r' %
+                                    (exception))
+            return result
+
+        # Log response information.
+        result.log_lines += [
+            u'< Response with status %d of content type %s:' %
+            (response.status_code, response.headers['Content-Type']),
+            response.text,
+            ]
+
+        # Do not completely fail problematic requests.
+        if response.status_code != 200 or \
+                not response.headers['Content-Type'].startswith('text/xml'):
+            result.log_lines.append(u'< Response is nondeterministic due to '
+                                    u'invalid response status code or MIME '
+                                    u'type')
+            return result
+
+        return self.parse_response_to_result(response.text, result)
 
 __all__ = ('Registry', 'ViesRegistry', )

setup.py

@@ -15,6 +15,7 @@
     'requests>=1.0.0,<3.0',
     'pycountry',
     'enum34; python_version < "3.4"',
+    'aiohttp>=3.7.4',
 ]
 
 tests_require = [