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: ```python # main.py def main(params): """Main automation entry point""" # Your automation logic return result if __name__ == "__main__": main() ``` ## Error Handling ### Always Handle Errors ```python 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 ```python import logging logger = logging.getLogger(__name__) logger.info("Processing started", extra={ "automation_id": automation_id, "input_size": len(data) }) ``` ## Security ### Never Hardcode Secrets ❌ **Bad:** ```python API_KEY = "sk_live_abc123" ``` ✅ **Good:** ```python import os API_KEY = os.environ["FREDDY_API_KEY"] ``` ### Validate All Inputs ```python 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: ```bash # .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 ```python # 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 ```python 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 ```python # 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: ```bash streamline run main.py --params '{"test": true}' ``` ### Use Mock Data ```python # 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 ```bash 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 ```bash # .gitignore .env *.env secrets/ config/local.json ``` ## Monitoring ### Log Important Events ```python logger.info("Automation started", extra={ "automation_id": automation_id, "trigger": "webhook" }) logger.info("Processing complete", extra={ "records_processed": count, "duration_ms": duration }) ``` ### Track Metrics ```python 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: ```bash 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 ```python 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 ```python 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 ```python 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 ```python 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 ```markdown # 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 ## Related Documentation - **[Getting Started](/docs/documentation/streamline/getting-started)** - Create your first automation - **[Project Structure](/docs/documentation/streamline/project-structure)** - Organize your code - **[Parameters](/docs/documentation/streamline/parameters)** - Configure automation parameters - **[GitHub Deployment](/docs/documentation/streamline/github-deployment)** - Deploy from GitHub - **[API Reference](/docs/api-reference/streamline/introduction)** - Complete API documentation