Skip to content
/
Project Structure
Last updated

Streamline Project Structure

Learn how to structure your Streamline automation projects for optimal organization and maintainability.

Basic Project Structure

A minimal Streamline automation requires:

my-automation/
├── streamline.yaml      # Required: Configuration file
├── main.py             # Your automation code
└── requirements.txt    # Python dependencies

Required Files

streamline.yaml

The configuration file that defines your automation:

name: my-automation
description: Brief description of what this automation does
execution_file: main.py
parameters:
  - name: param1
    type: string
    required: false
    default: "default_value"
    description: "Description of parameter"

Required fields:

  • name: Automation name (alphanumeric, hyphens, underscores)
  • execution_file: Path to the main Python file
  • description: Brief description (optional but recommended)

Parameter types:

  • string: Text values
  • integer: Whole numbers
  • float: Decimal numbers
  • boolean: true/false values
  • array: List of values
  • object: JSON objects

Execution File (main.py)

Your automation's entry point:

"""My automation description."""

def main(**kwargs):
    """
    Main automation function.
    
    Args:
        **kwargs: Parameters defined in streamline.yaml
        
    Returns:
        dict: Result data
    """
    # Your automation logic here
    
    return {
        "success": True,
        "data": "result"
    }

if __name__ == "__main__":
    result = main()
    print(result)

requirements.txt

Python dependencies:

requests>=2.31.0
pandas>=2.0.0
python-dotenv>=1.0.0

For larger projects, use this structure:

my-automation/
├── streamline.yaml
├── main.py
├── requirements.txt
├── README.md
├── .gitignore
├── src/
│   ├── __init__.py
│   ├── api_client.py
│   ├── data_processor.py
│   └── utils.py
├── config/
│   ├── settings.py
│   └── constants.py
└── tests/
    ├── __init__.py
    ├── test_api_client.py
    └── test_data_processor.py

Example Implementation

main.py

"""Main automation entry point."""
from src.api_client import APIClient
from src.data_processor import DataProcessor
from config.settings import Settings

def main(api_key: str, data_source: str):
    """
    Process data from external API.
    
    Args:
        api_key: API key for authentication
        data_source: Data source identifier
    """
    # Initialize components
    settings = Settings()
    client = APIClient(api_key)
    processor = DataProcessor()
    
    # Fetch data
    raw_data = client.fetch_data(data_source)
    
    # Process data
    processed_data = processor.transform(raw_data)
    
    # Return results
    return {
        "success": True,
        "records_processed": len(processed_data),
        "data": processed_data
    }

if __name__ == "__main__":
    import sys
    result = main(
        api_key=sys.argv[1],
        data_source=sys.argv[2]
    )
    print(result)

src/api_client.py

"""API client for external service."""
import requests

class APIClient:
    """Client for interacting with external API."""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.example.com"
    
    def fetch_data(self, source: str) -> list:
        """Fetch data from API."""
        response = requests.get(
            f"{self.base_url}/data/{source}",
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        response.raise_for_status()
        return response.json()

src/data_processor.py

"""Data processing utilities."""

class DataProcessor:
    """Process and transform data."""
    
    def transform(self, raw_data: list) -> list:
        """Transform raw data into processed format."""
        processed = []
        for item in raw_data:
            processed.append({
                "id": item["id"],
                "value": item["value"] * 2,
                "processed": True
            })
        return processed

config/settings.py

"""Configuration settings."""

class Settings:
    """Application settings."""
    
    TIMEOUT = 30
    MAX_RETRIES = 3
    BATCH_SIZE = 100

Multi-Automation Repository

Host multiple automations in one repository:

my-automations/
├── README.md
├── automation-1/
│   ├── streamline.yaml
│   ├── main.py
│   ├── requirements.txt
│   └── src/
│       └── ...
├── automation-2/
│   ├── streamline.yaml
│   ├── main.py
│   ├── requirements.txt
│   └── src/
│       └── ...
└── shared/
    ├── __init__.py
    ├── common_utils.py
    └── api_clients.py

Using Shared Code

automation-1/main.py

"""Automation 1 using shared utilities."""
import sys
sys.path.append('..')  # Add parent directory to path

from shared.common_utils import format_date
from shared.api_clients import ExternalAPIClient

def main():
    client = ExternalAPIClient()
    date = format_date("2025-11-27")
    # ... automation logic

Best Practices

1. Use Type Hints

from typing import Dict, List, Optional

def process_data(
    items: List[Dict],
    filter_key: Optional[str] = None
) -> Dict[str, any]:
    """Process data with type hints."""
    pass

2. Add Docstrings

def fetch_data(source: str) -> list:
    """
    Fetch data from external source.
    
    Args:
        source: Data source identifier
        
    Returns:
        List of data records
        
    Raises:
        ValueError: If source is invalid
        ConnectionError: If API is unreachable
    """
    pass

3. Handle Errors Gracefully

def main(**kwargs):
    """Main automation with error handling."""
    try:
        # Automation logic
        result = process_data()
        return {"success": True, "data": result}
    except ValueError as e:
        return {"success": False, "error": f"Invalid input: {e}"}
    except Exception as e:
        return {"success": False, "error": f"Unexpected error: {e}"}

4. Use Environment Variables

import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("API_KEY")
DATABASE_URL = os.getenv("DATABASE_URL")

5. Add Logging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def main():
    logger.info("Starting automation")
    # ... logic
    logger.info("Automation completed successfully")

6. Write Tests

# tests/test_data_processor.py
import unittest
from src.data_processor import DataProcessor

class TestDataProcessor(unittest.TestCase):
    def setUp(self):
        self.processor = DataProcessor()
    
    def test_transform(self):
        raw_data = [{"id": 1, "value": 10}]
        result = self.processor.transform(raw_data)
        self.assertEqual(result[0]["value"], 20)

if __name__ == "__main__":
    unittest.main()

File Naming Conventions

  • Use snake_case for Python files: data_processor.py
  • Use kebab-case for directories: my-automation/
  • Use descriptive names: api_client.py not client.py

.gitignore

Always include a .gitignore:

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
*.egg
*.egg-info/
dist/
build/

# Virtual environments
venv/
.venv/
env/
ENV/

# IDE
.vscode/
.idea/
*.swp
*.swo

# Environment variables
.env
.env.local

# OS
.DS_Store
Thumbs.db

# Testing
.pytest_cache/
.coverage
htmlcov/

# Logs
*.log

README.md Template

# My Automation

Brief description of what this automation does.

## Features

- Feature 1
- Feature 2
- Feature 3

## Parameters

- `param1` (string, required): Description
- `param2` (integer, optional): Description (default: 10)

## Setup

\`\`\`bash
pip install -r requirements.txt
\`\`\`

## Usage

\`\`\`bash
python main.py --param1 value1 --param2 value2
\`\`\`

## Testing

\`\`\`bash
python -m pytest tests/
\`\`\`

## License

MIT

Next Steps

Support

Need help with project structure?

Previous page
Next page