Claude Code for Python Automation: Complete Beginner's Guide 2026

Developer writes code → Runs script → Error occurs →
Reads traceback → Googles solution → Fixes code → Reruns →
Another error → Repeat...

Claude writes code → Executes automatically →
Reads traceback → Edits code → Reruns →
Iterates until it works → You watch it happen

# Install globally
npm install -g @anthropic-ai/claude-code

# Verify installation
claude-code --version

pip install anthropic-claude-code

# Option 1: Environment variable (recommended)
export ANTHROPIC_API_KEY='your-api-key-here'

# Add to ~/.bashrc or ~/.zshrc for persistence
echo 'export ANTHROPIC_API_KEY="your-key"' >> ~/.zshrc

# Option 2: Interactive setup
claude-code auth login

# Test Claude Code
claude-code --help

# Run test query
claude-code "print hello world in Python"

# Claude creates and runs:
print("Hello, World!")
# Output: Hello, World!

# Project: Data Processing Automation

## Overview
This project automates daily sales data processing, cleaning, and report generation.

## Tech Stack
- Python 3.10
- pandas for data manipulation
- requests for API calls
- openpyxl for Excel generation

## Code Style
- Use type hints for all functions
- Follow PEP 8 style guide
- Use descriptive variable names
- Add docstrings to all functions
- Prefer list comprehensions over loops where readable

## Project Structure

## Common Tasks
1. Data cleaning: Remove duplicates, handle missing values
2. Data transformation: Convert currencies, normalize dates
3. Report generation: Create Excel summary with charts
4. API integration: POST processed data to dashboard

## Dependencies
Install via: `pip install pandas requests openpyxl pytest`

## Testing
Run tests before committing: `pytest tests/`

## Deployment
Scripts run via cron job daily at 6 AM UTC

## Automation Workflow

### Daily Sales Processing
1. Fetch CSVs from `data/` directory
2. Clean data:
   - Remove rows where `amount < 0`
   - Fill missing `customer_name` with "Unknown"
   - Convert `date` to ISO format
3. Group by product category
4. Generate Excel report with:
   - Summary sheet (totals, averages)
   - Detailed sheet (all records)
   - Charts sheet (bar chart of sales by category)
5. Save to `reports/sales-YYYY-MM-DD.xlsx`
6. POST summary to API endpoint: `https://api.example.com/sales`

## Error Handling

- Always use try-except blocks for file operations
- Log all errors to `logs/errors.log`
- Send email notification on critical failures
- Retry API calls 3 times with exponential backoff
- Validate data before processing (check dtypes, nulls, ranges)

## Performance

- Use pandas vectorized operations (avoid iterrows)
- Process large files in chunks (chunksize=10000)
- Cache API responses for 1 hour
- Use multiprocessing for independent tasks
- Profile scripts with cProfile before optimization

## Security

- Never hardcode API keys (use environment variables)
- Store credentials in `.env` file (add to .gitignore)
- Validate all user inputs
- Use parameterized queries for database operations
- Encrypt sensitive data before storing

Create a Python script that:

1. Reads all CSV files from the 'data/' directory
2. For each CSV:
   - Load using pandas
   - Remove duplicate rows
   - Fill missing 'price' values with column median
   - Convert 'date' column to datetime format
   - Calculate 'revenue' = quantity * price
3. Combine all data into one DataFrame
4. Create Excel file with two sheets:
   - 'Summary': Total sales, average price, top 10 products
   - 'Details': All cleaned records
5. Save to 'reports/sales-report-YYYY-MM-DD.xlsx'

Requirements:
- Use pandas and openpyxl
- Add error handling for missing files
- Log processing status
- Generate report filename with current date

$ claude-code "Execute the data processing task"

[Claude Code working...]

✓ Created data_processor.py
✓ Installed dependencies (pandas, openpyxl)
✓ Wrote CSV reading logic
✓ Executed script
✗ Error: FileNotFoundError: 'data/' directory not found

[Claude Code auto-fixing...]

✓ Added directory existence check
✓ Created 'data/' directory if missing
✓ Reran script
✓ Success! Generated reports/sales-report-2026-06-09.xlsx

Script complete. Processing time: 2.3 seconds
Processed 1,247 records from 3 CSV files

import pandas as pd
import os
from datetime import datetime
from pathlib import Path
import logging

# Setup logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

def ensure_directories():
    """Create necessary directories if they don't exist"""
    Path('data').mkdir(exist_ok=True)
    Path('reports').mkdir(exist_ok=True)
    logging.info("Directories verified")

def process_csv_files():
    """Process all CSV files in data directory"""
    data_dir = Path('data')
    csv_files = list(data_dir.glob('*.csv'))

    if not csv_files:
        logging.warning("No CSV files found in data/ directory")
        return

    all_data = []

    for csv_file in csv_files:
        logging.info(f"Processing {csv_file.name}")

        # Read CSV
        df = pd.read_csv(csv_file)

        # Remove duplicates
        df = df.drop_duplicates()

        # Fill missing prices with median
        if 'price' in df.columns:
            median_price = df['price'].median()
            df['price'] = df['price'].fillna(median_price)

        # Convert date column
        if 'date' in df.columns:
            df['date'] = pd.to_datetime(df['date'])

        # Calculate revenue
        if 'quantity' in df.columns and 'price' in df.columns:
            df['revenue'] = df['quantity'] * df['price']

        all_data.append(df)

    # Combine all data
    combined_df = pd.concat(all_data, ignore_index=True)
    logging.info(f"Combined {len(combined_df)} total records")

    return combined_df

def generate_excel_report(df):
    """Generate Excel report with summary and details"""
    today = datetime.now().strftime('%Y-%m-%d')
    filename = f'reports/sales-report-{today}.xlsx'

    with pd.ExcelWriter(filename, engine='openpyxl') as writer:
        # Summary sheet
        summary = {
            'Total Sales': df['revenue'].sum() if 'revenue' in df.columns else 0,
            'Average Price': df['price'].mean() if 'price' in df.columns else 0,
            'Total Records': len(df)
        }
        summary_df = pd.DataFrame([summary])
        summary_df.to_excel(writer, sheet_name='Summary', index=False)

        # Top 10 products (if product column exists)
        if 'product' in df.columns and 'revenue' in df.columns:
            top_products = (df.groupby('product')['revenue']
                           .sum()
                           .sort_values(ascending=False)
                           .head(10))
            top_products.to_excel(writer, sheet_name='Top Products')

        # Details sheet
        df.to_excel(writer, sheet_name='Details', index=False)

    logging.info(f"Report generated: {filename}")
    return filename

def main():
    try:
        ensure_directories()
        df = process_csv_files()

        if df is not None and not df.empty:
            filename = generate_excel_report(df)
            print(f"✓ Success! Generated {filename}")
        else:
            print("✗ No data to process")

    except Exception as e:
        logging.error(f"Error: {e}")
        raise

if __name__ == "__main__":
    main()

Build a Python script that automates API data flow:

1. GET weather data from OpenWeatherMap API
   - Endpoint: https://api.openweathermap.org/data/2.5/forecast
   - For cities: New York, London, Tokyo
   - API key from environment variable

2. Process the data:
   - Extract: city, date, temperature, conditions
   - Convert temperature from Kelvin to Celsius
   - Format date as YYYY-MM-DD

3. POST processed data to our internal API
   - Endpoint: https://internal.example.com/api/weather
   - Include API authentication header

4. Error handling:
   - Retry failed requests 3 times
   - Log all API calls
   - Email notification if all retries fail

5. Run daily via cron job (6 AM UTC)

import requests
import os
from datetime import datetime
import time
import logging
import smtplib
from email.message import EmailMessage

class WeatherAutomation:
    def __init__(self):
        self.weather_api_key = os.getenv('OPENWEATHER_API_KEY')
        self.internal_api_key = os.getenv('INTERNAL_API_KEY')
        self.cities = ['New York', 'London', 'Tokyo']

    def fetch_weather(self, city, retries=3):
        """Fetch weather with retry logic"""
        url = "https://api.openweathermap.org/data/2.5/forecast"
        params = {'q': city, 'appid': self.weather_api_key}

        for attempt in range(retries):
            try:
                response = requests.get(url, params=params, timeout=10)
                response.raise_for_status()
                return response.json()
            except requests.exceptions.RequestException as e:
                logging.warning(f"Attempt {attempt+1} failed: {e}")
                if attempt < retries - 1:
                    time.sleep(2 ** attempt)  # Exponential backoff
                else:
                    raise

    def process_data(self, raw_data):
        """Transform API response"""
        processed = []
        for item in raw_data.get('list', []):
            processed.append({
                'city': raw_data['city']['name'],
                'date': datetime.fromtimestamp(item['dt']).strftime('%Y-%m-%d'),
                'temp_celsius': round(item['main']['temp'] - 273.15, 1),
                'conditions': item['weather'][0]['description']
            })
        return processed

    def post_to_internal_api(self, data):
        """Send data to internal system"""
        url = "https://internal.example.com/api/weather"
        headers = {'Authorization': f'Bearer {self.internal_api_key}'}

        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        logging.info(f"Posted {len(data)} records")

    def send_error_notification(self, error_message):
        """Email notification on failure"""
        # Email sending logic
        pass

    def run(self):
        """Main execution"""
        try:
            all_data = []
            for city in self.cities:
                raw_data = self.fetch_weather(city)
                processed_data = self.process_data(raw_data)
                all_data.extend(processed_data)

            self.post_to_internal_api(all_data)
            logging.info(f"✓ Successfully processed {len(all_data)} records")

        except Exception as e:
            logging.error(f"Critical error: {e}")
            self.send_error_notification(str(e))
            raise

if __name__ == "__main__":
    automation = WeatherAutomation()
    automation.run()

Create a web scraping script that:

1. Scrapes prices from 5 product pages
2. Stores data in SQLite database with timestamp
3. Compares with yesterday's prices
4. Generates alert if price drops > 10%
5. Sends Telegram notification with deal alerts

Use BeautifulSoup and requests.
Handle anti-scraping measures (delays, user-agent).

from bs4 import BeautifulSoup
import requests
import sqlite3
import time
from datetime import datetime
import random

class PriceScraper:
    def __init__(self):
        self.db_path = 'prices.db'
        self.setup_database()

    def setup_database(self):
        """Initialize SQLite database"""
        conn = sqlite3.connect(self.db_path)
        conn.execute('''
            CREATE TABLE IF NOT EXISTS prices (
                id INTEGER PRIMARY KEY,
                product_url TEXT,
                product_name TEXT,
                price REAL,
                timestamp DATETIME
            )
        ''')
        conn.commit()
        conn.close()

    def scrape_price(self, url):
        """Scrape single product with anti-detection"""
        headers = {
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
        }

        # Random delay (1-3 seconds)
        time.sleep(random.uniform(1, 3))

        response = requests.get(url, headers=headers)
        soup = BeautifulSoup(response.content, 'html.parser')

        # Extract price (adapt selectors to target site)
        price_element = soup.select_one('.price-current')
        price = float(price_element.text.strip().replace('$', ''))

        name_element = soup.select_one('.product-title')
        name = name_element.text.strip()

        return {'name': name, 'price': price, 'url': url}

    def check_price_drop(self, product_url, current_price):
        """Compare with previous price"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute('''
            SELECT price FROM prices
            WHERE product_url = ?
            ORDER BY timestamp DESC
            LIMIT 1
        ''', (product_url,))

        previous = cursor.fetchone()
        conn.close()

        if previous:
            old_price = previous[0]
            drop_percent = ((old_price - current_price) / old_price) * 100
            return drop_percent if drop_percent > 10 else 0
        return 0

    def send_telegram_alert(self, product_name, old_price, new_price, percent):
        """Send Telegram notification"""
        # Telegram bot API integration
        pass

# In CLAUDE.md
## Hooks

### pre-commit
- Run unit tests: `pytest tests/`
- Run linter: `flake8 *.py`
- Check type hints: `mypy *.py`
- If any fail, block commit

### post-edit
- Auto-format with Black: `black *.py`
- Sort imports: `isort *.py`
- Update docstrings: `pydocstyle *.py`

### on-error
- Log to file: append error to `logs/errors.log`
- Send Slack notification: curl webhook with error details
- Create GitHub issue (for recurring errors)

### scheduled (daily at 6 AM)
- Backup database: `python backup_db.py`
- Clean old logs: `find logs/ -mtime +30 -delete`
- Generate reports: `python generate_reports.py`

## Automation Hooks

### Pre-Commit Quality Checks
```bash
#!/bin/bash
echo "Running pre-commit checks..."

# Run tests
pytest tests/ || exit 1

# Check code style
flake8 *.py || exit 1

# Type checking
mypy *.py || exit 1

echo "✓ All checks passed"

black *.py
isort *.py
echo "✓ Code formatted"

# Add to crontab: crontab -e
0 6 * * * cd /path/to/project && python automate.py

hooks:
  pre-commit:
    - name: "Run Tests"
      command: "pytest tests/"
      blocking: true

    - name: "Lint Code"
      command: "flake8 *.py"
      blocking: true

  post-edit:
    - name: "Format Code"
      command: "black *.py"
      blocking: false

  on-error:
    - name: "Log Error"
      command: "echo '$(date): Error occurred' >> logs/errors.log"

  scheduled:
    - name: "Daily Backup"
      schedule: "0 6 * * *"  # Cron syntax
      command: "python backup.py"

"Build a complete data pipeline with ETL, machine learning,
API integration, real-time dashboards, and automated alerts"

Session 1: "Process a single CSV file"
Session 2: "Add data validation"
Session 3: "Generate simple report"
Session 4: "Add API integration"
Session 5: "Automate with scheduling"

"Fix the data thing"

"In data_processor.py, fix the function clean_sales_data()
to handle null values in the 'amount' column by:
1. Replacing nulls with 0 if 'status' is 'refund'
2. Replacing nulls with median if 'status' is 'sale'
3. Logging how many nulls were replaced"

## Recent Changes (Update Log)

### 2026-06-09
- Added API rate limiting (max 100 calls/hour)
- Changed database from SQLite to PostgreSQL
- New requirement: All prices must be in USD

### 2026-06-08
- Migrated from pandas 1.x to 2.x
- Updated data cleaning logic for new CSV format

# .gitignore
.env
*.pyc
__pycache__/
logs/
.claude/cache/

# README.md
## Running with Claude Code

1. Install dependencies: `pip install -r requirements.txt`
2. Set environment variables (see .env.example)
3. Run automation: `claude-code "run daily automation"`

# tests/test_data_processor.py
import pytest
from data_processor import clean_sales_data

def test_clean_sales_data_removes_duplicates():
    input_data = [
        {'id': 1, 'amount': 100},
        {'id': 1, 'amount': 100},  # duplicate
        {'id': 2, 'amount': 200}
    ]
    result = clean_sales_data(input_data)
    assert len(result) == 2

def test_clean_sales_data_handles_nulls():
    input_data = [{'id': 1, 'amount': None}]
    result = clean_sales_data(input_data)
    assert result[0]['amount'] == 0  # or median, depending on logic

claude-code "run pytest and fix any failing tests"

Error: ANTHROPIC_API_KEY environment variable not set

# Verify environment variable
echo $ANTHROPIC_API_KEY

# If empty, set it:
export ANTHROPIC_API_KEY='sk-ant-api...'

# Make permanent (add to ~/.zshrc or ~/.bashrc):
echo 'export ANTHROPIC_API_KEY="your-key"' >> ~/.zshrc
source ~/.zshrc

ModuleNotFoundError: No module named 'pandas'

# Claude Code should auto-install, but if not:
pip install pandas

# Or create requirements.txt and install all:
pip install -r requirements.txt

# Tell Claude about dependencies in CLAUDE.md:
## Dependencies
Install via: `pip install pandas requests openpyxl`

PermissionError: [Errno 13] Permission denied: 'output.xlsx'

# Check file permissions
ls -la output.xlsx

# Fix permissions
chmod 644 output.xlsx

# Or delete locked file
rm output.xlsx

# Tell Claude Code:
claude-code "fix the permission error by checking if file is locked before writing"

# Interrupt Claude Code:
Ctrl+C

# Restart with clearer instructions:
claude-code "In script.py, add a max_iterations=10 limit
to the while loop to prevent infinite loops"

Error: API rate limit exceeded

# Add rate limiting to your code
import time

def rate_limited_api_call(url):
    time.sleep(1)  # Wait 1 second between calls
    response = requests.get(url)
    return response

# Or use CLAUDE.md:
## API Guidelines
- Max 100 API calls per minute
- Add 1-second delay between calls
- Implement exponential backoff for failures

project/
├── CLAUDE.md
├── main.py
├── data_processor.py
├── api_client.py
├── config.py
├── requirements.txt
└── tests/
    ├── test_processor.py
    └── test_api.py

## File Structure

- `main.py`: Entry point, orchestrates workflow
- `data_processor.py`: Data cleaning and transformation
- `api_client.py`: API calls with retry logic
- `config.py`: Configuration and environment variables

## Workflow
1. main.py imports data_processor and api_client
2. data_processor handles all pandas operations
3. api_client manages external API interactions
4. Keep files < 300 lines for maintainability

# db_manager.py - Generated by Claude Code
import sqlite3
from contextlib import contextmanager

class DatabaseManager:
    def __init__(self, db_path='automation.db'):
        self.db_path = db_path
        self.setup_database()

    @contextmanager
    def get_connection(self):
        """Context manager for database connections"""
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            raise
        finally:
            conn.close()

    def setup_database(self):
        """Create tables if they don't exist"""
        with self.get_connection() as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS automation_runs (
                    id INTEGER PRIMARY KEY,
                    script_name TEXT,
                    status TEXT,
                    records_processed INTEGER,
                    execution_time REAL,
                    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
                )
            ''')

    def log_run(self, script_name, status, records, exec_time):
        """Log automation execution"""
        with self.get_connection() as conn:
            conn.execute('''
                INSERT INTO automation_runs
                (script_name, status, records_processed, execution_time)
                VALUES (?, ?, ?, ?)
            ''', (script_name, status, records, exec_time))

# config.py - Generated by Claude Code
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    # API Keys
    OPENWEATHER_API_KEY = os.getenv('OPENWEATHER_API_KEY')
    INTERNAL_API_KEY = os.getenv('INTERNAL_API_KEY')

    # Database
    DB_PATH = os.getenv('DB_PATH', 'automation.db')

    # Paths
    DATA_DIR = 'data/'
    REPORTS_DIR = 'reports/'
    LOGS_DIR = 'logs/'

    # API Settings
    API_TIMEOUT = 10
    API_RETRIES = 3
    API_BACKOFF_FACTOR = 2

    # Email Alerts
    SMTP_HOST = os.getenv('SMTP_HOST')
    SMTP_PORT = int(os.getenv('SMTP_PORT', 587))
    ALERT_EMAIL = os.getenv('ALERT_EMAIL')

    @classmethod
    def validate(cls):
        """Validate required config"""
        required = ['OPENWEATHER_API_KEY', 'INTERNAL_API_KEY']
        missing = [key for key in required if not getattr(cls, key)]

        if missing:
            raise ValueError(f"Missing config: {', '.join(missing)}")

# Usage in your scripts:
from config import Config
Config.validate()
api_key = Config.OPENWEATHER_API_KEY

$ claude-code "Build e-commerce analytics automation as described in requirements.md"

[Claude Code Process:]

✓ Read requirements.md
✓ Created project structure:
  - main.py (orchestrator)
  - shopify_client.py (API integration)
  - analytics_processor.py (metrics calculation)
  - report_generator.py (Excel creation)
  - email_sender.py (SMTP integration)
  - config.py (configuration)

✓ Installed dependencies: shopify-python-api, pandas, openpyxl
✓ Wrote shopify_client.py with pagination support
✓ Executed test run
✗ Error: ShopifyAPI authentication failed

[Auto-fixing...]
✓ Updated authentication to use private app credentials
✓ Reran test
✓ Successfully fetched 1,247 orders

✓ Wrote analytics_processor.py with metrics:
  - Total revenue
  - Average order value
  - Top 10 products
  - Sales by category
  - Customer acquisition trends

✓ Wrote report_generator.py with Excel formatting
✓ Wrote email_sender.py with HTML email template
✓ Created main.py to orchestrate the workflow

✓ Full test run successful
✓ Generated reports/analytics-2026-06-09.xlsx
✓ Sent email to 3 stakeholders

Automation complete. Total time: 8 minutes.

import logging
from datetime import datetime, timedelta
from shopify_client import ShopifyClient
from analytics_processor import AnalyticsProcessor
from report_generator import ReportGenerator
from email_sender import EmailSender
from config import Config

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('logs/automation.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def main():
    try:
        # Validate configuration
        Config.validate()
        logger.info("Configuration validated")

        # Initialize clients
        shopify = ShopifyClient(Config.SHOPIFY_API_KEY, Config.SHOPIFY_STORE)
        processor = AnalyticsProcessor()
        report_gen = ReportGenerator()
        email_sender = EmailSender(Config.SMTP_HOST, Config.SMTP_PORT)

        # Fetch orders (last 7 days)
        start_date = datetime.now() - timedelta(days=7)
        logger.info(f"Fetching orders since {start_date}")
        orders = shopify.fetch_orders(start_date)
        logger.info(f"Fetched {len(orders)} orders")

        # Process analytics
        logger.info("Calculating metrics")
        metrics = processor.calculate_metrics(orders)

        # Generate report
        today = datetime.now().strftime('%Y-%m-%d')
        report_path = f'reports/analytics-{today}.xlsx'
        logger.info(f"Generating report: {report_path}")
        report_gen.create_report(metrics, orders, report_path)

        # Send email
        logger.info("Sending email")
        email_sender.send_report(
            recipients=Config.STAKEHOLDER_EMAILS,
            report_path=report_path,
            metrics=metrics
        )

        logger.info("✓ Automation completed successfully")

    except Exception as e:
        logger.error(f"Automation failed: {e}", exc_info=True)
        # Send error notification
        email_sender.send_error_notification(str(e))
        raise

if __name__ == "__main__":
    main()

# Run daily at 8 AM
0 8 * * * cd /path/to/project && /usr/bin/python3 main.py >> logs/cron.log 2>&1