Running Python Scripts

Real-World Script Examples

Data Processing Script

This script processes CSV data, a common task in data analysis:

#!/usr/bin/env python3
"""
Process sales data to generate a summary report.
"""
import csv
import argparse
from collections import defaultdict
from datetime import datetime

def parse_args():
    parser = argparse.ArgumentParser(description='Generate sales report from CSV data')
    parser.add_argument('input_file', help='Input CSV file path')
    parser.add_argument('output_file', help='Output report file path')
    parser.add_argument('--year', type=int, help='Filter by year')
    return parser.parse_args()

def process_sales_data(input_file, year_filter=None):
    sales_by_region = defaultdict(float)
    sales_by_product = defaultdict(float)
    total_sales = 0.0
    
    with open(input_file, 'r', newline='') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            # Parse the date
            date = datetime.strptime(row['date'], '%Y-%m-%d')
            
            # Apply year filter if specified
            if year_filter and date.year != year_filter:
                continue
                
            # Extract data
            region = row['region']
            product = row['product']
            amount = float(row['amount'])
            
            # Update our aggregations
            sales_by_region[region] += amount
            sales_by_product[product] += amount
            total_sales += amount
    
    return {
        'total_sales': total_sales,
        'sales_by_region': sales_by_region,
        'sales_by_product': sales_by_product,
    }

def write_report(output_file, data):
    with open(output_file, 'w') as f:
        f.write("SALES REPORT\n")
        f.write("=" * 40 + "\n\n")
        
        f.write(f"Total Sales: ${data['total_sales']:.2f}\n\n")
        
        f.write("Sales by Region:\n")
        for region, amount in sorted(data['sales_by_region'].items()):
            f.write(f"  {region}: ${amount:.2f}\n")
        f.write("\n")
        
        f.write("Sales by Product:\n")
        for product, amount in sorted(data['sales_by_product'].items()):
            f.write(f"  {product}: ${amount:.2f}\n")

def main():
    args = parse_args()
    data = process_sales_data(args.input_file, args.year)
    write_report(args.output_file, data)
    print(f"Report written to {args.output_file}")

if __name__ == "__main__":
    main()

Automation Script

This script automates a common development workflow:

#!/usr/bin/env python3
"""
Automate the process of updating code, running tests, and deploying if tests pass.
"""
import os
import subprocess
import argparse
import logging
import sys

def setup_logging(verbose=False):
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format='%(asctime)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )

def parse_args():
    parser = argparse.ArgumentParser(description='Automate code update and deployment')
    parser.add_argument('repo_dir', help='Repository directory')
    parser.add_argument('--branch', default='main', help='Branch to update')
    parser.add_argument('--deploy', action='store_true', help='Deploy if tests pass')
    parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
    return parser.parse_args()

def run_command(command, cwd=None):
    """Run a shell command and return its output and status."""
    logging.debug(f"Running command: {command}")
    try:
        result = subprocess.run(
            command,
            shell=True,
            cwd=cwd,
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        return True, result.stdout
    except subprocess.CalledProcessError as e:
        return False, e.stderr

def update_code(repo_dir, branch):
    """Pull the latest code from the repository."""
    logging.info(f"Updating code in {repo_dir} (branch: {branch})")
    
    # Ensure we're on the right branch
    success, output = run_command(f"git checkout {branch}", cwd=repo_dir)
    if not success:
        logging.error(f"Failed to checkout branch {branch}: {output}")
        return False
    
    # Pull the latest changes
    success, output = run_command("git pull", cwd=repo_dir)
    if not success:
        logging.error(f"Failed to pull latest changes: {output}")
        return False
    
    logging.info("Code updated successfully")
    return True

def run_tests(repo_dir):
    """Run the test suite."""
    logging.info("Running tests...")
    success, output = run_command("python -m pytest", cwd=repo_dir)
    if not success:
        logging.error(f"Tests failed: {output}")
        return False
    
    logging.info("All tests passed!")
    return True

def deploy(repo_dir):
    """Deploy the application."""
    logging.info("Deploying application...")
    success, output = run_command("./deploy.sh", cwd=repo_dir)
    if not success:
        logging.error(f"Deployment failed: {output}")
        return False
    
    logging.info("Deployment successful!")
    return True

def main():
    args = parse_args()
    setup_logging(args.verbose)
    
    # Ensure the repository directory exists
    if not os.path.isdir(args.repo_dir):
        logging.error(f"Directory not found: {args.repo_dir}")
        return 1
    
    # Update the code
    if not update_code(args.repo_dir, args.branch):
        return 2
    
    # Run tests
    if not run_tests(args.repo_dir):
        return 3
    
    # Deploy if requested and tests passed
    if args.deploy:
        if not deploy(args.repo_dir):
            return 4
    else:
        logging.info("Skipping deployment (use --deploy to deploy)")
    
    logging.info("All tasks completed successfully")
    return 0

if __name__ == "__main__":
    sys.exit(main())

Web API Script

This script interacts with a web API and processes the results:

#!/usr/bin/env python3
"""
Fetch weather data from an API and display a forecast.
"""
import argparse
import requests
import json
import sys
from datetime import datetime

def parse_args():
    parser = argparse.ArgumentParser(description='Display weather forecast for a location')
    parser.add_argument('location', help='City name or postal code')
    parser.add_argument('--api-key', help='API key (or set WEATHER_API_KEY env var)')
    parser.add_argument('--days', type=int, default=3, help='Number of days to forecast')
    parser.add_argument('--output', choices=['text', 'json'], default='text', 
                        help='Output format')
    return parser.parse_args()

def get_api_key(args):
    """Get API key from args or environment variable."""
    import os
    if args.api_key:
        return args.api_key
    
    api_key = os.environ.get('WEATHER_API_KEY')
    if not api_key:
        sys.stderr.write("Error: API key not provided. Use --api-key or set WEATHER_API_KEY environment variable.\n")
        sys.exit(1)
    
    return api_key

def fetch_weather(location, api_key, days=3):
    """Fetch weather data from the API."""
    url = "https://api.example.com/weather"
    params = {
        'location': location,
        'days': days,
        'key': api_key
    }
    
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()  # Raise exception for 4XX/5XX responses
        return response.json()
    except requests.exceptions.RequestException as e:
        sys.stderr.write(f"Error fetching weather data: {e}\n")
        sys.exit(2)

def format_text_output(data):
    """Format weather data as human-readable text."""
    location = data['location']['name']
    country = data['location']['country']
    current = data['current']
    forecast = data['forecast']['forecastday']
    
    output = [
        f"Weather for {location}, {country}",
        f"Current: {current['temp_c']}°C, {current['condition']['text']}",
        "\nForecast:",
    ]
    
    for day in forecast:
        date = datetime.strptime(day['date'], '%Y-%m-%d').strftime('%a, %b %d')
        output.append(f"  {date}: {day['day']['avgtemp_c']}°C, {day['day']['condition']['text']}")
    
    return '\n'.join(output)

def main():
    args = parse_args()
    api_key = get_api_key(args)
    
    # Fetch the weather data
    data = fetch_weather(args.location, api_key, args.days)
    
    # Output the data in the requested format
    if args.output == 'json':
        print(json.dumps(data, indent=2))
    else:
        print(format_text_output(data))
    
    return 0

if __name__ == "__main__":
    sys.exit(main())