feat: track VAT validation in the database

This makes the handling more robust and better deals with retries

Author: nijel

weblate_web/payments/migrations/0055_customer_vat_validated.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.2.8 on 2025-12-12 07:59
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("payments", "0054_remove_payment_repeat_value"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="customer",
+            name="vat_validated",
+            field=models.DateTimeField(blank=True, db_index=True, null=True),
+        ),
+    ]

weblate_web/payments/models.py

@@ -46,7 +46,7 @@
 from weblate_web.utils import get_site_url
 
 from .utils import send_notification, validate_email
-from .validators import validate_vatin
+from .validators import VAT_VALIDITY_DAYS, validate_vatin, validate_vatin_offline
 
 if TYPE_CHECKING:
     from weblate_web.invoices.models import Invoice
@@ -110,6 +110,7 @@ class Customer(models.Model):
             "Please fill in European Union VAT ID, leave blank if not applicable."
         ),
     )
+    vat_validated = models.DateTimeField(blank=True, null=True, db_index=True)
     tax = models.CharField(
         max_length=200,
         blank=True,
@@ -188,6 +189,17 @@ def __str__(self) -> str:
             return f"{self.verbose_name} ({self.email})"
         return self.verbose_name
 
+    def save(self, **kwargs) -> None:
+        if self.vat:
+            try:
+                # This should be cached due to UI validation
+                validate_vatin(self.vat)
+            except ValidationError:
+                self.vat_validated = None
+            else:
+                self.vat_validated = timezone.now()
+        super().save(**kwargs)
+
     def get_absolute_url(self) -> str:
         return reverse("crm:customer-detail", kwargs={"pk": self.pk})
 
@@ -324,17 +336,34 @@ def merge(self, other: Customer, *, user: User | None = None) -> None:
         )
         other.delete()
 
-    def validate_vatin(self):
+    def validate_vatin(self, *, automated: bool = False):
         from weblate_web.crm.models import Interaction  # noqa: PLC0415
 
+        now = timezone.now()
+
+        # Skip payment originated validation if we have validated recently
+        if (
+            not automated
+            and self.vat_validated is not None
+            and self.vat_validated > now - timedelta(days=VAT_VALIDITY_DAYS)
+        ):
+            # Perform offline validation only
+            validate_vatin_offline(self.vat)
+            return
+
         try:
             validate_vatin(self.vat)
         except ValidationError as error:
-            self.interaction_set.create(
-                origin=Interaction.Origin.VIES,
-                summary=error.code or str(error.message),
-            )
+            # Log interactive validation error
+            if automated:
+                self.interaction_set.create(
+                    origin=Interaction.Origin.VIES,
+                    summary=error.code or str(error.message),
+                )
             raise
+        else:
+            self.vat_validated = timezone.now()
+            self.save(update_fields=["vat_validated"])
 
 
 RECURRENCE_CHOICES = [

weblate_web/payments/validators.py

@@ -1,4 +1,5 @@
-from typing import NotRequired, TypedDict
+from contextlib import suppress
+from typing import TYPE_CHECKING, NotRequired, TypedDict
 
 import sentry_sdk
 from django.core.cache import cache
@@ -8,6 +9,11 @@
 from vies.types import VATIN
 from zeep.exceptions import Error
 
+if TYPE_CHECKING:
+    from django_stubs_ext import StrOrPromise
+
+VAT_VALIDITY_DAYS = 7
+
 
 class VatinValidation(TypedDict):
     valid: bool
@@ -40,18 +46,18 @@ def cache_vies_data(
             }
             sentry_sdk.capture_exception()
         else:
-            data = {
-                "valid": vies_data.valid,
-                "fault_code": vies_data.get("fault_code", ""),
-                "fault_message": vies_data.get("fault_message", ""),
-            }
-            cache.set(key, data, 3600 * 24 * 7)
+            data = {"valid": vies_data.valid}
+            with suppress(AttributeError):
+                data["fault_code"] = vies_data.fault_code
+            with suppress(AttributeError):
+                data["fault_message"] = vies_data.fault_message
+            cache.set(key, data, 3600 * 24 * VAT_VALIDITY_DAYS)
 
     return result, data
 
 
-def validate_vatin(value: str | VATIN) -> None:
-    vatin, vies_data = cache_vies_data(value)
+def validate_vatin_offline(value: str | VATIN) -> None:
+    vatin = value if isinstance(value, VATIN) else VATIN.from_str(value)
     try:
         vatin.verify_country_code()
     except ValidationError as error:
@@ -65,12 +71,18 @@ def validate_vatin(value: str | VATIN) -> None:
         msg = _("{} does not match the country's VAT ID specifications.")
         raise ValidationError(msg.format(vatin), code="Invalid VAT syntax") from error
 
+
+def validate_vatin(value: str | VATIN) -> None:
+    vatin, vies_data = cache_vies_data(value)
+    validate_vatin_offline(vatin)
+
     if not vies_data["valid"]:
         retry_errors = {"MS_UNAVAILABLE", "MS_MAX_CONCURRENT_REQ", "TIMEOUT"}
         retry_codes = {"soap:Server", "other:Error", "env:Server"}
         code = "{}: {}".format(
             vies_data.get("fault_code"), vies_data.get("fault_message")
         )
+        msg: StrOrPromise
         if (
             vies_data.get("fault_message") in retry_errors
             or vies_data.get("fault_code") in retry_codes

weblate_web/remote.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 import operator
+from datetime import timedelta
 from time import sleep
 from typing import TYPE_CHECKING, Literal, TypedDict
 
@@ -29,12 +30,11 @@
 from dateutil.parser import parse
 from django.conf import settings
 from django.core.cache import cache
-from django.db.models import F
+from django.db.models import F, Q
 from django.utils import timezone
 from wlc import Weblate, WeblateException
 
 from weblate_web.payments.models import Customer
-from weblate_web.payments.validators import cache_vies_data
 
 if TYPE_CHECKING:
     from datetime import datetime
@@ -181,12 +181,17 @@ def fetch_vat_info(*, fetch_all: bool = False, delay: int = 30) -> None:
     customers = Customer.objects.exclude(vat="").exclude(vat=None)
     if not fetch_all:
         # Fetch data once a week
+        # - entries without validation set once a week (never validated or during migration)
+        # - validated entries two days before expiry to have two attempts to fetch the data
         weekday = timezone.now().weekday()
-        customers = customers.annotate(idmod=F("id") % 7).filter(idmod=weekday)
+        customers = customers.annotate(idmod=F("id") % 7).filter(
+            (Q(idmod=weekday) & Q(vat_validated=None))
+            | Q(vat_validity__gte=timezone.now() + timedelta(days=2))
+        )
 
     for customer in customers.iterator():
         # Avoid being rate limited at their side
         sleep(delay)
 
         # Actually fetch data
-        cache_vies_data(customer.vat, force=True)
+        customer.validate_vatin(automated=True)