Python SDK

pip install justkalm-sdk

pip install git+https://github.com/mybffjustin/justkalm-python.git

import asyncio
from justkalm import JustKalmClient, ProductCreateParams

async def main():
    # Initialize the client with your API key
    async with JustKalmClient(
        api_key="jk_live_your_api_key_here",
        environment="production"  # or "sandbox" for testing
    ) as client:
        
        # Create a product
        product = await client.products.create(ProductCreateParams(
            name="Organic Cotton T-Shirt",
            brand="EcoWear",
            category="apparel",
            materials={
                "organic_cotton": 0.95,
                "elastane": 0.05
            },
            certifications=["GOTS", "Fair Trade"]
        ))
        
        print(f"Created product: {product.id}")
        
        # Get valuation with health scoring
        valuation = await client.valuations.create(
            product_id=product.id,
            include_health=True,
            include_sustainability=True
        )
        
        print(f"Health Score: {valuation.health_score}/100")
        print(f"Sustainability Score: {valuation.sustainability_score}/100")
        print(f"Estimated Value: ${valuation.adjusted_value:.2f}")

if __name__ == "__main__":
    asyncio.run(main())

from justkalm import ValuationRequest

# Create a valuation
valuation = await client.valuations.create(
    product_id="prod_abc123",
    include_health=True,
    include_sustainability=True,
    include_market_analysis=True
)

# Access valuation results
print(f"Health Score: {valuation.health_score}/100")
print(f"Sustainability Score: {valuation.sustainability_score}/100")
print(f"Confidence: {valuation.confidence:.1%}")
print(f"Base Value: ${valuation.base_value:.2f}")
print(f"Adjusted Value: ${valuation.adjusted_value:.2f}")

# Access detailed breakdown
for factor in valuation.factors:
    print(f"  {factor.name}: {factor.impact:+.1%}")

# Get health details
health = valuation.health_details
print(f"Chemical Safety: {health.chemical_safety}/100")
print(f"Material Toxicity: {health.material_toxicity}/100")
for risk in health.identified_risks:
    print(f"  ⚠️ {risk.description}")

from justkalm.exceptions import (
    JustKalmError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NotFoundError,
    ServerError
)

try:
    product = await client.products.get("prod_invalid")
except NotFoundError as e:
    print(f"Product not found: {e.message}")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
    # The SDK handles retries automatically by default
except ValidationError as e:
    print(f"Invalid request: {e.message}")
    for error in e.errors:
        print(f"  - {error.field}: {error.message}")
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")
except ServerError as e:
    print(f"Server error (retryable): {e.message}")
except JustKalmError as e:
    # Catch-all for any SDK error
    print(f"Error: {e.message}")

from justkalm import JustKalmClient, RateLimitConfig

# Configure rate limiting
async with JustKalmClient(
    api_key="jk_live_...",
    rate_limit=RateLimitConfig(
        requests_per_second=10,
        requests_per_minute=500,
        burst_size=20,
        wait_on_limit=True  # Wait instead of raising error
    )
) as client:
    # SDK automatically respects limits
    for product_id in product_ids:
        await client.products.get(product_id)

# Check current rate limit status
status = client.rate_limit_status()
print(f"Remaining: {status.remaining}/{status.limit}")
print(f"Resets at: {status.reset_at}")

from justkalm import JustKalmClient, RetryConfig

# Configure retry behavior
async with JustKalmClient(
    api_key="jk_live_...",
    retry=RetryConfig(
        max_retries=3,
        initial_delay=0.5,      # 500ms initial delay
        max_delay=30.0,         # Max 30s delay
        exponential_base=2.0,   # Double delay each retry
        jitter=True,            # Add random jitter
        retryable_errors=[      # Errors to retry
            "rate_limit",
            "server_error",
            "connection_error"
        ]
    )
) as client:
    # Retries happen automatically
    product = await client.products.get("prod_abc123")

from justkalm import BatchProcessor, ProductCreateParams

# Process large datasets efficiently
async with client.batch(
    concurrency=5,           # 5 concurrent requests
    chunk_size=100,          # Process 100 items at a time
    on_error="continue"      # Continue on individual failures
) as batch:
    
    products = [
        ProductCreateParams(
            name=f"Product {i}",
            brand="BatchBrand",
            category="apparel"
        )
        for i in range(1000)
    ]
    
    results = await batch.create_products(products)
    
    print(f"Created: {results.successful}")
    print(f"Failed: {results.failed}")
    
    for error in results.errors:
        print(f"  Error on item {error.index}: {error.message}")

# Or use the streaming batch processor
async for result in client.batch_stream(products):
    if result.success:
        print(f"Created: {result.product.id}")
    else:
        print(f"Failed: {result.error.message}")