Strategy Pattern¶

Every business decision point in Vectis is an abstract strategy interface. The core engine calls strategies; extensions and modules provide implementations.

Strategy ABCs¶

Strategy	Purpose	Resolution
`PriceCalculation`	Tiered / contract pricing	`resolve`
`TaxCalculation`	Tax rates and remittance	`resolve_all`
`ShippingCalculator`	Rate calculation	`resolve_all`
`ShippingEligibility`	Method eligibility rules	`resolve_all`
`InventoryAllocation`	Stock reservation	`resolve`
`PaymentProcess`	Authorize, capture, charge, void, refund	`resolve`
`PaymentEligibility`	Payment method filtering	`resolve_all`
`OrderProcess`	Order state machine hooks	`resolve`
`SearchIndexing`	Full-text index sync	`resolve`
`FraudScoring`	Risk evaluation	`resolve`
`EmailDelivery`	Transactional email	`resolve`
`SmsDelivery`	Transactional SMS	`resolve` (via `SmsDeliveryStrategy` in `notification/sms_strategies.py`)
`CommissionCalculation`	Sales rep commissions	`resolve`
`ReturnProcess`	RMA / return workflow	`resolve`
`CatalogVisibility`	Account-level product access	`resolve_all`

PaymentProcessStrategy¶

The payment strategy is the most feature-rich strategy ABC. It defines the full payment lifecycle:

class PaymentProcessStrategy(ABC):
    @property
    @abstractmethod
    def gateway_name(self) -> str: ...

    @abstractmethod
    async def authorize(self, amount, currency, payment_data) -> PaymentResult: ...

    @abstractmethod
    async def capture(self, transaction_id, amount) -> PaymentResult: ...

    @abstractmethod
    async def refund(self, transaction_id, amount) -> PaymentResult: ...

    async def charge(self, amount, currency, payment_data) -> PaymentResult:
        """Auth + capture in one call. Default: authorize then capture."""
        ...

    async def void(self, transaction_id) -> PaymentResult:
        """Void an authorized but uncaptured transaction."""
        ...

    async def approve_held_transaction(self, transaction_id) -> PaymentResult:
        """Release a transaction held by fraud filters."""
        ...

The PaymentResult dataclass includes a held_for_review flag for fraud filter detection:

@dataclass
class PaymentResult:
    success: bool
    transaction_id: str | None = None
    error_message: str | None = None
    gateway_response: dict | None = None
    held_for_review: bool = False

Creating a Strategy¶

Define an ABC, then register implementations with strategy_resolver.

from abc import ABC, abstractmethod
from decimal import Decimal

class TaxCalculationStrategy(ABC):
    @abstractmethod
    async def calculate(self, line_items: list[dict], address: dict) -> Decimal: ...

Register a default at startup:

from vectis.core.strategy import strategy_resolver

class SimpleSalesTax(TaxCalculationStrategy):
    async def calculate(self, line_items, address):
        subtotal = sum(i["price"] * i["qty"] for i in line_items)
        return subtotal * Decimal("0.08")

strategy_resolver.register(TaxCalculationStrategy, SimpleSalesTax(), name="sales_tax")

Single vs Multi Resolution¶

resolve — returns the last-registered implementation (or a named one). Use for single-active-strategy decision points:

price_strategy = strategy_resolver.resolve(PriceCalculationStrategy)
price = await price_strategy.calculate(variant_id, account_id)

resolve_all — returns every implementation. Use when multiple strategies combine results (e.g., tax jurisdictions):

tax_strategies = strategy_resolver.resolve_all(TaxCalculationStrategy)
total_tax = sum([await s.calculate(items, addr) for s in tax_strategies])

Channel-Scoped Filtering¶

Extensions are activated per-channel via ChannelExtension. Pass channel_enabled to filter strategies at resolution time:

from vectis.core.strategy import _load_channel_extensions

enabled = await _load_channel_extensions(channel_id)
strategies = strategy_resolver.resolve_all(TaxCalculationStrategy, channel_enabled=enabled)

Core defaults (no extension_name) always pass the filter.

Note

The channel extension cache is invalidated when an admin toggles an extension. Call invalidate_channel_extension_cache(channel_id) if you modify ChannelExtension rows programmatically.

Overriding a Default¶

Extensions override by registering the same strategy type. resolve() returns the last-registered implementation, so extensions loaded after core win:

strategy_resolver.register(
    TaxCalculationStrategy, ExciseIQTax(), name="exciseiq", extension_name="exciseiq",
)

Warning

Two extensions registering the same strategy type: last wins. Use conflicts_with in ExtensionManifest to declare mutual exclusivity.