import re
import secrets
import string
from onegov.core.orm import Base
from onegov.core.orm.mixins import TimestampMixin
from onegov.core.orm.types import UUID
from onegov.core.utils import batched, hash_dictionary
from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Text
from sqlalchemy import UniqueConstraint
from sqlalchemy.orm import relationship, validates
from typing import Any, ClassVar, TYPE_CHECKING
if TYPE_CHECKING:
import uuid
from sqlalchemy.orm import Session
from .invoice import Invoice
[docs]
KNOWN_SCHEMAS: dict[str, type['Schema']] = {}
[docs]
INVALID_REFERENCE_CHARS_EX = re.compile(r'[^Q0-9A-F]+')
[docs]
REFERENCE_EX = re.compile(r'Q{1}[A-F0-9]{10}')
[docs]
class InvoiceReference(Base, TimestampMixin):
""" A reference pointing to an invoice. References are keys which are used
outside the application. Usually a code used on an invoice to enter through
online-banking.
Each invoice may have multiple references pointing to it. Each refernce
is however unique.
There are multiple schemas for references. Each schema generates its own
set of references using a Python class and some optional settings given
by the user (say a specific bank's account number).
Each schema may only reference an invoice once.
Though each schema has its own set of references, the references-space
is shared between all schemas. In other words, reference 'foo' of schema
A would conflict with reference 'foo' of schema B.
This is because we do not know which schema was used when we encounter
a reference.
In reality this should not be a problem as reference schemes provided
by banks usually cover a very large space, so multiple schemas are expected
to just generate random numbers until one is found that has not been used
yet (which should almost always happen on the first try).
"""
[docs]
__tablename__ = 'invoice_references'
#: the unique reference
[docs]
reference: 'Column[str]' = Column(Text, primary_key=True)
#: the referenced invoice
[docs]
invoice_id: 'Column[uuid.UUID]' = Column(
UUID, # type:ignore[arg-type]
ForeignKey('invoices.id'),
nullable=False
)
[docs]
invoice: 'relationship[Invoice]' = relationship(
'Invoice',
back_populates='references'
)
#: the schema used to generate the invoice
[docs]
schema: 'Column[str]' = Column(Text, nullable=False)
#: groups schema name and its config to identify records created by a
#: given schema and config
[docs]
bucket: 'Column[str]' = Column(Text, nullable=False)
[docs]
__table_args__ = (
UniqueConstraint(
'bucket', 'invoice_id', name='unique_bucket_invoice_id'),
)
@validates
[docs]
def validate_schema(self, field: str, value: str) -> str:
if value not in KNOWN_SCHEMAS:
raise ValueError(f'{value} is an unknown schema')
return value
@property
[docs]
def readable(self) -> str:
""" Returns the human formatted variant of the reference. """
return KNOWN_SCHEMAS[self.schema]().format(self.reference)
[docs]
class Schema:
""" Defines the methods that need to be implemented by schemas. Schemas
should generate numbers and be able to format them.
Schemas should never be deleted as we want to be able to display
past schemas even if a certain schema is no longer in use.
If a new schema comes along that replaces an old one in an incompatible
way, the new schema should get a new name and should be added alongside
the old one.
"""
[docs]
def __init_subclass__(cls, name: str, **kwargs: object) -> None:
super().__init_subclass__(**kwargs)
cls.name = name
KNOWN_SCHEMAS[cls.name] = cls
def __init__(self, **config: object) -> None:
""" Each schema may have a custom config. This is *only used for
creation of references*. For other uses like formatting the config
is not passed in.
"""
# FIXME: We should be explicit about what config items we expect
for k, v in config.items():
setattr(self, k, v)
@property
[docs]
def bucket(self) -> str:
""" Generates a unique identifer for the current schema and config. """
return self.render_bucket(self.name, self.config)
@classmethod
[docs]
def render_bucket(
cls,
schema_name: str,
schema_config: dict[str, Any] | None = None
) -> str:
if schema_config:
return f'{schema_name}-{hash_dictionary(schema_config)}'
return schema_name
[docs]
def link(
self,
session: 'Session',
invoice: 'Invoice',
optimistic: bool = False,
flush: bool = True
) -> InvoiceReference | None:
""" Creates a new :class:`InvoiceReference` for the given invoice.
The returned invoice should have a unique reference, so the chance
of ending up with a conflict error later down the line are slim.
If the schema already has a linke to the invoice, we skip the
creation.
By default we check our constraints before we write to the database.
To be faster in performance critical situation we can however also
chose to be 'optimistic' and forego those checks. Due to the random
nature of schema references this should usually work.
The constraints are set on the database, so they will be enforced
either way.
Additionally we can forego the session.flush if we want to.
"""
assert invoice.id, 'the invoice id must be konwn'
q = None if optimistic else session.query(InvoiceReference)
# check if we are already linked
if q is not None:
if q.filter_by(bucket=self.bucket, invoice_id=invoice.id).first():
return None
# find an unused reference
for i in range(10):
candidate = self.new()
if q is not None:
if q.filter_by(reference=candidate).first():
continue
break
else:
raise RuntimeError('No unique reference after 10 tries')
reference = InvoiceReference(
invoice_id=invoice.id,
reference=candidate,
schema=self.name,
bucket=self.bucket,
)
session.add(reference)
if flush:
session.flush()
return reference
[docs]
def new(self) -> str:
""" Returns a new reference in the most compact way possible. """
raise NotImplementedError()
[docs]
class FeriennetSchema(Schema, name='feriennet-v1'):
""" The default schema for customers without specific bank integrations.
The generated reference is entered as a note when conducting the
online-banking transaction.
"""
[docs]
def new(self) -> str:
return f'q{secrets.token_hex(5)}'
[docs]
class ESRSchema(Schema, name='esr-v1'):
""" The default schema for ESR by Postfinance. In it's default form it is
random and requires no configuration.
A ESR reference has 27 characters from 0-9. The first 26 characters can
be chosen freely and the last character is the checksum.
Some banks require that references have certain prefixes/suffixes, but
Postfinance who defined the ESR standard does not.
"""
[docs]
def new(self) -> str:
number = ''.join(secrets.choice(string.digits) for _ in range(26))
return number + self.checksum(number)
[docs]
def checksum(self, number: str) -> str:
""" Generates the modulo 10 checksum as required by Postfinance. """
table = (0, 9, 4, 6, 8, 2, 7, 1, 3, 5)
carry = 0
for n in number:
carry = table[(carry + int(n)) % 10]
return str((10 - carry) % 10)
[docs]
class RaiffeisenSchema(ESRSchema, name='raiffeisen-v1'):
""" Customised ESR for Raiffeisen. """
# FIXME: We should override __init__ so we error if this is missing
[docs]
esr_identification_number: str
[docs]
def new(self) -> str:
ident = self.esr_identification_number.replace('-', '').strip()
assert 3 <= len(ident) <= 7
rest = 26 - len(ident)
random = ''.join(secrets.choice(string.digits) for _ in range(rest))
number = f'{self.esr_identification_number}{random}'
return number + self.checksum(number)