# 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: ```yaml 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: ```python """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: ```txt requests>=2.31.0 pandas>=2.0.0 python-dotenv>=1.0.0 ``` ## Recommended Structure 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** ```python """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** ```python """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** ```python """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** ```python """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** ```python """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 ```python 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 ```python 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 ```python 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 ```python import os from dotenv import load_dotenv load_dotenv() API_KEY = os.getenv("API_KEY") DATABASE_URL = os.getenv("DATABASE_URL") ``` ### 5. Add Logging ```python 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 ```python # 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 ```markdown # 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 - **[Parameters & Configuration](/docs/documentation/streamline/parameters)** - Configure automation parameters - **[GitHub Deployment](/docs/documentation/streamline/github-deployment)** - Deploy from GitHub - **[Best Practices](/docs/documentation/streamline/best-practices)** - Follow recommended patterns ## Support Need help with project structure? - **Email**: support@aitronos.com - **Documentation**: [Streamline Overview](/docs/documentation/streamline/overview)