Skip to content
/
Best Practices
Last updated

Recommended patterns and practices for building reliable Streamline automations.

Project Structure

Organize Your Code

my-automation/
├── main.py              # Entry point
├── config/
│   ├── settings.json    # Configuration
│   └── secrets.env      # Environment variables
├── src/
│   ├── handlers/        # Event handlers
│   ├── utils/           # Utility functions
│   └── models/          # Data models
├── tests/
│   └── test_main.py     # Unit tests
└── requirements.txt     # Dependencies

Use Clear Entry Points

Always define a clear entry point for your automation:

# main.py
def main(params):
    """Main automation entry point"""
    # Your automation logic
    return result

if __name__ == "__main__":
    main()

Error Handling

Always Handle Errors

try:
    result = process_data(input_data)
except ValueError as e:
    logger.error(f"Invalid input: {e}")
    return {"success": False, "error": str(e)}
except Exception as e:
    logger.error(f"Unexpected error: {e}")
    raise

Use Structured Logging

import logging

logger = logging.getLogger(__name__)

logger.info("Processing started", extra={
    "automation_id": automation_id,
    "input_size": len(data)
})

Security

Never Hardcode Secrets

❌ Bad:

API_KEY = "sk_live_abc123"

✅ Good:

import os
API_KEY = os.environ["FREDDY_API_KEY"]

Validate All Inputs

def process_user_input(data):
    if not isinstance(data, dict):
        raise ValueError("Input must be a dictionary")
    
    required_fields = ["name", "email"]
    for field in required_fields:
        if field not in data:
            raise ValueError(f"Missing required field: {field}")
    
    return sanitize_data(data)

Use Environment Variables

Store sensitive configuration in environment variables:

# .env
FREDDY_API_KEY=your-api-key
DATABASE_URL=postgresql://...
WEBHOOK_SECRET=your-webhook-secret

Performance

Optimize for Speed

  • Cache frequently accessed data
  • Use batch operations when possible
  • Avoid unnecessary API calls
  • Implement pagination for large datasets
# Cache expensive operations
from functools import lru_cache

@lru_cache(maxsize=100)
def get_user_data(user_id):
    return api.fetch_user(user_id)

Handle Rate Limits

import time
from requests.exceptions import HTTPError

def api_call_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.json()
        except HTTPError as e:
            if e.response.status_code == 429:
                wait_time = 2 ** attempt
                logger.warning(f"Rate limited, waiting {wait_time}s")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("Max retries exceeded")

Testing

Write Unit Tests

# tests/test_main.py
import pytest
from main import process_data

def test_process_data_valid_input():
    result = process_data({"key": "value"})
    assert result["success"] is True

def test_process_data_invalid_input():
    with pytest.raises(ValueError):
        process_data(None)

Test Locally First

Always test your automation locally before deploying:

streamline run main.py --params '{"test": true}'

Use Mock Data

# tests/conftest.py
import pytest

@pytest.fixture
def mock_api_response():
    return {
        "id": "test_123",
        "status": "success",
        "data": {"result": "test"}
    }

Git Integration

Use Meaningful Commit Messages

git commit -m "Add error handling for API timeouts"
git commit -m "Fix: Resolve pagination issue in list endpoint"

Branch Strategy

  • main - Production-ready code
  • develop - Development branch
  • feature/* - Feature branches
  • hotfix/* - Emergency fixes

Keep Secrets Out of Git

# .gitignore
.env
*.env
secrets/
config/local.json

Monitoring

Log Important Events

logger.info("Automation started", extra={
    "automation_id": automation_id,
    "trigger": "webhook"
})

logger.info("Processing complete", extra={
    "records_processed": count,
    "duration_ms": duration
})

Track Metrics

import time

start_time = time.time()
result = process_data(input_data)
duration = (time.time() - start_time) * 1000

logger.info(f"Processing took {duration:.2f}ms")

Monitor Sync Logs

Regularly check sync logs for errors:

curl -H "X-API-Key: $FREDDY_API_KEY" \
  "https://api.aitronos.com/api/v1/streamline/automations/sauto_abc123/sync-logs?status=error"

Parameters

Use Type Hints

from typing import Dict, List, Optional

def process_items(
    items: List[Dict],
    filter_key: Optional[str] = None
) -> Dict:
    """Process a list of items with optional filtering"""
    pass

Provide Defaults

def main(params: Dict = None):
    params = params or {}
    
    # Set defaults
    batch_size = params.get("batch_size", 100)
    timeout = params.get("timeout", 30)
    
    # Your logic here

Validate Parameters

def validate_params(params: Dict) -> None:
    """Validate automation parameters"""
    if "api_key" not in params:
        raise ValueError("api_key is required")
    
    if params.get("batch_size", 0) < 1:
        raise ValueError("batch_size must be positive")

Documentation

Document Your Code

def process_webhook(payload: Dict) -> Dict:
    """
    Process incoming webhook payload.
    
    Args:
        payload: Webhook data containing event information
        
    Returns:
        Dict with processing results and status
        
    Raises:
        ValueError: If payload is invalid
        HTTPError: If API call fails
    """
    pass

Maintain a README

# My Automation

## Description
Brief description of what this automation does.

## Setup
1. Install dependencies: `pip install -r requirements.txt`
2. Configure environment: `cp .env.example .env`
3. Run locally: `streamline run`

## Parameters
- `batch_size` (int): Number of items to process per batch
- `timeout` (int): Request timeout in seconds

## Deployment
Deploy with: `streamline deploy`

Common Pitfalls

Avoid These Mistakes

❌ Hardcoding configuration values
❌ Ignoring error handling
❌ Not testing locally before deploying
❌ Committing secrets to Git
❌ Not validating input parameters
❌ Ignoring rate limits
❌ Not logging important events

✅ Use environment variables
✅ Implement comprehensive error handling
✅ Test thoroughly before deployment
✅ Use .gitignore for secrets
✅ Validate all inputs
✅ Implement retry logic
✅ Log all significant operations

Previous page
Next page