directivsys-sdk-python 0.1.1

Python SDK for the DirectivSys Analytics Platform

directivsys-sdk-python

directivsys-sdk-python is a Python translation of the DirectivSys .NET SDK. It provides a typed async client for DirectivSys analytics ingestion, batch synchronization, payload result polling, directive status updates, and webhook-driven directive execution.

Highlights

Capability	Description
Strict typing	Pydantic models and explicit type hints throughout the public API
Async-first client	Built around httpx.AsyncClient and asyncio
Batch ingestion	Supports polymorphic analytic item batches with $type discrimination
Directive execution	Includes a base executor, registry, and webhook processing service
Resilience	Retry and circuit-breaker behavior for directive execution
FastAPI support	Optional webhook router for FastAPI applications

Installation

pip install directivsys-sdk-python

For FastAPI webhook support:

pip install "directivsys-sdk-python[fastapi]"

For development and tests:

pip install "directivsys-sdk-python[dev]"

Quick start

import asyncio

from directivsys_sdk import DirectivSysClient, DirectivSysOptions
from directivsys_sdk.models import InventoryHealthPayload


async def main() -> None:
    options = DirectivSysOptions(api_key="your-api-key")

    async with DirectivSysClient(options) as client:
        response = await client.submit_inventory_health(
            InventoryHealthPayload(
                sku="SKU-123",
                stock_level=18,
                reorder_point=25,
                safety_stock=10,
                forecasted_demand_next30_days=42,
                average_lead_time_days=7,
            )
        )
        print(response.payload_id, response.status)


asyncio.run(main())

Webhook and executor usage

The SDK includes a framework-agnostic record processor and a FastAPI router helper. You register concrete directive executors in your application and allow the service to auto-run directives whose status is auto-execute.

import logging

from directivsys_sdk import BaseDirectiveExecutor, DirectivSysClient, DirectivSysOptions
from directivsys_sdk.executors import DirectiveRegistry
from directivsys_sdk.models import DirectiveExecutionResult, ImpactData
from directivsys_sdk.service import DirectivSysService
from directivsys_sdk.webhooks import create_directivsys_router
from fastapi import FastAPI


class ReorderSupplierExecutor(BaseDirectiveExecutor):
    async def execute_internal(self, parameters: dict[str, object]) -> DirectiveExecutionResult:
        sku = str(parameters.get("sku", ""))
        quantity = int(parameters.get("quantity", 0))
        return DirectiveExecutionResult(
            success=True,
            message=f"Created reorder workflow for {sku}",
            impact_data=ImpactData(
                entity_id=sku,
                property_changed="PurchaseOrderStatus",
                original_value=None,
                new_value="Pending",
                insight_delta=quantity,
            ),
        )


app = FastAPI()
options = DirectivSysOptions(api_key="your-api-key")
client = DirectivSysClient(options)
registry = DirectiveRegistry()
registry.register("reorderSupplier", ReorderSupplierExecutor(client, logging.getLogger("reorder")))
service = DirectivSysService(registry)
app.include_router(create_directivsys_router(service, options))

Package layout

Module	Purpose
directivsys_sdk.client	Async API client
directivsys_sdk.models	Typed payloads, records, items, and responses
directivsys_sdk.executors	Base directive executor and registry
directivsys_sdk.service	Webhook record processing service
directivsys_sdk.resilience	Retry and circuit-breaker behavior
directivsys_sdk.webhooks.fastapi	FastAPI router helper

Notes on concurrency

The SDK is intentionally async-first. It is designed to preserve the concurrency-oriented spirit of the .NET SDK without pretending that Python’s event loop is identical to .NET’s task scheduler. Async boundaries are explicit, HTTP calls are non-blocking, and directive execution resilience is centralized rather than hidden across ad hoc helper code.