Adding a Provider

Step-by-step guide for implementing a new provider (external integration plugin).

Overview

Providers implement the plugin pattern for external integrations. Two structural approaches:

Approach	When to Use	Location
Subapp	Provider needs its own models/migrations	apps/{parent}/{provider_name}/
Module	Stateless provider (no models)	apps/{parent}/providers/{provider_name}.py

Step 1: Define the Protocol

If no protocol exists, create one in apps/{parent}/providers/__init__.py:

from typing import Protocol, Any
from decimal import Decimal
from dataclasses import dataclass, field
from common.providers import ProviderMetadataProtocol, ProviderRegistry

@dataclass(frozen=True)
class ProcessorResult:
    success: bool
    transaction_ref: str
    message: str
    details: dict[str, Any] = field(default_factory=dict)

class PaymentProcessorProtocol(ProviderMetadataProtocol, Protocol):
    async def authorize(self, *, amount: Decimal, method: str, **kwargs: Any) -> ProcessorResult: ...
    async def capture(self, *, authorization_ref: str, amount: Decimal) -> ProcessorResult: ...
    async def void(self, *, transaction_ref: str) -> ProcessorResult: ...
    async def refund(self, *, transaction_ref: str, amount: Decimal) -> ProcessorResult: ...
    async def health_check(self) -> Any: ...

payment_processor_registry = ProviderRegistry[PaymentProcessorProtocol]()

If you're adding a new implementation of an existing protocol, skip this step.

Step 2: Create the Provider Structure

For a subapp (e.g., adding Stripe to payments):

apps/payments/stripe/
├── __init__.py
├── apps.py
├── provider.py
├── schemas.py
├── migrations/
│   └── __init__.py       # Only if models needed
└── tests/
    └── test_provider.py

Step 3: Implement the Provider

# apps/payments/stripe/provider.py
import logging
from decimal import Decimal
from typing import Any
from common.provider_schemas import BaseProviderConfig, BaseProviderSecret

logger = logging.getLogger(__name__)

class StripeProcessor:
    def __init__(
        self,
        config: dict[str, Any] | None = None,
        secrets: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:
        if config is None or secrets is None:
            raise ValueError(
                "StripeProcessor requires explicit config and secrets."
            )
        self._api_key = secrets.get("api_key", "")
        self._webhook_secret = secrets.get("webhook_secret", "")

    # ProviderMetadataProtocol classmethods
    @classmethod
    def service_type(cls) -> str:
        return "payment"

    @classmethod
    def capabilities(cls) -> list[str]:
        return ["card", "ach"]

    @classmethod
    def config_model(cls) -> type[BaseProviderConfig]:
        from .schemas import StripeConfig
        return StripeConfig

    @classmethod
    def secret_model(cls) -> type[BaseProviderSecret]:
        from .schemas import StripeSecret
        return StripeSecret

    @classmethod
    def config_schema(cls) -> dict[str, Any]:
        return cls.config_model().model_json_schema()

    @classmethod
    def secret_schema(cls) -> dict[str, Any]:
        return cls.secret_model().model_json_schema()

    @classmethod
    def default_config(cls) -> dict[str, Any]:
        return cls.config_model().defaults()

    # Protocol methods (all async)
    async def health_check(self) -> Any:
        from common.provider_schemas import HealthCheckResult
        import time
        start = time.monotonic()
        try:
            # Verify credentials
            elapsed = int((time.monotonic() - start) * 1000)
            return HealthCheckResult(healthy=True, message="OK", latency_ms=elapsed)
        except Exception as exc:
            elapsed = int((time.monotonic() - start) * 1000)
            return HealthCheckResult(healthy=False, message=str(exc), latency_ms=elapsed)

    async def authorize(self, *, amount: Decimal, method: str, **kwargs: Any) -> ProcessorResult:
        try:
            # Call Stripe API
            logger.info("StripeProcessor.authorize: amount=%s", amount)
            return ProcessorResult(success=True, transaction_ref="pi_xxx", message="Authorized")
        except Exception as exc:
            logger.error("StripeProcessor.authorize error: %s", exc)
            return ProcessorResult(success=False, transaction_ref="", message=str(exc))

Step 4: Define Config/Secret Schemas

# apps/payments/stripe/schemas.py
from typing import Literal
from pydantic import Field
from common.provider_schemas import (
    BaseProviderConfig,
    BaseProviderConfigIn,
    BaseProviderSecret,
)

class StripeConfig(BaseProviderConfig):
    webhook_endpoint: str = Field(
        default="",
        description="Webhook endpoint path for Stripe events",
    )

class StripeSecret(BaseProviderSecret):
    api_key: str = Field(description="Stripe API key (sk_live_...)")
    webhook_secret: str = Field(description="Stripe webhook signing secret (whsec_...)")

class StripeProviderConfigIn(BaseProviderConfigIn):
    provider_key: Literal["stripe"] = Field(default="stripe")
    service_type: Literal["payment"] = Field(default="payment")
    config: StripeConfig = Field(default_factory=StripeConfig)
    config_secret: StripeSecret = Field(description="Stripe API credentials")
    capabilities: list[str] = Field(default=["card", "ach"])

Secret fields are encrypted at rest via EncryptedJSONField on the ProviderConfig model.

Step 5: AppConfig Registration

# apps/payments/stripe/apps.py
from django.apps import AppConfig

class StripeConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "apps.payments.stripe"
    label = "payments_stripe"

    def ready(self) -> None:
        from apps.payments.providers import payment_processor_registry
        from .provider import StripeProcessor
        payment_processor_registry.register("stripe", StripeProcessor)

Registration happens in ready() so the provider is available after Django initialization.

Step 6: Add to INSTALLED_APPS

# config/settings/base.py
PAYMENT_PROVIDER_APPS = [
    "apps.payments.authorizenet",
    "apps.payments.stripe",         # New provider
]

Step 7: Write Tests

# apps/payments/stripe/tests/test_provider.py
from unittest.mock import MagicMock, patch
import pytest
from decimal import Decimal
from ..provider import StripeProcessor

@pytest.fixture()
def processor():
    return StripeProcessor(
        config={"webhook_endpoint": "/webhooks/stripe"},
        secrets={"api_key": "sk_test_xxx", "webhook_secret": "whsec_xxx"},
    )

class TestStripeProcessor:
    async def test_authorize_success(self, processor):
        result = await processor.authorize(amount=Decimal("100.00"), method="tok_visa")
        assert result.success is True
        assert result.transaction_ref != ""

    async def test_health_check(self, processor):
        result = await processor.health_check()
        assert result.healthy is True

Mock external API calls in tests — never make real network requests.

Step 8: Configure via Admin

Create a ProviderConfig record for the tenant:

1. Navigate to Providers > Provider Configs in the admin dashboard
2. Create a new config with provider_key: "stripe" and service_type: "payment"
3. Enter the public config and secret credentials
4. Secrets are encrypted before storage

Or via the API:

POST /api/v1/providers/configs
{
    "provider_key": "stripe",
    "service_type": "payment",
    "config": {"webhook_endpoint": "/webhooks/stripe"},
    "config_secret": {"api_key": "sk_live_...", "webhook_secret": "whsec_..."},
    "capabilities": ["card", "ach"]
}

Existing Protocols

Protocol	Parent App	Implementations
PaymentProcessorProtocol	payments	Authorize.Net, Stripe
DocumentStorageProtocol	documents	S3, Local
EmailProviderProtocol	communications	SendGrid, SMTP
SMSProviderProtocol	communications	Twilio
LetterProviderProtocol	communications	(extensible)
ComplianceRuleProtocol	compliance	Built-in rules
ReportGeneratorProtocol	reporting	Built-in generators
LendingProgramProtocol	programs	Installment, Lease, LOC

See Also

• Provider Pattern --- Architecture details
• Data Protection --- Credential encryption
• Adding an App --- General app creation