diff --git a/dev-requirements.txt b/dev-requirements.txt index aa29d76..846de8c 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -7,3 +7,4 @@ sphinx sphinx_rtd_theme unittest2 enum34 +aiohttp>=3.7.4 \ No newline at end of file diff --git a/pyvat/aio/__init__.py b/pyvat/aio/__init__.py new file mode 100644 index 0000000..13e9c65 --- /dev/null +++ b/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__, +) diff --git a/pyvat/aio/registries.py b/pyvat/aio/registries.py new file mode 100644 index 0000000..61e4cc3 --- /dev/null +++ b/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', ) diff --git a/pyvat/registries.py b/pyvat/registries.py index b5a1f24..0555a43 100644 --- a/pyvat/registries.py +++ b/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'%s%s' % - (country_code, vat_number) + def generate_request_data(self, vat_number, country_code): + return ( + u'%s%s' % + (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): # # # - 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', ) diff --git a/setup.py b/setup.py index 46e0774..2998eff 100644 --- a/setup.py +++ b/setup.py @@ -15,6 +15,7 @@ 'requests>=1.0.0,<3.0', 'pycountry', 'enum34; python_version < "3.4"', + 'aiohttp>=3.7.4', ] tests_require = [