108 lines
3.8 KiB
Python
108 lines
3.8 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] = []
|
|
|
|
|
|
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 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") -> SyncResult:
|
|
"""Upload a document/receipt. Override in connectors that support it."""
|
|
return SyncResult(success=False, errorMessage="Document upload not supported by this connector")
|