Python Programming

Python Programming for Data Engineering

Production-grade Python development for building scalable data pipelines, ETL systems, and data-intensive applications.

Quick Start

# Modern Python 3.12+ data engineering setup
from dataclasses import dataclass
from typing import Generator
from collections.abc import Iterator
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

@dataclass
class DataRecord:
    """Type-safe data container with validation."""
    id: int
    value: float
    category: str

    def __post_init__(self):
        if self.value < 0:
            raise ValueError(f"Value must be non-negative, got {self.value}")

def process_records(records: Iterator[dict]) -> Generator[DataRecord, None, None]:
    """Memory-efficient generator for processing large datasets."""
    for idx, record in enumerate(records):
        try:
            yield DataRecord(
                id=record['id'],
                value=float(record['value']),
                category=record.get('category', 'unknown')
            )
        except (KeyError, ValueError) as e:
            logger.warning(f"Skipping invalid record {idx}: {e}")
            continue

# Usage
if __name__ == "__main__":
    sample_data = [{"id": 1, "value": "100.5", "category": "A"}]
    for record in process_records(iter(sample_data)):
        logger.info(f"Processed: {record}")

Core Concepts

1. Type-Safe Data Structures (2024-2025 Standard)

from typing import TypedDict, NotRequired, Literal
from dataclasses import dataclass, field
from datetime import datetime

# TypedDict for JSON-like structures
class PipelineConfig(TypedDict):
    source: str
    destination: str
    batch_size: int
    retry_count: NotRequired[int]
    mode: Literal["batch", "streaming"]

# Dataclass for domain objects
@dataclass(frozen=True, slots=True)
class ETLJob:
    """Immutable, memory-efficient job definition."""
    job_id: str
    created_at: datetime = field(default_factory=datetime.utcnow)
    config: dict = field(default_factory=dict)

    def to_dict(self) -> dict:
        return {"job_id": self.job_id, "created_at": self.created_at.isoformat()}

2. Generator Patterns for Large Data

from typing import Generator, Iterable
import csv
from pathlib import Path

def read_csv_chunks(
    file_path: Path,
    chunk_size: int = 10000
) -> Generator[list[dict], None, None]:
    """
    Memory-efficient CSV reader using generators.
    Processes files of any size without loading into memory.
    """
    with open(file_path, 'r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        chunk = []
        for row in reader:
            chunk.append(row)
            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []
        if chunk:  # Don't forget the last chunk
            yield chunk

def transform_pipeline(
    records: Iterable[dict],
    transformers: list[callable]
) -> Generator[dict, None, None]:
    """Composable transformation pipeline."""
    for record in records:
        result = record
        for transform in transformers:
            result = transform(result)
            if result is None:
                break
        if result is not None:
            yield result

3. Async Programming for I/O-Bound Tasks

import asyncio
import aiohttp
from typing import AsyncGenerator
import logging

logger = logging.getLogger(__name__)

async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    max_retries: int = 3,
    backoff_factor: float = 2.0
) -> dict | None:
    """
    Fetch URL with exponential backoff retry logic.
    Production pattern for API data ingestion.
    """
    for attempt in range(max_retries):
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                resp.raise_for_status()
                return await resp.json()
        except aiohttp.ClientError as e:
            wait_time = backoff_factor ** attempt
            logger.warning(f"Attempt {attempt+1} failed for {url}: {e}. Retrying in {wait_time}s")
            await asyncio.sleep(wait_time)
    logger.error(f"All retries exhausted for {url}")
    return None

async def fetch_all_pages(
    base_url: str,
    page_count: int,
    concurrency_limit: int = 10
) -> AsyncGenerator[dict, None]:
    """Concurrent API fetching with rate limiting."""
    semaphore = asyncio.Semaphore(concurrency_limit)

    async def bounded_fetch(session: aiohttp.ClientSession, url: str):
        async with semaphore:
            return await fetch_with_retry(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, f"{base_url}?page={i}") for i in range(page_count)]
        for result in asyncio.as_completed(tasks):
            data = await result
            if data:
                yield data

4. Error Handling & Observability

import functools
import time
import logging
from typing import TypeVar, Callable, ParamSpec

P = ParamSpec('P')
R = TypeVar('R')

def with_retry(
    max_attempts: int = 3,
    exceptions: tuple = (Exception,),
    backoff_factor: float = 2.0
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Decorator for automatic retry with exponential backoff.
    Use for flaky operations (network, database connections).
    """
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    wait_time = backoff_factor ** attempt
                    logging.warning(
                        f"{func.__name__} attempt {attempt+1} failed: {e}. "
                        f"Retrying in {wait_time}s"
                    )
                    time.sleep(wait_time)
            raise last_exception
        return wrapper
    return decorator

def log_execution_time(func: Callable[P, R]) -> Callable[P, R]:
    """Decorator for performance monitoring."""
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        try:
            result = func(*args, **kwargs)
            duration = time.perf_counter() - start
            logging.info(f"{func.__name__} completed in {duration:.3f}s")
            return result
        except Exception as e:
            duration = time.perf_counter() - start
            logging.error(f"{func.__name__} failed after {duration:.3f}s: {e}")
            raise
    return wrapper

Tools & Technologies

Learning Path

Phase 1: Foundations (Weeks 1-3)

Week 1: Core syntax, data types, control flow
Week 2: Functions, modules, file I/O
Week 3: OOP (classes, inheritance, composition)

Phase 2: Intermediate (Weeks 4-6)

Week 4: Generators, iterators, decorators
Week 5: Type hints, dataclasses, protocols
Week 6: Error handling, logging, testing basics

Phase 3: Advanced (Weeks 7-9)

Week 7: Async/await, concurrent programming
Week 8: Memory optimization, profiling
Week 9: Package structure, dependency management

Phase 4: Production Mastery (Weeks 10-12)

Week 10: CI/CD integration, linting, formatting
Week 11: Performance optimization patterns
Week 12: Production deployment patterns

Production Patterns

Configuration Management

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """Type-safe configuration with environment variable support."""
    database_url: str
    api_key: str
    batch_size: int = 1000
    debug: bool = False

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

@lru_cache
def get_settings() -> Settings:
    """Cached settings singleton."""
    return Settings()

Connection Pooling

from contextlib import contextmanager
from typing import Generator
import psycopg2
from psycopg2 import pool

class DatabasePool:
    """Thread-safe connection pool for PostgreSQL."""

    def __init__(self, dsn: str, min_conn: int = 2, max_conn: int = 10):
        self._pool = pool.ThreadedConnectionPool(min_conn, max_conn, dsn)

    @contextmanager
    def get_connection(self) -> Generator:
        conn = self._pool.getconn()
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            self._pool.putconn(conn)

    def close(self):
        self._pool.closeall()

Troubleshooting Guide

Common Failure Modes

Debug Checklist

# 1. Check Python version
python --version  # Should be 3.12+

# 2. Verify virtual environment
which python  # Should point to venv

# 3. Check installed packages
uv pip list | grep <package>

# 4. Run with verbose logging
python -m mymodule -v 2>&1 | tee debug.log

# 5. Profile memory usage
python -m memory_profiler script.py

# 6. Profile CPU
python -m cProfile -s cumtime script.py

Log Interpretation

# Structured logging for easier debugging
import structlog

logger = structlog.get_logger()

def process_batch(batch_id: str, records: list):
    logger.info("batch_started", batch_id=batch_id, record_count=len(records))
    try:
        # processing...
        logger.info("batch_completed", batch_id=batch_id, success=True)
    except Exception as e:
        logger.error("batch_failed", batch_id=batch_id, error=str(e), exc_info=True)
        raise

Unit Test Template

import pytest
from unittest.mock import Mock, patch
from your_module import process_records, DataRecord

class TestProcessRecords:
    """Unit tests following AAA pattern (Arrange-Act-Assert)."""

    def test_valid_records_processed(self):
        # Arrange
        input_data = [{"id": 1, "value": "10.5", "category": "A"}]

        # Act
        result = list(process_records(iter(input_data)))

        # Assert
        assert len(result) == 1
        assert result[0].id == 1
        assert result[0].value == 10.5

    def test_invalid_records_skipped(self):
        # Arrange
        input_data = [{"id": 1}]  # Missing 'value'

        # Act
        result = list(process_records(iter(input_data)))

        # Assert
        assert len(result) == 0

    def test_negative_value_raises_error(self):
        # Arrange & Act & Assert
        with pytest.raises(ValueError, match="non-negative"):
            DataRecord(id=1, value=-5.0, category="A")

    @patch('your_module.external_api_call')
    def test_with_mocked_dependency(self, mock_api):
        # Arrange
        mock_api.return_value = {"status": "ok"}

        # Act
        result = function_using_api()

        # Assert
        mock_api.assert_called_once()
        assert result["status"] == "ok"

Best Practices

Code Style (2025 Standards)

# ✅ DO: Use type hints everywhere
def calculate_metrics(data: list[float]) -> dict[str, float]: ...

# ✅ DO: Prefer composition over inheritance
@dataclass
class Pipeline:
    reader: DataReader
    transformer: Transformer
    writer: DataWriter

# ✅ DO: Use context managers for resources
with open_connection() as conn:
    process(conn)

# ❌ DON'T: Use bare except
try: ...
except: pass  # Never do this

# ❌ DON'T: Mutate function arguments
def process(items: list) -> list:
    items.append("new")  # Avoid this
    return items.copy()  # Return new list instead

Performance Tips

# ✅ Use generators for large data
def process_large_file(path):
    with open(path) as f:
        for line in f:  # Memory efficient
            yield transform(line)

# ✅ Use set/dict for O(1) lookups
valid_ids = set(load_valid_ids())  # Not list
if item_id in valid_ids: ...

# ✅ Use local variables in hot loops
def hot_loop(items):
    local_func = expensive_lookup  # Cache reference
    for item in items:
        local_func(item)

Resources

Official Documentation

• Python Docs
• PEP Index
• Typing Module

Production References

• Real Python
• Python Patterns
• High Performance Python (Book)

Community

• Python Discord
• r/Python
• PyPI - Package Index

Next Skills

After mastering Python programming:

• → sql-databases - Query and manage relational data
• → etl-tools - Build data pipelines with Airflow
• → big-data - Scale with Spark and distributed systems
• → machine-learning - Apply ML with scikit-learn

---

Skill Certification Checklist:

• [ ] Can write type-safe Python with mypy validation
• [ ] Can implement generators for large data processing
• [ ] Can use async/await for concurrent I/O
• [ ] Can write comprehensive unit tests with pytest
• [ ] Can profile and optimize Python performance

Python Programming Guide

Overview

This guide provides comprehensive documentation for the python-programming skill in the custom-plugin-data-engineer plugin.

Category: General

Quick Start

Prerequisites

• Familiarity with general concepts
• Development environment set up
• Plugin installed and configured

Basic Usage

# Invoke the skill
claude "python-programming - [your task description]"

# Example
claude "python-programming - analyze the current implementation"

Core Concepts

Key Principles

1. Consistency - Follow established patterns 2. Clarity - Write readable, maintainable code 3. Quality - Validate before deployment

Best Practices

• Always validate input data
• Handle edge cases explicitly
• Document your decisions
• Write tests for critical paths

Common Tasks

Task 1: Basic Implementation

# Example implementation pattern
def implement_python_programming(input_data):
    """
    Implement python-programming functionality.

    Args:
        input_data: Input to process

    Returns:
        Processed result
    """
    # Validate input
    if not input_data:
        raise ValueError("Input required")

    # Process
    result = process(input_data)

    # Return
    return result

Task 2: Advanced Usage

For advanced scenarios, consider:

• Configuration customization via assets/config.yaml
• Validation using scripts/validate.py
• Integration with other skills

Troubleshooting

Common Issues

Related Resources

• SKILL.md - Skill specification
• config.yaml - Configuration options
• validate.py - Validation script

---

Last updated: 2025-12-30

Python Programming Patterns

Design Patterns

Pattern 1: Input Validation

Always validate input before processing:

def validate_input(data):
    if data is None:
        raise ValueError("Data cannot be None")
    if not isinstance(data, dict):
        raise TypeError("Data must be a dictionary")
    return True

Pattern 2: Error Handling

Use consistent error handling:

try:
    result = risky_operation()
except SpecificError as e:
    logger.error(f"Operation failed: {e}")
    handle_error(e)
except Exception as e:
    logger.exception("Unexpected error")
    raise

Pattern 3: Configuration Loading

Load and validate configuration:

import yaml

def load_config(config_path):
    with open(config_path) as f:
        config = yaml.safe_load(f)
    validate_config(config)
    return config

Anti-Patterns to Avoid

❌ Don't: Swallow Exceptions

# BAD
try:
    do_something()
except:
    pass

✅ Do: Handle Explicitly

# GOOD
try:
    do_something()
except SpecificError as e:
    logger.warning(f"Expected error: {e}")
    return default_value

Category-Specific Patterns: General

Recommended Approach

1. Start with the simplest implementation 2. Add complexity only when needed 3. Test each addition 4. Document decisions

Common Integration Points

• Configuration: assets/config.yaml
• Validation: scripts/validate.py
• Documentation: references/GUIDE.md

---

Pattern library for python-programming skill