130 lines
5.1 KiB
Python
130 lines
5.1 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""Abstract base class and standard data models for accounting system connectors."""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import List, Optional, Dict, Any
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class AccountingBookingLine(BaseModel):
|
|
"""System-independent booking line (one debit or credit entry)."""
|
|
accountNumber: str
|
|
accountLabel: Optional[str] = None
|
|
debitAmount: float = 0.0
|
|
creditAmount: float = 0.0
|
|
currency: str = "CHF"
|
|
taxCode: Optional[str] = None
|
|
taxRate: Optional[float] = None
|
|
description: str = ""
|
|
costCenter: Optional[str] = None
|
|
reference: Optional[str] = None
|
|
|
|
|
|
class AccountingBooking(BaseModel):
|
|
"""System-independent booking (journal entry): 1 booking = 1..N lines."""
|
|
externalId: Optional[str] = None
|
|
reference: str
|
|
bookingDate: str
|
|
description: str = ""
|
|
lines: List[AccountingBookingLine] = []
|
|
externalDocumentIds: Optional[List[str]] = None # e.g. RMA Beleg-IDs, sent before booking for linking
|
|
externalDocumentLabels: Optional[List[str]] = None # display names for links (e.g. file names), one per id
|
|
|
|
|
|
class AccountingChart(BaseModel):
|
|
"""Account from the chart of accounts."""
|
|
accountNumber: str
|
|
label: str
|
|
accountType: Optional[str] = None
|
|
|
|
|
|
class SyncResult(BaseModel):
|
|
"""Result of a sync operation."""
|
|
success: bool
|
|
externalId: Optional[str] = None
|
|
externalReference: Optional[str] = None
|
|
errorMessage: Optional[str] = None
|
|
rawResponse: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
class ConnectorConfigField(BaseModel):
|
|
"""Describes a configuration field required by a connector."""
|
|
key: str
|
|
label: Dict[str, str]
|
|
fieldType: str = "text"
|
|
secret: bool = False
|
|
required: bool = True
|
|
placeholder: Optional[str] = None
|
|
|
|
|
|
class BaseAccountingConnector(ABC):
|
|
"""Abstract base for all accounting system connectors.
|
|
|
|
Each connector translates between the standardised AccountingBooking format
|
|
and the native API format of its target system.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def getConnectorType(self) -> str:
|
|
"""Unique type identifier, e.g. 'rma', 'bexio', 'abacus'."""
|
|
|
|
@abstractmethod
|
|
def getConnectorLabel(self) -> Dict[str, str]:
|
|
"""I18n display label."""
|
|
|
|
@abstractmethod
|
|
def getRequiredConfigFields(self) -> List[ConnectorConfigField]:
|
|
"""Config fields the frontend must collect for this connector."""
|
|
|
|
@abstractmethod
|
|
async def testConnection(self, config: Dict[str, Any]) -> SyncResult:
|
|
"""Verify the connection with the given credentials."""
|
|
|
|
@abstractmethod
|
|
async def getChartOfAccounts(self, config: Dict[str, Any]) -> List[AccountingChart]:
|
|
"""Load the chart of accounts from the external system."""
|
|
|
|
@abstractmethod
|
|
async def pushBooking(self, config: Dict[str, Any], booking: AccountingBooking) -> SyncResult:
|
|
"""Push a single booking to the external system."""
|
|
|
|
@abstractmethod
|
|
async def getBookingStatus(self, config: Dict[str, Any], externalId: str) -> SyncResult:
|
|
"""Query the status of a previously pushed booking."""
|
|
|
|
async def getBookingByExternalId(self, config: Dict[str, Any], externalId: str) -> SyncResult:
|
|
"""Fetch the booking in the external system by its external ID (UUID).
|
|
success=True: record exists. success=False: not found or error (e.g. deleted in Buha).
|
|
Override in connectors that support exact lookup; default = not supported."""
|
|
return SyncResult(success=False, errorMessage="Lookup by external ID not supported by this connector")
|
|
|
|
async def isBookingSynced(self, config: Dict[str, Any], booking: AccountingBooking) -> SyncResult:
|
|
"""Check with the external system if this booking already exists.
|
|
success=True: booking exists in external system (do not push again).
|
|
success=False: not found or error (allow push).
|
|
Default: success=True (trust local sync record; override in connectors that can verify via API, e.g. RMA)."""
|
|
return SyncResult(success=True)
|
|
|
|
async def pushInvoice(self, config: Dict[str, Any], invoice: Dict[str, Any]) -> SyncResult:
|
|
"""Push an invoice. Override in connectors that support it."""
|
|
return SyncResult(success=False, errorMessage="Not supported by this connector")
|
|
|
|
async def getCustomers(self, config: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""Load the customer list. Override in connectors that support it."""
|
|
return []
|
|
|
|
async def getVendors(self, config: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""Load the vendor list. Override in connectors that support it."""
|
|
return []
|
|
|
|
async def uploadDocument(
|
|
self,
|
|
config: Dict[str, Any],
|
|
fileName: str,
|
|
fileContent: bytes,
|
|
mimeType: str = "application/pdf",
|
|
comment: Optional[str] = None,
|
|
) -> SyncResult:
|
|
"""Upload a document/receipt (e.g. beleg). comment can link to booking reference. Override in connectors that support it."""
|
|
return SyncResult(success=False, errorMessage="Document upload not supported by this connector")
|