coulomb/fin-hub
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add DATEV accounting adapter boundary

Browse source
This commit is contained in:
tegwick 2026-08-11 14:54:54 +02:00
parent fcec177ded
commit 90b7e97533
8 changed files with 583 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

19
src/fin_hub/accounting/__init__.py Normal file
View file

@ -0,0 +1,19 @@
"""Provider-neutral handoff of reporting data to external accounting systems."""
from fin_hub.accounting.adapters import (
AccountingCapability,
AdapterRegistry,
DatevDuoAdapter,
QontoComplementAdapter,
SchemaMigrations,
default_registry,
)
__all__ = [
"AccountingCapability",
"AdapterRegistry",
"DatevDuoAdapter",
"QontoComplementAdapter",
"SchemaMigrations",
"default_registry",
]

255
src/fin_hub/accounting/adapters.py Normal file
View file

@ -0,0 +1,255 @@
"""Versioned, side-effect-free adapters for external accounting handoff.
Adapters prepare provider-specific drafts. They deliberately do not authenticate,
transmit, issue invoices, assign invoice numbers, or perform bookkeeping.
"""
from __future__ import annotations
from collections import deque
from dataclasses import asdict, dataclass
from enum import StrEnum
from typing import Callable, Mapping, Protocol
from fin_hub.services.billing import BillingBasisExport
class AccountingCapability(StrEnum):
BILLING_BASIS_HANDOFF = "billing_basis_handoff"
INVOICE_DRAFT_WORKFLOW = "invoice_draft_workflow"
STRUCTURED_INVOICE_TRANSFER = "structured_invoice_transfer"
DOCUMENT_ARCHIVE = "document_archive"
BANK_DATA_SYNC = "bank_data_sync"
@dataclass(frozen=True)
class AdapterManifest:
adapter_id: str
provider: str
role: str
adapter_version: str
input_schema_version: str
capabilities: frozenset[AccountingCapability]
@dataclass(frozen=True)
class PreparedTransfer:
schema_version: str
artifact_type: str
adapter_id: str
adapter_version: str
provider: str
provider_role: str
source_schema_version: str
records: tuple[Mapping[str, object], ...]
exceptions: tuple[Mapping[str, object], ...]
disclaimer: str
def as_dict(self) -> dict:
return asdict(self)
class AccountingAdapter(Protocol):
manifest: AdapterManifest
def prepare(self, export: BillingBasisExport) -> PreparedTransfer: ...
Migration = Callable[[BillingBasisExport], BillingBasisExport]
class SchemaMigrations:
"""Explicit migration graph; no implicit major-version compatibility."""
def __init__(self) -> None:
self._steps: dict[tuple[str, str], Migration] = {}
def register(self, source: str, target: str, migration: Migration) -> None:
if source == target:
raise ValueError("a schema migration must change the version")
key = (source, target)
if key in self._steps:
raise ValueError(f"migration already registered: {source} -> {target}")
self._steps[key] = migration
def migrate(self, export: BillingBasisExport, target: str) -> BillingBasisExport:
if export.schema_version == target:
return export
queue = deque([(export.schema_version, ())])
visited = {export.schema_version}
path: tuple[tuple[str, str], ...] | None = None
while queue:
version, steps = queue.popleft()
for source, destination in sorted(self._steps):
if source != version or destination in visited:
continue
candidate = (*steps, (source, destination))
if destination == target:
path = candidate
queue.clear()
break
visited.add(destination)
queue.append((destination, candidate))
if path is None:
raise ValueError(
"no explicit billing-basis schema migration from "
f"{export.schema_version} to {target}"
)
migrated = export
for source, destination in path:
migrated = self._steps[(source, destination)](migrated)
if migrated.schema_version != destination:
raise ValueError(
f"migration {source} -> {destination} produced "
f"schema {migrated.schema_version}"
)
return migrated
def _prepared(
manifest: AdapterManifest,
export: BillingBasisExport,
records: tuple[Mapping[str, object], ...],
) -> PreparedTransfer:
return PreparedTransfer(
schema_version="0.1",
artifact_type="accounting_handoff_draft",
adapter_id=manifest.adapter_id,
adapter_version=manifest.adapter_version,
provider=manifest.provider,
provider_role=manifest.role,
source_schema_version=export.schema_version,
records=records,
exceptions=tuple(exception.as_dict() for exception in export.exceptions),
disclaimer=(
f"{export.disclaimer} Provider draft only; requires authorized completion "
"of customer, tax, invoice-number, issue-date, delivery, and retention data."
),
)
class DatevDuoAdapter:
"""Default handoff boundary for DATEV Unternehmen online."""
manifest = AdapterManifest(
adapter_id="datev-duo-v1",
provider="DATEV Unternehmen online",
role="authoritative_bookkeeping_and_document_archive",
adapter_version="1.0.0",
input_schema_version="0.1",
capabilities=frozenset(
{
AccountingCapability.BILLING_BASIS_HANDOFF,
AccountingCapability.STRUCTURED_INVOICE_TRANSFER,
AccountingCapability.DOCUMENT_ARCHIVE,
}
),
)
def prepare(self, export: BillingBasisExport) -> PreparedTransfer:
records = tuple(
{
"external_reference": record.billing_basis_id,
"customer_reference": record.client_id,
"engagement_reference": record.cost_attribution_key,
"performance_period": record.period_month,
"currency": record.currency,
"net_billing_basis": record.revenue,
"price_reference": record.price_id,
"source_evidence_ids": (
*record.financial_fact_ids,
*record.allocation_ids,
),
"requires_legal_and_tax_completion": True,
}
for record in export.records
)
return _prepared(self.manifest, export, records)
class QontoComplementAdapter:
"""Optional Qonto invoice-workflow draft; DATEV remains authoritative."""
manifest = AdapterManifest(
adapter_id="qonto-invoice-draft-v1",
provider="Qonto",
role="complementary_invoice_workflow_and_datev_transport",
adapter_version="1.0.0",
input_schema_version="0.1",
capabilities=frozenset(
{
AccountingCapability.BILLING_BASIS_HANDOFF,
AccountingCapability.INVOICE_DRAFT_WORKFLOW,
AccountingCapability.STRUCTURED_INVOICE_TRANSFER,
AccountingCapability.BANK_DATA_SYNC,
}
),
)
def prepare(self, export: BillingBasisExport) -> PreparedTransfer:
records = tuple(
{
"external_reference": record.billing_basis_id,
"customer_reference": record.client_id,
"description": record.cost_attribution_key,
"service_period": record.period_month,
"currency": record.currency,
"net_draft_amount": record.revenue,
"requires_legal_and_tax_completion": True,
"bookkeeping_destination": "DATEV Unternehmen online",
}
for record in export.records
)
return _prepared(self.manifest, export, records)
class AdapterRegistry:
def __init__(
self,
adapters: tuple[AccountingAdapter, ...],
*,
default_adapter_id: str,
migrations: SchemaMigrations | None = None,
) -> None:
self._adapters = {adapter.manifest.adapter_id: adapter for adapter in adapters}
if len(self._adapters) != len(adapters):
raise ValueError("adapter IDs must be unique")
if default_adapter_id not in self._adapters:
raise ValueError(f"unknown default adapter: {default_adapter_id}")
self.default_adapter_id = default_adapter_id
self.migrations = migrations or SchemaMigrations()
def manifests(self) -> tuple[AdapterManifest, ...]:
return tuple(
adapter.manifest for _, adapter in sorted(self._adapters.items())
)
def prepare(
self,
export: BillingBasisExport,
*,
adapter_id: str | None = None,
required_capabilities: frozenset[AccountingCapability] = frozenset(),
) -> PreparedTransfer:
selected_id = adapter_id or self.default_adapter_id
try:
adapter = self._adapters[selected_id]
except KeyError as error:
raise ValueError(f"unknown accounting adapter: {selected_id}") from error
missing = required_capabilities - adapter.manifest.capabilities
if missing:
names = ", ".join(sorted(missing))
raise ValueError(f"adapter {selected_id} lacks capabilities: {names}")
compatible = self.migrations.migrate(
export, adapter.manifest.input_schema_version
)
return adapter.prepare(compatible)
def default_registry() -> AdapterRegistry:
return AdapterRegistry(
(DatevDuoAdapter(), QontoComplementAdapter()),
default_adapter_id=DatevDuoAdapter.manifest.adapter_id,
)