Instruction file imported from gangeshgudmalwar-tal/assignment2-adv (.github/instructions/file-types/scripts.instructions.md). Copyright stays with the author.

Scripts Guidelines

Scope: Bash, Python, Node.js scripts for automation, utilities
Applies to: scripts/*.sh, scripts/*.py, scripts/*.js
Principles: Fail-fast, idempotent, well-documented
Last Updated: 2025-12-30

Script Structure

Bash Script Template

#!/bin/bash
# scripts/setup_database.sh
#
# Set up database for development environment.
#
# Usage: ./scripts/setup_database.sh [environment]
#
# Arguments:
#   environment: dev, test, or prod (default: dev)
#
# Environment Variables:
#   DATABASE_URL: PostgreSQL connection string
#   REDIS_URL: Redis connection string
#
# Exit Codes:
#   0: Success
#   1: General error
#   2: Invalid arguments
#   3: Dependencies missing

set -euo pipefail  # Exit on error, undefined vars, pipe failures

# Configuration
readonly SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
readonly PROJECT_ROOT="$(dirname "$SCRIPT_DIR")"
readonly DEFAULT_ENV="dev"

# Colors for output
readonly RED='\033[0;31m'
readonly GREEN='\033[0;32m'
readonly YELLOW='\033[1;33m'
readonly NC='\033[0m'  # No Color

# Logging functions
log_info() {
    echo -e "${GREEN}[INFO]${NC} $*" >&2
}

log_warn() {
    echo -e "${YELLOW}[WARN]${NC} $*" >&2
}

log_error() {
    echo -e "${RED}[ERROR]${NC} $*" >&2
}

# Cleanup function
cleanup() {
    local exit_code=$?
    if [[ $exit_code -ne 0 ]]; then
        log_error "Script failed with exit code $exit_code"
    fi
    # Cleanup code here
    exit $exit_code
}

trap cleanup EXIT

# Validation functions
validate_environment() {
    local env="$1"
    case "$env" in
        dev|test|prod) return 0 ;;
        *) return 1 ;;
    esac
}

check_dependencies() {
    local deps=("docker" "docker-compose")
    for dep in "${deps[@]}"; do
        if ! command -v "$dep" >/dev/null 2>&1; then
            log_error "Missing dependency: $dep"
            return 1
        fi
    done
}

# Main functions
setup_database() {
    local env="$1"
    log_info "Setting up database for $env environment"
    
    # Load environment variables
    local env_file="$PROJECT_ROOT/.env.$env"
    if [[ -f "$env_file" ]]; then
        set -a
        source "$env_file"
        set +a
    fi
    
    # Validate required variables
    if [[ -z "${DATABASE_URL:-}" ]]; then
        log_error "DATABASE_URL environment variable not set"
        return 1
    fi
    
    # Start database services
    log_info "Starting database services..."
    docker-compose -f "$PROJECT_ROOT/docker-compose.yml" up -d postgres redis
    
    # Wait for services to be ready
    log_info "Waiting for database to be ready..."
    local max_attempts=30
    local attempt=1
    while [[ $attempt -le $max_attempts ]]; do
        if docker-compose -f "$PROJECT_ROOT/docker-compose.yml" exec -T postgres \
            pg_isready -U postgres >/dev/null 2>&1; then
            log_info "Database is ready"
            break
        fi
        
        log_info "Waiting for database... (attempt $attempt/$max_attempts)"
        sleep 2
        ((attempt++))
    done
    
    if [[ $attempt -gt $max_attempts ]]; then
        log_error "Database failed to start"
        return 1
    fi
    
    # Run migrations
    log_info "Running database migrations..."
    if ! alembic upgrade head; then
        log_error "Failed to run migrations"
        return 1
    fi
    
    # Seed initial data
    if [[ "$env" == "dev" ]]; then
        log_info "Seeding development data..."
        python -m scripts.seed_database
    fi
    
    log_info "Database setup completed successfully"
}

# Main script
main() {
    local environment="${1:-$DEFAULT_ENV}"
    
    # Validate arguments
    if ! validate_environment "$environment"; then
        log_error "Invalid environment: $environment. Must be dev, test, or prod."
        echo "Usage: $0 [environment]" >&2
        exit 2
    fi
    
    # Check dependencies
    if ! check_dependencies; then
        exit 3
    fi
    
    # Change to project root
    cd "$PROJECT_ROOT"
    
    # Run setup
    if setup_database "$environment"; then
        log_info "Setup completed successfully"
        exit 0
    else
        log_error "Setup failed"
        exit 1
    fi
}

# Run main function with all arguments
main "$@"

Python Script Template

#!/usr/bin/env python3
# scripts/seed_database.py
"""
Seed database with initial data for development.

Usage:
    python -m scripts.seed_database [--environment ENV] [--reset]

Arguments:
    --environment ENV: dev, test, or prod (default: dev)
    --reset: Drop all data before seeding

Environment Variables:
    DATABASE_URL: PostgreSQL connection string

Exit Codes:
    0: Success
    1: Error
    2: Invalid arguments
"""

import argparse
import asyncio
import os
import sys
from pathlib import Path

# Add project root to path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from src.app.modules.restaurants.models import Restaurant
from src.app.modules.users.models import User
from src.app.shared.db.base import Base


class DatabaseSeeder:
    """Database seeder for development data."""
    
    def __init__(self, database_url: str):
        self.database_url = database_url
        self.engine = create_async_engine(database_url, echo=False)
        self.async_session = sessionmaker(
            self.engine, class_=AsyncSession, expire_on_commit=False
        )
    
    async def reset_database(self):
        """Drop all tables and recreate."""
        print("Resetting database...")
        async with self.engine.begin() as conn:
            await conn.run_sync(Base.metadata.drop_all)
            await conn.run_sync(Base.metadata.create_all)
        print("Database reset complete")
    
    async def seed_restaurants(self, session: AsyncSession):
        """Seed restaurant data."""
        restaurants = [
            Restaurant(
                name="Mario's Pizza",
                cuisine="Italian",
                location="Mumbai Central",
                is_open=True
            ),
            Restaurant(
                name="Spice Garden",
                cuisine="Indian",
                location="Pune Station",
                is_open=True
            ),
            Restaurant(
                name="Dragon Palace",
                cuisine="Chinese",
                location="Nagpur",
                is_open=False
            ),
        ]
        
        for restaurant in restaurants:
            session.add(restaurant)
        
        await session.commit()
        print(f"Seeded {len(restaurants)} restaurants")
    
    async def seed_users(self, session: AsyncSession):
        """Seed user data."""
        from src.app.modules.users.auth import hash_password
        
        users = [
            User(
                email="customer@example.com",
                password_hash=hash_password("password123"),
                role="CUSTOMER"
            ),
            User(
                email="restaurant@example.com",
                password_hash=hash_password("password123"),
                role="RESTAURANT"
            ),
            User(
                email="driver@example.com",
                password_hash=hash_password("password123"),
                role="DRIVER"
            ),
        ]
        
        for user in users:
            session.add(user)
        
        await session.commit()
        print(f"Seeded {len(users)} users")
    
    async def run(self, reset: bool = False):
        """Run the seeding process."""
        try:
            if reset:
                await self.reset_database()
            
            async with self.async_session() as session:
                await self.seed_restaurants(session)
                await self.seed_users(session)
            
            print("Seeding completed successfully")
            
        except Exception as e:
            print(f"Seeding failed: {e}", file=sys.stderr)
            raise


def main():
    parser = argparse.ArgumentParser(description="Seed database with initial data")
    parser.add_argument(
        "--environment", "-e",
        choices=["dev", "test", "prod"],
        default="dev",
        help="Environment to seed"
    )
    parser.add_argument(
        "--reset",
        action="store_true",
        help="Drop all data before seeding"
    )
    
    args = parser.parse_args()
    
    # Get database URL
    env_file = project_root / f".env.{args.environment}"
    if env_file.exists():
        import dotenv
        dotenv.load_dotenv(env_file)
    
    database_url = os.environ.get("DATABASE_URL")
    if not database_url:
        print("DATABASE_URL environment variable not set", file=sys.stderr)
        sys.exit(1)
    
    # Run seeder
    seeder = DatabaseSeeder(database_url)
    asyncio.run(seeder.run(reset=args.reset))


if __name__ == "__main__":
    main()

Error Handling & Logging

Robust Error Handling

#!/bin/bash
# scripts/deploy.sh

deploy_service() {
    local service="$1"
    local image_tag="$2"
    
    log_info "Deploying $service with tag $image_tag"
    
    # Validate inputs
    if [[ -z "$service" || -z "$image_tag" ]]; then
        log_error "Service name and image tag are required"
        return 1
    fi
    
    # Check if service exists
    if ! docker-compose config --services | grep -q "^${service}$"; then
        log_error "Service '$service' not found in docker-compose.yml"
        return 1
    fi
    
    # Update image tag
    if ! sed -i.bak "s|${service}:.*|${service}:${image_tag}|" docker-compose.yml; then
        log_error "Failed to update image tag in docker-compose.yml"
        return 1
    fi
    
    # Deploy with rollback on failure
    if docker-compose up -d "$service"; then
        log_info "Successfully deployed $service"
        
        # Wait for health check
        if ! wait_for_service "$service" 60; then
            log_error "Service failed health check, rolling back"
            rollback_deployment "$service"
            return 1
        fi
    else
        log_error "Failed to deploy $service"
        rollback_deployment "$service"
        return 1
    fi
}

wait_for_service() {
    local service="$1"
    local timeout="$2"
    local start_time=$(date +%s)
    
    while (( $(date +%s) - start_time < timeout )); do
        if docker-compose exec -T "$service" health_check_command >/dev/null 2>&1; then
            return 0
        fi
        sleep 5
    done
    
    return 1
}

rollback_deployment() {
    local service="$1"
    log_warn "Rolling back deployment of $service"
    
    # Restore backup
    if [[ -f "docker-compose.yml.bak" ]]; then
        mv docker-compose.yml.bak docker-compose.yml
        docker-compose up -d "$service"
    fi
}

Structured Logging

# scripts/monitor_system.py

import logging
import json
import sys
from datetime import datetime

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler('logs/monitor.log')
    ]
)

logger = logging.getLogger(__name__)

class StructuredLogger:
    """Logger that outputs JSON for better parsing."""
    
    @staticmethod
    def log(level: str, message: str, **extra):
        """Log structured message."""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": level.upper(),
            "message": message,
            "script": "monitor_system",
            **extra
        }
        
        # Log to structured logger
        getattr(logger, level.lower())(json.dumps(log_entry))

# Usage
log = StructuredLogger()

def check_service_health(service_name: str) -> bool:
    """Check if service is healthy."""
    try:
        # Health check logic
        is_healthy = perform_health_check(service_name)
        
        if is_healthy:
            log.log("info", f"Service {service_name} is healthy")
        else:
            log.log("warning", f"Service {service_name} is unhealthy", 
                    service=service_name, status="unhealthy")
        
        return is_healthy
        
    except Exception as e:
        log.log("error", f"Failed to check {service_name} health",
                service=service_name, error=str(e))
        return False

Idempotency & Safety

Idempotent Operations

#!/bin/bash
# scripts/create_backup.sh

create_backup() {
    local backup_name="$1"
    local backup_dir="/backups"
    
    # Create backup directory if it doesn't exist
    mkdir -p "$backup_dir"
    
    # Check if backup already exists
    if [[ -f "${backup_dir}/${backup_name}.tar.gz" ]]; then
        log_warn "Backup ${backup_name} already exists, skipping"
        return 0
    fi
    
    log_info "Creating backup: ${backup_name}"
    
    # Create backup
    if tar -czf "${backup_dir}/${backup_name}.tar.gz" \
        --exclude='*.log' \
        --exclude='node_modules' \
        --exclude='.git' \
        .; then
        
        log_info "Backup created successfully: ${backup_name}.tar.gz"
        
        # Verify backup
        if ! tar -tzf "${backup_dir}/${backup_name}.tar.gz" >/dev/null; then
            log_error "Backup verification failed"
            rm -f "${backup_dir}/${backup_name}.tar.gz"
            return 1
        fi
        
        # Cleanup old backups (keep last 7)
        find "$backup_dir" -name "*.tar.gz" -mtime +7 -delete
        
        return 0
    else
        log_error "Failed to create backup"
        return 1
    fi
}

Atomic Operations

# scripts/migrate_data.py

import tempfile
import shutil
import os

class AtomicFileWriter:
    """Write files atomically to prevent corruption."""
    
    def __init__(self, target_path: str):
        self.target_path = target_path
        self.temp_file = None
    
    def __enter__(self):
        self.temp_file = tempfile.NamedTemporaryFile(
            mode='w',
            dir=os.path.dirname(self.target_path),
            delete=False,
            suffix='.tmp'
        )
        return self.temp_file
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.temp_file:
            temp_path = self.temp_file.name
            self.temp_file.close()
            
            if exc_type is None:
                # Success: atomically move temp file to target
                os.rename(temp_path, self.target_path)
            else:
                # Failure: clean up temp file
                os.unlink(temp_path)

# Usage
def update_config(new_config: dict):
    """Atomically update configuration file."""
    config_path = "config.json"
    
    with AtomicFileWriter(config_path) as f:
        json.dump(new_config, f, indent=2)
    
    print("Configuration updated atomically")

Configuration Management

Environment-Based Configuration

#!/bin/bash
# scripts/run_tests.sh

# Load environment-specific configuration
load_config() {
    local environment="${1:-dev}"
    local config_file="config/${environment}.env"
    
    if [[ ! -f "$config_file" ]]; then
        log_error "Configuration file not found: $config_file"
        return 1
    fi
    
    # Export all variables from config file
    set -a
    source "$config_file"
    set +a
    
    log_info "Loaded configuration from $config_file"
}

# Validate configuration
validate_config() {
    local required_vars=("DATABASE_URL" "REDIS_URL" "API_KEY")
    
    for var in "${required_vars[@]}"; do
        if [[ -z "${!var:-}" ]]; then
            log_error "Required configuration variable not set: $var"
            return 1
        fi
    done
}

# Main script
main() {
    local environment="${1:-dev}"
    
    load_config "$environment" || exit 1
    validate_config || exit 1
    
    log_info "Running tests for $environment environment"
    
    # Run tests with configuration
    pytest tests/ \
        --environment="$environment" \
        --cov=src/ \
        --cov-report=html \
        --junitxml="reports/test-results.xml"
}

Command-Line Argument Parsing

# scripts/bulk_import.py

import argparse
import csv
import sys

def parse_arguments():
    """Parse command line arguments."""
    parser = argparse.ArgumentParser(
        description="Bulk import data from CSV file",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --file data.csv --table restaurants
  %(prog)s --file users.csv --table users --batch-size 100
        """
    )
    
    parser.add_argument(
        "--file", "-f",
        required=True,
        help="CSV file to import"
    )
    
    parser.add_argument(
        "--table", "-t",
        required=True,
        choices=["restaurants", "users", "orders"],
        help="Target table for import"
    )
    
    parser.add_argument(
        "--batch-size", "-b",
        type=int,
        default=50,
        help="Number of records to import per batch"
    )
    
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Validate data without importing"
    )
    
    return parser.parse_args()

def validate_csv(file_path: str, table: str) -> bool:
    """Validate CSV file format."""
    required_columns = {
        "restaurants": ["name", "cuisine", "location"],
        "users": ["email", "role"],
        "orders": ["customer_id", "restaurant_id", "total_amount"]
    }
    
    try:
        with open(file_path, 'r', newline='') as f:
            reader = csv.DictReader(f)
            
            # Check required columns
            if not all(col in reader.fieldnames for col in required_columns[table]):
                print(f"Missing required columns for {table}", file=sys.stderr)
                return False
            
            # Validate first few rows
            for i, row in enumerate(reader):
                if i >= 5:  # Check first 5 rows
                    break
                
                # Table-specific validation
                if table == "users" and "@" not in row["email"]:
                    print(f"Invalid email format: {row['email']}", file=sys.stderr)
                    return False
        
        return True
        
    except Exception as e:
        print(f"Error validating CSV: {e}", file=sys.stderr)
        return False

def main():
    args = parse_arguments()
    
    if not validate_csv(args.file, args.table):
        sys.exit(1)
    
    if args.dry_run:
        print("Dry run: validation passed")
        return
    
    # Import logic here
    print(f"Importing {args.file} into {args.table} (batch size: {args.batch_size})")

if __name__ == "__main__":
    main()

Testing Scripts

Unit Tests for Scripts

# tests/test_scripts.py

import pytest
import subprocess
import tempfile
import os
from pathlib import Path

class TestSetupDatabaseScript:
    """Test the setup_database.sh script."""
    
    def test_script_exists_and_is_executable(self):
        """Test that the script exists and is executable."""
        script_path = Path("scripts/setup_database.sh")
        assert script_path.exists()
        assert os.access(script_path, os.X_OK)
    
    def test_script_has_shebang(self):
        """Test that the script has a proper shebang."""
        script_path = Path("scripts/setup_database.sh")
        with open(script_path) as f:
            first_line = f.readline().strip()
            assert first_line == "#!/bin/bash"
    
    def test_script_help_output(self):
        """Test that the script shows help information."""
        result = subprocess.run(
            ["./scripts/setup_database.sh", "--help"],
            capture_output=True,
            text=True,
            cwd=Path(__file__).parent.parent
        )
        
        assert result.returncode == 0
        assert "Usage:" in result.stdout
        assert "Set up database" in result.stdout

@pytest.mark.integration
def test_setup_database_integration(docker_compose):
    """Integration test for database setup script."""
    # This would require Docker Compose setup
    pass

Script Testing Helpers

#!/bin/bash
# scripts/test_script.sh

# Test helper functions
assert_equals() {
    local expected="$1"
    local actual="$2"
    local message="${3:-Assertion failed}"
    
    if [[ "$expected" != "$actual" ]]; then
        echo "FAIL: $message" >&2
        echo "Expected: $expected" >&2
        echo "Actual: $actual" >&2
        return 1
    fi
}

assert_file_exists() {
    local file="$1"
    
    if [[ ! -f "$file" ]]; then
        echo "FAIL: File does not exist: $file" >&2
        return 1
    fi
}

assert_command_succeeds() {
    local command="$1"
    
    if ! eval "$command"; then
        echo "FAIL: Command failed: $command" >&2
        return 1
    fi
}

# Example test
test_backup_creation() {
    echo "Testing backup creation..."
    
    # Create test directory
    local test_dir=$(mktemp -d)
    cd "$test_dir"
    
    # Create some test files
    echo "test content" > test.txt
    mkdir subdir
    echo "subdir content" > subdir/file.txt
    
    # Run backup script
    if ../scripts/create_backup.sh test_backup; then
        assert_file_exists "test_backup.tar.gz"
        assert_equals "test content" "$(tar -xzf test_backup.tar.gz -O test.txt)"
        echo "PASS: Backup creation test"
    else
        echo "FAIL: Backup creation test"
        return 1
    fi
}

Documentation & Help

Auto-Generated Help

# scripts/generate_docs.py

import argparse
import inspect
import sys
from pathlib import Path

def generate_script_help(script_path: Path) -> str:
    """Generate help documentation for a script."""
    
    # Read script content
    content = script_path.read_text()
    
    # Extract docstring
    lines = content.split('\n')
    docstring_lines = []
    in_docstring = False
    
    for line in lines:
        if line.strip().startswith('"""'):
            if not in_docstring:
                in_docstring = True
                continue
            else:
                break
        
        if in_docstring:
            docstring_lines.append(line)
    
    return '\n'.join(docstring_lines).strip()

def main():
    parser = argparse.ArgumentParser(description="Generate documentation for scripts")
    parser.add_argument("--output", "-o", default="SCRIPT_DOCS.md", help="Output file")
    
    args = parser.parse_args()
    
    scripts_dir = Path("scripts")
    docs = ["# Script Documentation\n"]
    
    for script_file in scripts_dir.glob("*.py"):
        if script_file.name.startswith('test_'):
            continue
            
        docs.append(f"## {script_file.name}\n")
        docs.append("```python")
        docs.append(generate_script_help(script_file))
        docs.append("```\n")
    
    Path(args.output).write_text('\n'.join(docs))
    print(f"Documentation generated: {args.output}")

if __name__ == "__main__":
    main()

Script Best Practices:

• Fail-fast: Use set -euo pipefail in Bash
• Idempotent: Scripts should be safe to run multiple times
• Atomic: Use temporary files and atomic moves
• Logged: All operations should be logged with timestamps
• Tested: Scripts should have unit and integration tests
• Documented: Clear usage instructions and examples

Common Script Patterns:

# Check if running as root
if [[ $EUID -eq 0 ]]; then
    log_error "This script should not be run as root"
    exit 1
fi

# Check if required tools are installed
command -v docker >/dev/null 2>&1 || { echo "Docker is required"; exit 1; }

# Create temporary files safely
temp_file=$(mktemp) || exit 1
trap "rm -f $temp_file" EXIT

# Progress indicators
show_progress() {
    local current=$1
    local total=$2
    local percent=$((current * 100 / total))
    echo -ne "Progress: $current/$total ($percent%)\\r"
}