Service Interfaces¶

Protocol-based dependency injection layer for calculatecost4. Each interface defines a Python Protocol class in calculatecost4/interfaces/, with one or more concrete adapters in calculatecost4/adapters/. Core business logic depends only on protocols, never on adapters or external services directly.

Design¶

The interface layer enforces a strict boundary between core logic and external I/O:

Handler
  -> Core (depends on Protocol interfaces only)
       -> Adapter (implements Protocol, calls external service)
            -> External Service (DynamoDB, HTTP, Lambda)

All adapters that hit external services use diskcache.Cache('/tmp/') with appropriate TTLs (default 1000 seconds). Caching lives in the adapter layer, not in core logic.

PricingService¶

File: calculatecost4/interfaces/pricing.py

Abstracts product price lookups. The core pipeline calls this to enrich each product with its selling price and original/list price.

Methods¶

Method	Signature	Returns	On failure
`get_price`	`(cprcode: int, brcode: int) -> float`	Current selling price	Raises `ProductPriceNotFoundException`
`get_original_price`	`(cprcode: int, brcode: int) -> float`	Original/list price	Returns `0` (does not raise)

Adapter: DynamoDBPricingService¶

File: calculatecost4/adapters/dynamodb_pricing.py

Uses PynamoDB PriceModel against DynamoDB table price-database-2-dev-manual. Each record contains compressed feather data. The adapter decompresses and loads the data into a pandas DataFrame, then performs the price lookup by cprcode and brcode.

External service: DynamoDB table price-database-2-dev-manual

Legacy equivalent: product/getPrice.py

Mock Example¶

class MockPricingService:
    def __init__(self, prices: dict[tuple[int, int], float]):
        self._prices = prices

    def get_price(self, cprcode: int, brcode: int) -> float:
        key = (cprcode, brcode)
        if key not in self._prices:
            raise ProductPriceNotFoundException(f"No price for {cprcode}/{brcode}")
        return self._prices[key]

    def get_original_price(self, cprcode: int, brcode: int) -> float:
        return self._prices.get((cprcode, brcode), 0)

ProductLookupService¶

File: calculatecost4/interfaces/product_lookup.py

Abstracts product master data lookups. The core pipeline calls this to retrieve product attributes (primarily weight) for each product in the order.

Methods¶

Method	Signature	Returns	On failure
`get_product`	`(cprcode: int) -> dict`	Product data dict from database	Raises `ProductNotFoundError`

The returned dict must include at minimum a weight key. Other product attributes may be present but are not required by the core pipeline.

Adapter: DynamoDBProductLookup¶

File: calculatecost4/adapters/dynamodb_product.py

Uses PynamoDB DynamoProduct model against DynamoDB table product-table-dev-manual with a GSI on cprcode.

External service: DynamoDB table product-table-dev-manual (GSI on cprcode)

Legacy equivalent: product/getProduct.py

Mock Example¶

class MockProductLookup:
    def __init__(self, products: dict[int, dict] | None = None):
        self._products = products or {}

    def get_product(self, cprcode: int) -> dict:
        if cprcode not in self._products:
            raise ProductNotFoundError(f"Product {cprcode} not found")
        return self._products[cprcode]

Usage with defaults:

mock = MockProductLookup({
    146098: {"weight": 0.5},
    200001: {"weight": 1.2},
})

ControlledProductService¶

File: calculatecost4/interfaces/controlled_products.py

Abstracts the controlled-product list fetch. The core pipeline calls this once per request to determine which cprcodes are controlled products. Controlled products are excluded from totalExcludeControlledProducts.

Methods¶

Method	Signature	Returns	On failure
`get_controlled_product_list`	`() -> list[int]`	List of cprcodes that are controlled products	Implementation-defined

Adapter: HTTPControlledProducts¶

File: calculatecost4/adapters/http_controlled_products.py

Performs an HTTP GET to https://shop.villamarket.com/api/group/read/5000000153 and parses the response into a list of integer cprcodes.

External service: HTTP endpoint https://shop.villamarket.com/api/group/read/5000000153

Legacy equivalent: product/getControlledProductList.py

Mock Example¶

class MockControlledProducts:
    def __init__(self, controlled: list[int] | None = None):
        self._controlled = controlled or []

    def get_controlled_product_list(self) -> list[int]:
        return self._controlled

Usage:

mock = MockControlledProducts([100001, 100002, 100003])

ShippingPriceService¶

File: calculatecost4/interfaces/shipping_price.py

Abstracts shipping fee calculation. Two distinct methods cover the two delivery models: Bangkok polygon-based delivery and nationwide weight-based delivery.

Methods¶

Method	Signature	Returns	On failure
`get_regular_price`	`(lat: float, lon: float, brcode: int) -> float`	Bangkok delivery fee	Raises `BranchNotAvailableException` if branch not in polygon
`get_nationwide_price`	`(postcode: str, weight: float, mode: str) -> float`	Nationwide delivery cost	Implementation-defined

get_regular_price is used for both REGULAR and EXPRESS shipping modes. get_nationwide_price is used for NATIONWIDE mode. PICKUP mode has zero cost and does not call either method.

Adapters¶

Two separate adapters implement the two methods. They may be combined into a single adapter or kept separate depending on implementation preference.

HTTPRegularPrice

File: calculatecost4/adapters/http_regular_price.py

Calls the Bangkok polygon API with latitude, longitude, and branch code. Returns the delivery fee for the given location. Raises BranchNotAvailableException if the branch does not service the requested coordinates.

External service: HTTP polygon API

Legacy equivalent: shipping/regularPrice.py

LambdaNationwidePrice

File: calculatecost4/adapters/lambda_nationwide.py

Invokes AWS Lambda function get-nationwide-delivery-cost2-master with postcode, total weight, and nationwide mode (DRY, FRESH, MIX). Returns the calculated shipping cost.

External service: Lambda get-nationwide-delivery-cost2-master

Legacy equivalent: shipping/nationwidePrice.py

Mock Example¶

class MockShippingPrice:
    def __init__(
        self,
        regular_price: float = 60.0,
        nationwide_price: float = 270.0,
        unavailable_branches: set[int] | None = None,
    ):
        self._regular = regular_price
        self._nationwide = nationwide_price
        self._unavailable = unavailable_branches or set()

    def get_regular_price(self, lat: float, lon: float, brcode: int) -> float:
        if brcode in self._unavailable:
            raise BranchNotAvailableException(f"Branch {brcode} not in polygon")
        return self._regular

    def get_nationwide_price(self, postcode: str, weight: float, mode: str) -> float:
        return self._nationwide

CouponService¶

File: calculatecost4/interfaces/coupon.py

Abstracts coupon validation and discount calculation. The core pipeline calls this when the order includes one or more coupon codes. Returns a DiscountResult containing per-coupon discount breakdowns.

Methods¶

Method	Signature	Returns	On failure
`check_coupons`	`(order_dict: dict) -> DiscountResult`	Validated discounts for the order	Implementation-defined

The order_dict parameter is a serialized CouponOrder containing products, branch, owner, shipping, coupon codes, and financial summaries. See 02-business-logic.md section 5 for the exact payload structure.

DiscountResult.discounts is a list of Discount objects, each containing:

Field	Type	Description
`coupon`	str	The coupon code
`discount`	float	Cart-level discount amount
`bogoDiscount`	float	Buy-one-get-one discount amount
`two4Discount`	float	Two-for discount amount
`percentageDiscount`	float	Percentage discount multiplier
`freeshipping`	bool	Whether this coupon grants free shipping
`shippingDiscount`	float	Shipping discount amount

Adapter: LambdaCouponService¶

File: calculatecost4/adapters/lambda_coupon.py

Invokes AWS Lambda function coupon3-checker-dev with the order dict payload. Parses the Lambda response into a DiscountResult.

This adapter will be internalized in a future phase (coupon logic moved into calculatecost4 directly).

External service: Lambda coupon3-checker-dev

Legacy equivalent: coupon/couponCalculator.py

Mock Example¶

class MockCouponService:
    def __init__(self, discounts: list[Discount] | None = None):
        self._discounts = discounts or []

    def check_coupons(self, order_dict: dict) -> DiscountResult:
        return DiscountResult(discounts=self._discounts)

Usage with a configured discount:

mock = MockCouponService([
    Discount(
        coupon="SAVE20",
        discount=20.0,
        bogoDiscount=0.0,
        two4Discount=0.0,
        percentageDiscount=0.0,
        freeshipping=False,
        shippingDiscount=0.0,
    )
])

VoucherService¶

File: calculatecost4/interfaces/voucher.py

Abstracts voucher retrieval. The core pipeline calls this for each voucher ID in the order to fetch voucher details and validate eligibility.

Methods¶

Method	Signature	Returns	On failure
`get_voucher`	`(voucher_id: str) -> Voucher`	Voucher object with value, status, ownership	Raises `VoucherNotFoundError`

The returned Voucher object must include at minimum:

Field	Type	Description
`value`	float	Discount value (negative number)
`isActive`	bool	Whether the voucher is currently active
`usedOnOrder`	str	Order ID if already redeemed, empty string if unused
`usedOnOwner`	str	Owner ID restriction (empty if unrestricted)

Adapter: LambdaVoucherService¶

File: calculatecost4/adapters/lambda_voucher.py

Invokes AWS Lambda function get-voucher-master with the voucher ID. Parses the Lambda response into a Voucher object.

External service: Lambda get-voucher-master

Legacy equivalent: voucher/getVoucher.py

Mock Example¶

class MockVoucherService:
    def __init__(self, vouchers: dict[str, Voucher] | None = None):
        self._vouchers = vouchers or {}

    def get_voucher(self, voucher_id: str) -> Voucher:
        if voucher_id not in self._vouchers:
            raise VoucherNotFoundError(f"Voucher {voucher_id} not found")
        return self._vouchers[voucher_id]

Usage:

mock = MockVoucherService({
    "WNOB0OWQ": Voucher(
        value=-100.0,
        isActive=True,
        usedOnOrder="",
        usedOnOwner="",
    ),
})

Interface-Adapter-External Service Mapping¶

Interface	Adapter	External Service
PricingService	DynamoDBPricingService	DynamoDB: `price-database-2-dev-manual`
ProductLookupService	DynamoDBProductLookup	DynamoDB: `product-table-dev-manual` (GSI: cprcode)
ControlledProductService	HTTPControlledProducts	HTTP: `shop.villamarket.com/api/group/read/5000000153`
ShippingPriceService	HTTPRegularPrice	HTTP: polygon API
ShippingPriceService	LambdaNationwidePrice	Lambda: `get-nationwide-delivery-cost2-master`
CouponService	LambdaCouponService	Lambda: `coupon3-checker-dev`
VoucherService	LambdaVoucherService	Lambda: `get-voucher-master`

Caching Strategy¶

All adapters that call external services use diskcache.Cache('/tmp/') for caching responses. The cache lives in the adapter layer so that:

Core logic is unaware of caching and always calls the protocol methods directly.
Cache TTLs can be tuned per-adapter without changing the interface contract.
Tests use mock implementations that bypass caching entirely.

Default TTL is 1000 seconds. Adapters may override this per-method if the underlying data has different staleness tolerances (e.g., controlled product list changes rarely, prices change more frequently).

Lambda /tmp/ is shared across warm invocations of the same container, so the diskcache persists across sequential invocations within the same Lambda lifecycle.

Writing a New Mock¶

All mocks follow the same pattern:

Accept configuration via constructor arguments (dicts, lists, fixed return values).
Implement the protocol methods using the injected data.
Raise the same domain exceptions as the real adapter for error paths.
Do not import any adapter code, AWS SDK, or external libraries.

class MockSomeService:
    def __init__(self, data=None):
        self._data = data or {}

    def some_method(self, key: str) -> dict:
        if key not in self._data:
            raise SomeDomainException(f"Not found: {key}")
        return self._data[key]

Mocks are used in unit tests to exercise the core pipeline without network calls, AWS credentials, or DynamoDB tables. Each test configures its mock with exactly the data needed for that test case.