Python SDK

from flowmyna import FlowMyna


# Initialize the client
client = FlowMyna(api_key="fm_live_your_key_here")


# Record an event
client.record_event(
    event="Order Placed",
    objects=[
        {"type": "Order", "id": "ORD-123"},
        {"type": "Customer", "id": "CUST-456"}
    ],
    properties={"total": 149.99}
)


# Upsert an object
client.upsert_object(
    type="Customer",
    id="CUST-456",
    properties={
        "name": "Jane Doe",
        "email": "jane@example.com"
    }
)

import os
from flowmyna import FlowMyna


# Using environment variable (recommended)
client = FlowMyna(api_key=os.environ["FLOWMYNA_API_KEY"])


# With custom configuration
client = FlowMyna(
    api_key="fm_live_xxx",
    timeout=60,          # 60 second timeout
    max_retries=5,       # Retry up to 5 times
)


# For testing/development
client = FlowMyna(
    api_key="fm_live_xxx",
    base_url="https://staging-api.flowmyna.com/api/public/v1"
)

from datetime import datetime, timezone


# Simple event
client.record_event(
    event="Order Created",
    objects=[{"type": "Order", "id": "ORD-123"}]
)


# With timestamp (datetime object)
client.record_event(
    event="Order Shipped",
    objects=[
        {"type": "Order", "id": "ORD-123"},
        {"type": "Shipment", "id": "SHIP-456"}
    ],
    timestamp=datetime.now(timezone.utc)
)


# With timestamp (ISO string)
client.record_event(
    event="Order Delivered",
    objects=[{"type": "Order", "id": "ORD-123"}],
    timestamp="2024-01-15T10:30:00Z"
)


# With properties
client.record_event(
    event="Payment Processed",
    objects=[
        {"type": "Order", "id": "ORD-123"},
        {"type": "Payment", "id": "PAY-789"}
    ],
    properties={
        "amount": 149.99,
        "currency": "USD",
        "method": "credit_card"
    }
)


# With object properties (merged on upsert)
client.record_event(
    event="Customer Login",
    objects=[{
        "type": "Customer",
        "id": "CUST-456",
        "properties": {
            "last_login": datetime.now(timezone.utc).isoformat(),
            "login_count": 42
        }
    }]
)

# Upsert a customer
result = client.upsert_object(
    type="Customer",
    id="CUST-456",
    properties={
        "name": "Jane Doe",
        "email": "jane@example.com",
        "tier": "gold",
        "signup_date": "2023-01-15"
    }
)


print(f"Object {'created' if result.created else 'updated'}")
print(f"Internal ID: {result.object_id}")


# Update properties (properties are merged)
client.upsert_object(
    type="Customer",
    id="CUST-456",
    properties={
        "lifetime_value": 5420.00,  # Added
        "tier": "platinum"          # Updated
    }
)


# Upsert without properties (just ensures object exists)
client.upsert_object(type="Order", id="ORD-123")

# Batch record events (up to 100 per call)
events = [
    {
        "event": "Case Opened",
        "timestamp": "2024-01-01T09:00:00Z",
        "objects": [{"type": "Case", "id": "CASE-001"}],
        "properties": {"priority": "high"}
    },
    {
        "event": "Case Assigned",
        "timestamp": "2024-01-01T09:15:00Z",
        "objects": [
            {"type": "Case", "id": "CASE-001"},
            {"type": "Agent", "id": "AGENT-A"}
        ]
    },
    {
        "event": "Case Resolved",
        "timestamp": "2024-01-01T14:30:00Z",
        "objects": [{"type": "Case", "id": "CASE-001"}]
    }
]


result = client.record_event_batch(events)
print(f"Processed: {result.processed}, Failed: {result.failed}")


# Batch upsert objects
customers = [
    {"type": "Customer", "id": f"CUST-{i:03d}", "properties": {"index": i}}
    for i in range(100)
]


result = client.upsert_object_batch(customers)
print(f"Upserted {result.processed} objects")

# Check API key and connection
health = client.health()


print(f"Status: {health.status}")
print(f"Workspace: {health.workspace_name}")
print(f"Dataset: {health.dataset_name}")
print(f"Key Name: {health.api_key_name}")


# Use in startup validation
def validate_flowmyna_connection():
    try:
        health = client.health()
        print(f"✓ Connected to Flow Myna")
        print(f"  Workspace: {health.workspace_name}")
        print(f"  Dataset: {health.dataset_name}")
        return True
    except Exception as e:
        print(f"✗ Flow Myna connection failed: {e}")
        return False

from flowmyna import FlowMyna
from flowmyna.exceptions import (
    FlowMynaError,
    AuthenticationError,
    ValidationError,
    RateLimitError,
    ServerError
)


client = FlowMyna(api_key="fm_live_xxx")


try:
    client.record_event(
        event="Order Placed",
        objects=[{"type": "Order", "id": "ORD-123"}]
    )
except AuthenticationError as e:
    # Invalid, revoked, or expired API key
    print(f"Authentication failed: {e}")
    
except ValidationError as e:
    # Invalid request data
    print(f"Validation error: {e}")
    
except RateLimitError as e:
    # Rate limited (429)
    print(f"Rate limited. Retry after: {e.retry_after}")
    
except ServerError as e:
    # Server error (5xx)
    print(f"Server error: {e}")
    
except FlowMynaError as e:
    # Base exception for all SDK errors
    print(f"Flow Myna error: {e}")

import asyncio
from flowmyna import AsyncFlowMyna


async def main():
    client = AsyncFlowMyna(api_key="fm_live_xxx")
    
    # Record event asynchronously
    await client.record_event(
        event="Order Placed",
        objects=[{"type": "Order", "id": "ORD-123"}]
    )
    
    # Batch operations
    events = [
        {"event": f"Event {i}", "objects": [{"type": "Test", "id": f"T-{i}"}]}
        for i in range(100)
    ]
    result = await client.record_event_batch(events)
    
    # Close the client when done
    await client.close()


asyncio.run(main())


# Or use as context manager
async def with_context_manager():
    async with AsyncFlowMyna(api_key="fm_live_xxx") as client:
        await client.record_event(
            event="Order Placed",
            objects=[{"type": "Order", "id": "ORD-123"}]
        )

import os
from datetime import datetime, timezone
from flowmyna import FlowMyna
from flowmyna.exceptions import FlowMynaError


# Initialize client
client = FlowMyna(api_key=os.environ["FLOWMYNA_API_KEY"])


# Validate connection on startup
health = client.health()
print(f"Connected to {health.workspace_name}")


def process_order(order):
    """Record an order through its lifecycle"""
    try:
        # Upsert customer with details
        client.upsert_object(
            type="Customer",
            id=order.customer_id,
            properties={
                "email": order.customer_email,
                "name": order.customer_name
            }
        )
        
        # Record order creation
        client.record_event(
            event="Order Created",
            objects=[
                {"type": "Order", "id": order.id, "properties": {
                    "total": order.total,
                    "currency": order.currency,
                    "items_count": len(order.items)
                }},
                {"type": "Customer", "id": order.customer_id}
            ],
            properties={"channel": order.channel}
        )
        
        # Record each item
        for item in order.items:
            client.record_event(
                event="Item Added",
                objects=[
                    {"type": "Order", "id": order.id},
                    {"type": "Product", "id": item.product_id}
                ],
                properties={
                    "quantity": item.quantity,
                    "price": item.price
                }
            )
        
    except FlowMynaError as e:
        print(f"Failed to record order {order.id}: {e}")


def import_historical_orders(orders):
    """Batch import historical orders"""
    events = []
    
    for order in orders:
        events.append({
            "event": "Order Created",
            "timestamp": order.created_at.isoformat(),
            "objects": [
                {"type": "Order", "id": order.id},
                {"type": "Customer", "id": order.customer_id}
            ]
        })
        
        # Send in batches of 100
        if len(events) >= 100:
            result = client.record_event_batch(events)
            print(f"Imported batch: {result.processed} events")
            events = []
    
    # Send remaining events
    if events:
        result = client.record_event_batch(events)
        print(f"Imported final batch: {result.processed} events")