259 lines
9.8 KiB
Python
259 lines
9.8 KiB
Python
"""
|
|
Fusion Accounting - External Tax Provider (Abstract)
|
|
=====================================================
|
|
|
|
Defines an abstract interface for external tax calculation services such as
|
|
Avalara AvaTax, Vertex, TaxJar, or any custom tax API. Concrete providers
|
|
inherit this model and implement the core calculation and voiding methods.
|
|
|
|
The provider model stores connection credentials and exposes a registry of
|
|
active providers per company so that invoice and order workflows can delegate
|
|
tax computation transparently.
|
|
|
|
Copyright (c) Nexa Systems Inc. - All rights reserved.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from odoo import api, fields, models, _
|
|
from odoo.exceptions import UserError, ValidationError
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FusionExternalTaxProvider(models.Model):
|
|
"""Abstract base for external tax calculation providers.
|
|
|
|
Each concrete provider (AvaTax, Vertex, etc.) inherits this model
|
|
and implements :meth:`calculate_tax` and :meth:`void_transaction`.
|
|
Only one provider may be active per company at any time.
|
|
"""
|
|
|
|
_name = "fusion.external.tax.provider"
|
|
_description = "Fusion External Tax Provider"
|
|
_order = "sequence, name"
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Fields
|
|
# -------------------------------------------------------------------------
|
|
name = fields.Char(
|
|
string="Provider Name",
|
|
required=True,
|
|
help="Human-readable label for this tax provider configuration.",
|
|
)
|
|
code = fields.Char(
|
|
string="Provider Code",
|
|
required=True,
|
|
help="Short technical identifier for the provider type (e.g. 'avatax', 'vertex').",
|
|
)
|
|
sequence = fields.Integer(
|
|
string="Sequence",
|
|
default=10,
|
|
help="Ordering priority when multiple providers are defined.",
|
|
)
|
|
provider_type = fields.Selection(
|
|
selection=[('generic', 'Generic')],
|
|
string="Provider Type",
|
|
default='generic',
|
|
required=True,
|
|
help="Discriminator used to load provider-specific configuration views.",
|
|
)
|
|
api_key = fields.Char(
|
|
string="API Key",
|
|
groups="account.group_account_manager",
|
|
help="Authentication key for the external tax service. "
|
|
"Stored encrypted; only visible to accounting managers.",
|
|
)
|
|
api_url = fields.Char(
|
|
string="API URL",
|
|
help="Base URL of the tax service endpoint.",
|
|
)
|
|
company_id = fields.Many2one(
|
|
comodel_name='res.company',
|
|
string="Company",
|
|
required=True,
|
|
default=lambda self: self.env.company,
|
|
help="Company to which this provider configuration belongs.",
|
|
)
|
|
is_active = fields.Boolean(
|
|
string="Active",
|
|
default=False,
|
|
help="Only one provider may be active per company. "
|
|
"Activating this provider will deactivate others for the same company.",
|
|
)
|
|
state = fields.Selection(
|
|
selection=[
|
|
('draft', 'Not Configured'),
|
|
('test', 'Test Passed'),
|
|
('error', 'Connection Error'),
|
|
],
|
|
string="Connection State",
|
|
default='draft',
|
|
readonly=True,
|
|
copy=False,
|
|
help="Reflects the result of the most recent connection test.",
|
|
)
|
|
last_test_message = fields.Text(
|
|
string="Last Test Result",
|
|
readonly=True,
|
|
copy=False,
|
|
help="Diagnostic message from the most recent connection test.",
|
|
)
|
|
log_requests = fields.Boolean(
|
|
string="Log API Requests",
|
|
default=False,
|
|
help="When enabled, all API requests and responses are written to the server log "
|
|
"at DEBUG level. Useful for troubleshooting but may expose sensitive data.",
|
|
)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# SQL Constraints
|
|
# -------------------------------------------------------------------------
|
|
_sql_constraints = [
|
|
(
|
|
'unique_code_per_company',
|
|
'UNIQUE(code, company_id)',
|
|
'Only one provider configuration per code is allowed per company.',
|
|
),
|
|
]
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Constraint: single active provider per company
|
|
# -------------------------------------------------------------------------
|
|
@api.constrains('is_active', 'company_id')
|
|
def _check_single_active_provider(self):
|
|
"""Ensure at most one provider is active for each company."""
|
|
for provider in self.filtered('is_active'):
|
|
siblings = self.search([
|
|
('company_id', '=', provider.company_id.id),
|
|
('is_active', '=', True),
|
|
('id', '!=', provider.id),
|
|
])
|
|
if siblings:
|
|
raise ValidationError(_(
|
|
"Only one external tax provider may be active per company. "
|
|
"Provider '%(existing)s' is already active for %(company)s.",
|
|
existing=siblings[0].name,
|
|
company=provider.company_id.name,
|
|
))
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Public API
|
|
# -------------------------------------------------------------------------
|
|
@api.model
|
|
def get_provider(self, company=None):
|
|
"""Return the active external tax provider for the given company.
|
|
|
|
:param company: ``res.company`` record or ``None`` for the current company.
|
|
:returns: A single ``fusion.external.tax.provider`` record or an empty recordset.
|
|
"""
|
|
target_company = company or self.env.company
|
|
return self.search([
|
|
('company_id', '=', target_company.id),
|
|
('is_active', '=', True),
|
|
], limit=1)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Abstract Methods (to be implemented by concrete providers)
|
|
# -------------------------------------------------------------------------
|
|
def calculate_tax(self, order_lines):
|
|
"""Compute tax amounts for a collection of order/invoice lines.
|
|
|
|
Concrete providers must override this method and return a list of
|
|
dictionaries with at least the following keys per input line:
|
|
|
|
* ``line_id`` - The ``id`` of the originating ``account.move.line``.
|
|
* ``tax_amount`` - The computed tax amount in document currency.
|
|
* ``tax_details`` - A list of dicts ``{tax_name, tax_rate, tax_amount, jurisdiction}``.
|
|
* ``doc_code`` - An external document reference for later void/commit.
|
|
|
|
:param order_lines: Recordset of ``account.move.line`` (or compatible)
|
|
containing the products, quantities, and addresses.
|
|
:returns: ``list[dict]`` as described above.
|
|
:raises UserError: When the provider encounters a non-recoverable error.
|
|
"""
|
|
raise UserError(_(
|
|
"The external tax provider '%(name)s' does not implement tax calculation. "
|
|
"Please configure a concrete provider such as AvaTax.",
|
|
name=self.name,
|
|
))
|
|
|
|
def void_transaction(self, doc_code, doc_type='SalesInvoice'):
|
|
"""Void (cancel) a previously committed tax transaction.
|
|
|
|
:param doc_code: The external document code returned by :meth:`calculate_tax`.
|
|
:param doc_type: The transaction type (default ``'SalesInvoice'``).
|
|
:returns: ``True`` on success.
|
|
:raises UserError: When the void operation fails.
|
|
"""
|
|
raise UserError(_(
|
|
"The external tax provider '%(name)s' does not implement transaction voiding.",
|
|
name=self.name,
|
|
))
|
|
|
|
def test_connection(self):
|
|
"""Verify connectivity and credentials with the external service.
|
|
|
|
Concrete providers should override this to perform an actual API ping
|
|
and update :attr:`state` and :attr:`last_test_message` accordingly.
|
|
|
|
:returns: ``True`` if the test succeeds.
|
|
"""
|
|
raise UserError(_(
|
|
"The external tax provider '%(name)s' does not implement a connection test.",
|
|
name=self.name,
|
|
))
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Actions
|
|
# -------------------------------------------------------------------------
|
|
def action_test_connection(self):
|
|
"""Button action: run the connection test and display the result."""
|
|
self.ensure_one()
|
|
try:
|
|
self.test_connection()
|
|
self.write({
|
|
'state': 'test',
|
|
'last_test_message': _("Connection successful."),
|
|
})
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _("Connection Test"),
|
|
'message': _("Connection to '%s' succeeded.", self.name),
|
|
'type': 'success',
|
|
'sticky': False,
|
|
},
|
|
}
|
|
except Exception as exc:
|
|
self.write({
|
|
'state': 'error',
|
|
'last_test_message': str(exc),
|
|
})
|
|
return {
|
|
'type': 'ir.actions.client',
|
|
'tag': 'display_notification',
|
|
'params': {
|
|
'title': _("Connection Test Failed"),
|
|
'message': str(exc),
|
|
'type': 'danger',
|
|
'sticky': True,
|
|
},
|
|
}
|
|
|
|
def action_activate(self):
|
|
"""Activate this provider and deactivate all others for the same company."""
|
|
self.ensure_one()
|
|
self.search([
|
|
('company_id', '=', self.company_id.id),
|
|
('is_active', '=', True),
|
|
('id', '!=', self.id),
|
|
]).write({'is_active': False})
|
|
self.is_active = True
|
|
|
|
def action_deactivate(self):
|
|
"""Deactivate this provider."""
|
|
self.ensure_one()
|
|
self.is_active = False
|