Flow Myna

Best Practices

Follow these guidelines to get the most out of the Flow Myna API and ensure high-quality process mining data.

Naming Conventions

Event Names

Use clear, action-oriented names that describe what happened:

✅ Good

  • Order Placed
  • Payment Received
  • Invoice Sent
  • Case Assigned
  • Shipment Delivered

❌ Avoid

  • order_placed (use spaces)
  • PAYMENT (unclear)
  • event1 (not descriptive)
  • OrderPaymentProcessing (too long)
  • X (meaningless)

Object Types

Use singular nouns that represent business entities:

✅ Good

  • Order
  • Customer
  • Invoice
  • Support Ticket
  • Purchase Order

❌ Avoid

  • Orders (use singular)
  • CUST (unclear abbreviation)
  • obj_123 (not descriptive)
  • CustomerOrders (combine types)
  • data (too generic)

Object IDs

Use descriptive, unique identifiers that are easy to trace back to your source systems:

✅ Good

  • ORD-2024-00123
  • CUST-jane-doe-456
  • INV-Q1-2024-0001
  • TKT-SUPPORT-789

❌ Avoid

  • 1 (too simple)
  • abc (not unique)
  • order (not an ID)
  • 123456789012345678901234... (too long)

Consistency is Key

The most important rule: be consistent. Once you choose a naming pattern, stick with it.

Consistent Naming Example

# Define constants for event types
class Events:
    ORDER_CREATED = "Order Created"
    ORDER_PAID = "Order Paid"
    ORDER_SHIPPED = "Order Shipped"
    ORDER_DELIVERED = "Order Delivered"


class ObjectTypes:
    ORDER = "Order"
    CUSTOMER = "Customer"
    SHIPMENT = "Shipment"


# Use constants throughout your code
client.track(
    event=Events.ORDER_CREATED,
    objects=[
        {"type": ObjectTypes.ORDER, "id": order_id},
        {"type": ObjectTypes.CUSTOMER, "id": customer_id}
    ]
)

Case matters! "Order" and "order" are different object types. Always use consistent casing.

Timestamps

Always use ISO 8601 format with timezone information:

✅ Valid Formats

  • 2024-01-15T10:30:00Z
  • 2024-01-15T10:30:00.123Z
  • 2024-01-15T10:30:00+02:00
  • 2024-01-15T10:30:00-05:00

❌ Invalid Formats

  • Jan 15, 2024
  • 01/15/2024
  • 1705312200
  • 2024-01-15 10:30:00

Timestamp Handling

from datetime import datetime, timezone


# Always use UTC
timestamp = datetime.now(timezone.utc).isoformat()
# Result: "2024-01-15T10:30:00.123456+00:00"


# Or use the 'Z' suffix for UTC
timestamp = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
# Result: "2024-01-15T10:30:00Z"

Efficient Batching

For high-volume integrations, batch your requests:

✅ Efficient

1 request with 50 events

~1 API call

❌ Inefficient

50 requests with 1 event each

~50 API calls

Efficient Batch Processing

# Collect events in batches of 100
BATCH_SIZE = 100
events_batch = []


for record in source_data:
    events_batch.append({
        "event": record["event_type"],
        "timestamp": record["timestamp"],
        "objects": [{"type": "Case", "id": record["case_id"]}]
    })
    
    # Send when batch is full
    if len(events_batch) >= BATCH_SIZE:
        client.track_batch(events_batch)
        events_batch = []


# Don't forget remaining events
if events_batch:
    client.track_batch(events_batch)

Security Best Practices

  • Use environment variables — Never hardcode API keys
  • Use secrets managers — AWS Secrets Manager, HashiCorp Vault, etc.
  • Never commit keys to git — Add to .gitignore
  • Separate keys per environment — Dev, staging, production
  • Rotate keys periodically — Create new keys, revoke old ones
  • Revoke unused keys — Clean up keys that are no longer needed

Data Quality Tips

  • Include relevant properties — Add context that helps with analysis
  • Link related objects — Include all objects involved in an event
  • Use accurate timestamps — Historical accuracy matters for process mining
  • Validate before sending — Check data quality at the source
  • Handle duplicates — Implement idempotency in your integration

Data Validation Example

def validate_event(event_data):
    """Validate event data before sending to API"""
    errors = []
    
    # Check required fields
    if not event_data.get("event"):
        errors.append("Event name is required")
    elif len(event_data["event"]) > 200:
        errors.append("Event name must be <= 200 characters")
    
    # Check objects
    objects = event_data.get("objects", [])
    if not objects:
        errors.append("At least one object is required")
    
    for i, obj in enumerate(objects):
        if not obj.get("type"):
            errors.append(f"Object {i}: type is required")
        if not obj.get("id"):
            errors.append(f"Object {i}: id is required")
    
    # Check timestamp format if provided
    timestamp = event_data.get("timestamp")
    if timestamp:
        try:
            datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
        except ValueError:
            errors.append("Invalid timestamp format")
    
    if errors:
        raise ValueError(f"Validation errors: {errors}")
    
    return True