Plugin Development

This guide covers how to develop a new warecache plugin. Plugins are Python scripts that implement a defined interface, allowing the server to sync data from distributors and suppliers automatically.

Getting Started

The easiest way to create a new plugin is to copy the example template:

cp -r plugins/examples/example_distributor plugins/examples/my_distributor

Edit my_distributor/plugin.py and replace the stub implementations with real API calls.

Directory Structure

my_distributor/
  plugin.py           # Required - main plugin code
  cred.json           # Optional - API credentials
  token.json          # Optional - OAuth tokens
  cache.json          # Optional - cached API responses
  requirements.txt    # Optional - Python dependencies

Only plugin.py is required. The server passes the absolute path to the plugin directory during initialization, so the plugin can locate any additional files it needs.

Required Functions

Every plugin must implement these three functions.

`get_metadata()`

Called once at load time. Returns plugin metadata as a dictionary.

def get_metadata():
    return {
        "name": "MyDistributor",
        "api_version": 1,
        "sync_interval_seconds": 86400,
        "features": {
            "sync_pricing": True,
            "update_lifecycle": False,
            "populate_info": True,
            "check_stock": False,
            "search": False,
            "place_order": False,
            "track_order": False,
            "get_order_history": False,
            "cancel_order": False,
            "optimize_order": False,
        }
    }

Key	Type	Description
`name`	str	Distributor name. Must match the distributor in the database.
`api_version`	int	Plugin API version. Currently `1`.
`sync_interval_seconds`	int	How often the server calls sync functions (in seconds).
`features`	dict[str,bool]	Which optional functions this plugin supports.

`initialize(plugin_dir)`

Called once after loading. Set up credentials, API clients, and any global state.

def initialize(plugin_dir):
    cred_path = os.path.join(plugin_dir, "cred.json")
    with open(cred_path) as f:
        credentials = json.load(f)
    # Set up API client, validate credentials, etc.
    return True

Parameter	Type	Description
`plugin_dir`	str	Absolute path to the plugin directory.

Returns: True if initialization succeeded, False to disable the plugin.

`resolve_dist_pn(parts_list)`

Maps manufacturer part numbers to distributor part numbers. Called for parts in the database that do not yet have a mapping for this distributor.

def resolve_dist_pn(parts_list):
    result = {}
    for entry in parts_list:
        mpn = entry["mpn"]
        manufacturer = entry["manufacturer"]
        dist_pn = my_api.search(mpn, manufacturer)
        if dist_pn:
            result[mpn] = dist_pn
    return result

Parameter	Type	Description
`parts_list`	list[dict]	List of dicts with `"mpn"` and `"manufacturer"` keys.

Returns: dict[str, str] mapping {mpn: distributor_part_number}. Omit MPNs that could not be resolved.

Parts are batched in groups of 50 per call.

Feature-Gated Functions

These functions are only required if the corresponding feature is enabled in get_metadata(). The server validates that each enabled feature has a matching function with the correct signature at load time.

`get_pricing(dist_pn_list)`

Required for: sync_pricing

Fetches price breaks for a list of distributor part numbers.

def get_pricing(dist_pn_list):
    results = []
    for pn in dist_pn_list:
        results.extend([
            {"dist_pn": pn, "quantity": 1, "price": "1.25"},
            {"dist_pn": pn, "quantity": 10, "price": "1.10"},
            {"dist_pn": pn, "quantity": 100, "price": "0.85"},
        ])
    return results

Returns: list[dict] with one entry per price break tier. Each dict must contain:

Key	Type	Description
`dist_pn`	str	Distributor part number.
`quantity`	int	Minimum order quantity for this tier.
`price`	str	Unit price as a decimal string (e.g., `"0.45"`). Do not use floats.

Multiple entries per part number are expected (one per price break).

`get_parts(dist_pn_list)`

Required for: update_lifecycle or populate_info

Fetches part data for a list of distributor part numbers. This function is shared between the lifecycle and info population features.

def get_parts(dist_pn_list):
    results = []
    for pn in dist_pn_list:
        results.append({
            "dist_pn": pn,
            "lifecycle": "Production",
            "description": "IC OPAMP GP 2 CIRCUIT 8DIP",
            "rohs": "Compliant",
            "package": "DIP-8",
        })
    return results

Returns: list[dict] with one entry per part. Each dict must contain dist_pn. All other keys are optional:

Key	Type	Description
`dist_pn`	str	Distributor part number. Required.
`lifecycle`	str	One of: `Production`, `Nrnd`, `NRND`, `LastTimeBuy`, `Obsolete`, `Unknown`. Case-sensitive.
`description`	str	Part description.
`manufacturer`	str	Ignored (known key, not stored).
`manufacturer_pn`	str	Ignored (known key, not stored).
`stock`	int	Ignored (known key, not stored).
Any other key	str	Stored as a Part info field if the field is currently empty.

Custom keys (any key not in the known set above) are stored in the Part's info fields via field mapping. Existing values are never overwritten.

`search(query)`

Required for: search

Searches the distributor for parts matching a query string.

def search(query):
    return [
        {"dist_pn": "296-1395-5-ND", "description": "IC OPAMP GP 2 CIRCUIT 8DIP"},
    ]

Returns: list[dict] of search results.

Additional Features

The following features have defined interfaces but are not yet integrated into the server sync loop. They can be declared in metadata for forward compatibility.

Feature	Function	Signature
`check_stock`	`check_stock(dist_pn_list)`	list[str] -> list[dict]
`place_order`	`place_order(items)`	list[dict] -> list[dict]
`track_order`	`track_order(order_id)`	str -> dict
`get_order_history`	`get_order_history()`	() -> list[dict]
`cancel_order`	`cancel_order(order_id)`	str -> dict
`optimize_order`	`optimize_order(items)`	list[dict] -> list[dict]

Logging

Plugins use standard Python logging. The server bridges Python logs into the Rust log system automatically, so plugin log messages appear alongside server logs.

import logging
log = logging.getLogger(__name__)

log.info("Loaded credentials")
log.warning("Rate limited, retrying")
log.error("API returned 500")

Log levels map as: CRITICAL/ERROR -> Error, WARNING -> Warn, INFO -> Info, DEBUG -> Debug.

Lifecycle

Discovery: Server scans the plugin directory for subdirectories containing plugin.py
Loading: Module is imported, get_metadata() is called, function signatures are validated
Initialization: initialize(plugin_dir) is called. If it returns False, the plugin is disabled
Sync loop: The server periodically calls the plugin's functions based on enabled features and the configured sync interval
Hot reload: If enabled, file changes trigger automatic module reloading (500ms debounce)

Error Handling

If initialize() fails, the plugin is disabled in the database but other plugins continue normally
If a sync call raises an exception, the error is logged and the sync cycle continues
If hot-reload encounters a syntax error, the last working version is retained
Invalid lifecycle values are logged as errors and the part is not updated