Integrations

The Integration framework provides a standardized way to connect external services (payment gateways, shipping providers, SMS services, etc.) to Fullfinity.

Overview

Integrations follow a provider-agnostic architecture:

┌─────────────────────────────────────────────────────────────────┐
│                     CORE MODULES                                │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │
│  │  invoicing  │  │    sales    │  │   portal    │              │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘              │
│         │                │                │                     │
│         └────────────────┼────────────────┘                     │
│                          ▼                                      │
│         ┌────────────────────────────────┐                      │
│         │     integrations (base)        │                      │
│         │  - Integration model           │                      │
│         │  - IntegrationMapper model     │                      │
│         └────────────────┬───────────────┘                      │
│                          │                                      │
│         ┌────────────────┼────────────────┐                     │
│         ▼                ▼                ▼                     │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐              │
│  │payment_stripe│ │shipping_ups │  │  sms_twilio │              │
│  │(integration) │ │(integration)│  │(integration)│              │
│  └─────────────┘  └─────────────┘  └─────────────┘              │
└─────────────────────────────────────────────────────────────────┘

Integration Model

The base Integration model in the integrations module:

class Integration(Model):
    _verbose_name = "Integration"

    name = Char(description="Integration Name", max_length=100, required=True)
    description = Text(description="Description")
    image = File(description="Image")
    # An Integration is a SHARED record (one row per provider, not per company).
    # Everything company-specific — credentials, `enabled`, journal/fee accounts — is
    # `company_scoped`, so each company plugs its own values into the same record.
    # Hence there is NO `company` FK.
    enabled = Boolean(description="Enabled", default=False, company_scoped=True)
    module = ManyToOne("Module", related_name="integrations", on_delete="SET NULL")
    state = Selection(
        choices=["Installed", "Not Installed", "Enabled"],
        calculate="get_state",
        store=False,
    )
    integration_category = Selection(
        choices=[
            "Online Payment", "Messaging", "Shipping", "Banking",
            "Email", "Calendar", "Marketplace", "E-Invoicing",
        ],
        description="Integration Category",
        hint="Which category this integration belongs to (set from each module's manifest).",
    )
    cron_sync_method = Char(description="Cron Sync Method", max_length=100)
    last_synced_date = Datetime(description="Last Synced On")

Key Fields

Field	Description
`name`	Human-readable integration name
`enabled`	Whether the integration is active
`module`	Reference to the Module that provides this integration
`integration_category`	Selection — see Integration Categories (set from the module manifest)
`cron_sync_method`	Method name for scheduled sync jobs

Computed State

state is a calculated (non-stored) Selection. A @Model.calculate method runs against the whole recordset, so iterate self and assign each record (assigning self.state = … raises CalculateAssignmentError):

@Model.calculate("module", "enabled")
async def get_state(self):
    """State is computed from module installation + enabled flag."""
    for record in self:
        await record.fetch_related("module")
        record.state = "Not Installed"
        if record.module and record.module.state == "Installed":
            record.state = "Installed"
            if record.enabled:
                record.state = "Enabled"

Creating an Integration Module

1. Module Manifest

Integration modules declare their category in manifest.yaml:

name: Stripe Payments
identifier: payment_stripe
version: '1.0'
category: module_category_integrations
integration_category: Online Payment         # Links to backbone module
description: Accept payments via Stripe
dependencies:
- online_payment                            # Backbone module dependency
icon: CreditCard
image: /static/img/stripe_logo.png
static_paths:
- img

Key manifest fields for integrations:

Field	Description
`category`	Always `module_category_integrations` for integration modules
`integration_category`	The backbone module this integration extends
`dependencies`	Must include the backbone module
`image`	Logo shown in integration list — required; the install is refused without it

You never create the Integration record yourself. The catalog is generated from these manifest keys: every module in the module_category_integrations category gets exactly one row, keyed by the module identifier, refreshed on every install, update and uninstall. Two consequences worth designing around:

Your provider is listed before it is installed. That is how a user installs it — they find your logo and description in the catalog and press Install. So the row exists while your module’s code does not; anything reading the catalog must treat state as the source of truth, not mere existence.
name, description, image, cron_sync_method and integration_category are refreshed from the manifest on each of those events. Edit them in the manifest, not on the record — a value written to those fields is overwritten on the next module operation. Everything else (credentials, enabled, provider settings your model adds) is yours and is never touched.

2. Extend Integration Model

Create a model that inherits from Integration:

from fullfinity.engine.base import *

class MyServiceIntegration(Model):
    """Extends Integration with MyService-specific fields."""
    __inherit__ = "Integration"

    # Provider-specific credential fields
    api_key = Char(
        max_length=255,
        description="API Key",
        hint="Your MyService API key",
        company_scoped=True,  # Stored per-company
    )
    api_secret = Char(
        max_length=255,
        description="API Secret",
        company_scoped=True,
    )
    sandbox_mode = Boolean(
        description="Sandbox Mode",
        default=True,
    )

    async def _is_my_service(self) -> bool:
        """Check if this integration is MyService."""
        await self.fetch_related("module")
        return self.module and self.module.identifier == "my_service"

    async def enable_integration(self):
        """Validate credentials before enabling."""
        if await self._is_my_service():
            if not self.api_key:
                raise UserError("Please enter your API Key before enabling.")
        await super().enable_integration()

3. Provider Detection Pattern

Integration methods use a detection pattern to only process their own integrations:

async def some_method(self, *args, **kwargs):
    # Check if this integration is ours
    if not await self._is_my_service():
        return await super().some_method(*args, **kwargs)

    # Handle our integration-specific logic
    ...

This pattern allows multiple integration modules to extend the same backbone.

Integration Categories

integration_category is a Selection — its value must be one of the choices defined on the Integration model. Current categories and their backbone modules:

Category	Backbone Module	Purpose
`Online Payment`	`online_payment`	Payment gateways (Stripe, PayPal, Razorpay, Adyen, Mollie, Square, Authorize.Net)
`Messaging`	`messaging`	Messaging providers (e.g. Twilio)
`Shipping`	`shipping_carrier`	Shipping carriers (DHL, FedEx, UPS, Royal Mail)
`Banking`	`bank_feed`	Bank feed aggregators (Plaid, GoCardless, TrueLayer, Basiq, Salt Edge)
`Email`	`core`	Mailbox providers using external OAuth (Gmail, Microsoft 365)
`Calendar`	`meetings`	Calendar providers (Google Calendar)
`Marketplace`	`sales`	Sales channels that import orders and publish listings (Amazon)
`E-Invoicing`	`einvoice`	E-invoice transmission networks (Storecove, Sovos)

To add a new category, extend the integration_category Selection choices on the Integration model first; only then can a manifest use it.

Always declare one. The catalog’s search view groups and browse-filters on this field, so a provider that leaves it blank still lists — in an unnamed bucket users have to scroll to find.

Test vs Live endpoints, and the sandbox override

Every category resolves its provider’s host the same way — there is no per-category mode field. Integration declares:

Field / method	What it is
`integration_mode`	`Selection(["Test", "Live"])`, `company_scoped` — one company can pilot in Test while another runs Live against the same shared record
`sandbox_url`	`Char`, `company_scoped` — send this provider’s requests somewhere else entirely. Test mode only
`api_base_url(test_url, live_url)`	The host to call: the vendor’s own, or the sandbox override
`supports_test_mode`	Computed Boolean — whether this provider distinguishes the two environments at all. Backs the two controls on the form

Resolve every request’s host through api_base_url; never branch on integration_mode yourself. That is what makes the override reach your provider, and what guarantees a Live integration can never be redirected:

MYVENDOR_TEST = "https://sandbox.myvendor.com"
MYVENDOR_LIVE = "https://api.myvendor.com"

def _base_url(self) -> str:
    return self.api_base_url(MYVENDOR_TEST, MYVENDOR_LIVE)

A vendor with one host for both networks passes the same URL twice — self.api_base_url(MYVENDOR_API, MYVENDOR_API). Many vendors pick the network from the API key rather than the hostname, so this looks redundant; it is not. It is the only reason the sandbox override can reach that provider, which is what lets you develop against a mock with no account at all.

Going Live with a Sandbox URL still stored is refused at the write, not ignored at the call — so nobody transacts for real against a stub believing otherwise.

Developing against a mock instead of a sandbox

Most vendors gate sandbox access behind an account you cannot get quickly — a merchant account, a developer signup, sometimes a regulated-entity onboarding. Two mock servers ship in this repo so a provider can be built and verified with no vendor credentials at all:

Mock	Serves	Run it
`tests/carrier_mock.py`	DHL, FedEx, UPS, Royal Mail	`python3 tests/carrier_mock.py` (:8787)
`tests/payment_mock.py`	Stripe, PayPal, Square, Adyen, Razorpay, Mollie, Authorize.Net	`python3 tests/payment_mock.py` (:8788)
`tests/bank_feed_mock.py`	Plaid, GoCardless, TrueLayer, Basiq, Salt Edge	`python3 tests/bank_feed_mock.py` (:8789)

Set the printed URL as the integration’s Sandbox URL (Test mode) and drive the app normally. Each takes ?mock_status=<code> to return that vendor’s own error envelope and ?mock_delay=<seconds> for timeout handling, plus knobs for the states that matter to its category — ?mock_decline=1 for a payment refused on a 200 response, ?mock_expired=1 for a lapsed bank consent, ?mock_empty=1 for an account with nothing to report.

Build your mock’s responses from the vendor’s published schema, not from what your provider happens to read. A mock that mirrors the provider’s assumptions confirms the provider agrees with itself; it cannot catch a field you invented. Every shape in these files came from the vendors’ own API references — which is how they caught a Stripe field that no longer exists, and an endpoint constant that swallowed its own path.

A Sandbox URL is a BASE, and the provider appends its own path to it. Two consequences:

Pass a host to api_base_url, never a full endpoint. Folding the path into the constant means the override replaces it wholesale and every request lands on the bare host.
A query-string knob does not survive being driven through a provider — it would land mid-path. To put a whole flow in a state, start a second mock already in it (serve(port=0, empty=True)).

Each provider ships an *_against_mock.py test that drives its real HTTP path against these in CI. Those tests can catch a wrong path, a wrong amount unit or a reversed sign; the transport-patched tests beside them cannot, by construction.

Bank feeds have one assertion worth copying: the five aggregators disagree about the amount sign — Plaid sends positive for money out, the others are already statement convention — and fetch_bank_transactions must normalise to positive-is-money-in whatever the vendor sends. A provider that copies another’s handling produces a ledger where every deposit is a withdrawal, reconciles to exactly the wrong number, and raises nothing.

Opting your category in

supports_test_mode is False by default: a Test/Live switch that changes no endpoint is a dead control, and showing one is worse than showing none. A category interface opts in by overriding the compute, narrowing to its own records and delegating the rest:

@Model.calculate()
async def calc_supports_test_mode(self):
    await super().calc_supports_test_mode()      # never skip this — several modules extend
    for record in self:                          # one compute, and skipping it discards
        if record.integration_category == "Banking":   # whatever the others decided
            record.supports_test_mode = True

The Mode and Sandbox URL fields are already on integration_form_view, gated on this compute — you do not add them to the form yourself.

Integration Mapper

For syncing data with external systems, use IntegrationMapper:

class IntegrationMapper(Model):
    _verbose_name = "Integration Mapper"

    local_id = Char(description="Document ID", max_length=255, required=True)
    remote_id = Char(description="Remote ID", max_length=255, required=True)
    integration = ManyToOne("Integration", related_name="mappings", on_delete="CASCADE")
    model = Char(description="Model", max_length=255, required=True)

Example usage:

# Store mapping after creating remote resource
async def sync_contact(self, contact):
    # Create in remote system
    remote_customer = await self._api_create_customer(contact)

    # Store mapping
    IntegrationMapper = get_model("IntegrationMapper")
    await IntegrationMapper.create(
        integration=self.id,
        model="Contact",
        local_id=str(contact.id),
        remote_id=remote_customer["id"],
    )

# Look up remote ID
async def get_remote_id(self, contact):
    IntegrationMapper = get_model("IntegrationMapper")
    mapping = await IntegrationMapper.filter(
        integration=self.id,
        model="Contact",
        local_id=str(contact.id),
    ).first()
    return mapping.remote_id if mapping else None

Cron Sync

For scheduled synchronization:

class MyServiceIntegration(Model):
    __inherit__ = "Integration"

    # Set sync method name (called by cron)
    cron_sync_method = "sync_my_service"

    async def sync_my_service(cls):
        """Cron job: Sync data with MyService."""
        env = env_ctx.get()
        Integration = get_model("Integration")

        # Get all enabled MyService integrations
        integrations = await Integration.filter(
            module__identifier="my_service",
            enabled=True,
        ).all()

        for integration in integrations:
            await integration._do_sync()

    async def _do_sync(self):
        """Perform actual sync logic."""
        # Fetch updates from external service
        # Update local records
        # Update last_synced_date
        self.last_synced_date = datetime.now()
        await self.save()

View Extension

Add provider-specific fields to Integration form:

Give the inheriting view its own identifier and point inherited_view: at the base view. Patch it with arch.operations (an add action targeting a stable node, position, and the value to insert). Gate each field on the record’s own identifier Char plus its state — the module relation’s fields never reach the client, so Q(module__identifier=…) cannot be used in a visible: clause. Credential fields use the PasswordInput widget.

- data_type: UiView
  name: My Service Integration Form Extension
  identifier: my_service_integration_form_extension
  type: Form
  model: Integration
  inherited_view: integration_form_view
  arch:
    operations:
    - action: add
      target:
        field: description
      position: after
      value:
        type: field
        name: api_key
        properties:
          widget: TextInput
          visible: Q(identifier='my_service') & Q(state__neq='Not Installed')
    - action: add
      target:
        field: api_key
      position: after
      value:
        type: field
        name: api_secret
        properties:
          widget: PasswordInput
          visible: Q(identifier='my_service') & Q(state__neq='Not Installed')

Best Practices

Use company_scoped=True for credential fields to store per-company
Validate credentials in enable_integration() before allowing enable
Use the detection pattern (_is_my_service()) in all override methods
Store mappings for synced records to avoid duplicates
Handle API errors gracefully with user-friendly messages
Log sync operations for debugging

Next Steps

Payment Integrations - Building payment gateway integrations
Creating Modules - General module development