From a3179b3f8d89001ec716388515965442af288b9c Mon Sep 17 00:00:00 2001 From: Rick Voormolen Date: Tue, 23 Mar 2021 14:45:37 +0100 Subject: [PATCH 1/5] Split generic functions to BaseViesRegistry --- pyvat/registries.py | 135 +++++++++++++++++++++++--------------------- 1 file changed, 70 insertions(+), 65 deletions(-) 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', ) From 44fd0da7c8d7e91dc8cb9b7942e87b8e7a139436 Mon Sep 17 00:00:00 2001 From: Rick Voormolen Date: Tue, 23 Mar 2021 14:46:20 +0100 Subject: [PATCH 2/5] Create aiohttp ViesRegistry --- pyvat/aio/registries.py | 61 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) create mode 100644 pyvat/aio/registries.py 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', ) From d4b153ef8268376a932729fcfbc5ad997daab60b Mon Sep 17 00:00:00 2001 From: Rick Voormolen Date: Tue, 23 Mar 2021 14:46:37 +0100 Subject: [PATCH 3/5] Clone __init__, adjust to use Async vies registry --- pyvat/aio/__init__.py | 256 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 256 insertions(+) create mode 100644 pyvat/aio/__init__.py diff --git a/pyvat/aio/__init__.py b/pyvat/aio/__init__.py new file mode 100644 index 0000000..ee43c53 --- /dev/null +++ b/pyvat/aio/__init__.py @@ -0,0 +1,256 @@ +import re +import pycountry +from ..item_type import ItemType +from ..party import Party +from .registries import ViesRegistry +from ..result import VatNumberCheckResult +from ..vat_charge import VatCharge, VatChargeAction +from ..vat_rules import VAT_RULES + + +__version__ = '1.3.15' + +from .. import WHITESPACE_EXPRESSION +"""Whitespace expression. + +Used for cleaning VAT numbers. +""" + +from .. import VAT_NUMBER_EXPRESSIONS +"""VAT number expressions. + +Mapping form ISO 3166-1-alpha-2 country codes to the whitespace-less expression +a valid VAT number from the given country must match excluding the country code +prefix. + +EU VAT number structures are retrieved from `VIES +`_. +""" + + +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. +""" + + +def decompose_vat_number(vat_number, country_code=None): + """Decompose a VAT number and an optional country code. + + :param vat_number: VAT number. + :param country_code: + Optional country code. Default ``None`` prompting detection from the + VAT number. + :returns: + a :class:`tuple` containing the VAT number and country code or + ``(vat_number, None)`` if decomposition failed. + """ + + # Clean the VAT number. + vat_number = WHITESPACE_EXPRESSION.sub('', vat_number).upper() + + # Attempt to determine the country code of the VAT number if possible. + if not country_code: + country_code = vat_number[0:2] + + if any(c.isdigit() for c in country_code): + # Country code should not contain digits + return (vat_number, None) + + # Non-ISO code used for Greece. + if country_code == 'EL': + country_code = 'GR' + + if country_code not in VAT_REGISTRIES: + try: + if not pycountry.countries.get(alpha_2=country_code): + return (vat_number, None) + except KeyError: + # country code not found + return (vat_number, None) + vat_number = vat_number[2:] + elif vat_number[0:2] == country_code: + vat_number = vat_number[2:] + elif country_code == 'GR' and vat_number[0:2] == 'EL': + vat_number = vat_number[2:] + + return vat_number, country_code + + +def is_vat_number_format_valid(vat_number, country_code=None): + """Test if the format of a VAT number is valid. + + :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: + ``True`` if the format of the VAT number can be fully asserted as valid + or ``False`` if not. + """ + + vat_number, country_code = decompose_vat_number(vat_number, country_code) + + if not vat_number or not country_code: + return False + elif not any(c.isdigit() for c in vat_number): + return False + elif country_code not in VAT_NUMBER_EXPRESSIONS: + return False + elif not VAT_NUMBER_EXPRESSIONS[country_code].match(vat_number): + return False + + return True + + +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) + + +def get_sale_vat_charge(date, + item_type, + buyer, + seller): + """Get the VAT charge for performing the sale of an item. + + Currently only supports determination of the VAT charge for + telecommunications, broadcasting and electronic services in the EU. + + :param date: Sale date. + :type date: datetime.date + :param item_type: Type of the item being sold. + :type item_type: ItemType + :param buyer: Buyer. + :type buyer: Party + :param seller: Seller. + :type seller: Party + :rtype: VatCharge + """ + + # Only telecommunications, broadcasting and electronic services are + # currently supported. + if not item_type.is_electronic_service and \ + not item_type.is_telecommunications_service and \ + not item_type.is_broadcasting_service: + raise NotImplementedError( + 'VAT charge determination for items that are not ' + 'telecommunications, broadcasting or electronic services is ' + 'currently not supported' + ) + + # Determine the rules for the countries in which the buyer and seller + # reside. + buyer_vat_rules = VAT_RULES.get(buyer.country_code, None) + seller_vat_rules = VAT_RULES.get(seller.country_code, None) + + # Test if the country to which the item is being sold enforces specific + # VAT rules for selling to the given country. + if buyer_vat_rules: + try: + return buyer_vat_rules.get_sale_to_country_vat_charge(date, + item_type, + buyer, + seller) + except NotImplementedError: + pass + + # Fall back to applying VAT rules for selling from the seller's country. + if seller_vat_rules: + try: + return seller_vat_rules.get_sale_from_country_vat_charge(date, + item_type, + buyer, + seller) + except NotImplementedError: + pass + + # Nothing we can do from here. + raise NotImplementedError( + 'cannot determine VAT charge for a sale of item %r between %r and %r' % + (item_type, seller, buyer) + ) + + +__all__ = ( + 'check_vat_number', + 'get_sale_vat_charge', + 'is_vat_number_format_valid', + ItemType.__name__, + Party.__name__, + VatCharge.__name__, + VatChargeAction.__name__, +) From 745f3308093c1ddbf69c99203b49a8a9a9293937 Mon Sep 17 00:00:00 2001 From: Rick Voormolen Date: Tue, 23 Mar 2021 14:46:56 +0100 Subject: [PATCH 4/5] Update requirements and setup --- dev-requirements.txt | 1 + setup.py | 1 + 2 files changed, 2 insertions(+) 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/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 = [ From b0703b439b8c8e9d183054e999c9d9f06ed792d9 Mon Sep 17 00:00:00 2001 From: Rick Voormolen Date: Tue, 23 Mar 2021 14:58:48 +0100 Subject: [PATCH 5/5] Remove redundant code --- pyvat/aio/__init__.py | 162 ++---------------------------------------- 1 file changed, 4 insertions(+), 158 deletions(-) diff --git a/pyvat/aio/__init__.py b/pyvat/aio/__init__.py index ee43c53..13e9c65 100644 --- a/pyvat/aio/__init__.py +++ b/pyvat/aio/__init__.py @@ -1,32 +1,10 @@ -import re -import pycountry from ..item_type import ItemType from ..party import Party from .registries import ViesRegistry from ..result import VatNumberCheckResult from ..vat_charge import VatCharge, VatChargeAction -from ..vat_rules import VAT_RULES - - -__version__ = '1.3.15' - -from .. import WHITESPACE_EXPRESSION -"""Whitespace expression. - -Used for cleaning VAT numbers. -""" - -from .. import VAT_NUMBER_EXPRESSIONS -"""VAT number expressions. - -Mapping form ISO 3166-1-alpha-2 country codes to the whitespace-less expression -a valid VAT number from the given country must match excluding the country code -prefix. - -EU VAT number structures are retrieved from `VIES -`_. -""" +from .. import __version__ VIES_REGISTRY = ViesRegistry() """VIES registry instance. @@ -68,77 +46,8 @@ validating the VAT number. """ - -def decompose_vat_number(vat_number, country_code=None): - """Decompose a VAT number and an optional country code. - - :param vat_number: VAT number. - :param country_code: - Optional country code. Default ``None`` prompting detection from the - VAT number. - :returns: - a :class:`tuple` containing the VAT number and country code or - ``(vat_number, None)`` if decomposition failed. - """ - - # Clean the VAT number. - vat_number = WHITESPACE_EXPRESSION.sub('', vat_number).upper() - - # Attempt to determine the country code of the VAT number if possible. - if not country_code: - country_code = vat_number[0:2] - - if any(c.isdigit() for c in country_code): - # Country code should not contain digits - return (vat_number, None) - - # Non-ISO code used for Greece. - if country_code == 'EL': - country_code = 'GR' - - if country_code not in VAT_REGISTRIES: - try: - if not pycountry.countries.get(alpha_2=country_code): - return (vat_number, None) - except KeyError: - # country code not found - return (vat_number, None) - vat_number = vat_number[2:] - elif vat_number[0:2] == country_code: - vat_number = vat_number[2:] - elif country_code == 'GR' and vat_number[0:2] == 'EL': - vat_number = vat_number[2:] - - return vat_number, country_code - - -def is_vat_number_format_valid(vat_number, country_code=None): - """Test if the format of a VAT number is valid. - - :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: - ``True`` if the format of the VAT number can be fully asserted as valid - or ``False`` if not. - """ - - vat_number, country_code = decompose_vat_number(vat_number, country_code) - - if not vat_number or not country_code: - return False - elif not any(c.isdigit() for c in vat_number): - return False - elif country_code not in VAT_NUMBER_EXPRESSIONS: - return False - elif not VAT_NUMBER_EXPRESSIONS[country_code].match(vat_number): - return False - - return True +from .. import decompose_vat_number +from .. import is_vat_number_format_valid def check_vat_number(vat_number, country_code=None): @@ -180,70 +89,7 @@ def check_vat_number(vat_number, country_code=None): return VAT_REGISTRIES[country_code].check_vat_number(vat_number, country_code) - -def get_sale_vat_charge(date, - item_type, - buyer, - seller): - """Get the VAT charge for performing the sale of an item. - - Currently only supports determination of the VAT charge for - telecommunications, broadcasting and electronic services in the EU. - - :param date: Sale date. - :type date: datetime.date - :param item_type: Type of the item being sold. - :type item_type: ItemType - :param buyer: Buyer. - :type buyer: Party - :param seller: Seller. - :type seller: Party - :rtype: VatCharge - """ - - # Only telecommunications, broadcasting and electronic services are - # currently supported. - if not item_type.is_electronic_service and \ - not item_type.is_telecommunications_service and \ - not item_type.is_broadcasting_service: - raise NotImplementedError( - 'VAT charge determination for items that are not ' - 'telecommunications, broadcasting or electronic services is ' - 'currently not supported' - ) - - # Determine the rules for the countries in which the buyer and seller - # reside. - buyer_vat_rules = VAT_RULES.get(buyer.country_code, None) - seller_vat_rules = VAT_RULES.get(seller.country_code, None) - - # Test if the country to which the item is being sold enforces specific - # VAT rules for selling to the given country. - if buyer_vat_rules: - try: - return buyer_vat_rules.get_sale_to_country_vat_charge(date, - item_type, - buyer, - seller) - except NotImplementedError: - pass - - # Fall back to applying VAT rules for selling from the seller's country. - if seller_vat_rules: - try: - return seller_vat_rules.get_sale_from_country_vat_charge(date, - item_type, - buyer, - seller) - except NotImplementedError: - pass - - # Nothing we can do from here. - raise NotImplementedError( - 'cannot determine VAT charge for a sale of item %r between %r and %r' % - (item_type, seller, buyer) - ) - +from .. import get_sale_vat_charge __all__ = ( 'check_vat_number',