Python Core Development Specialist

Specialized in Python class design, type safety, error handling, and asynchronous programming.

When to Use This Skill

• Designing Python classes with dataclasses or regular classes
• Implementing type hints and type-safe code
• Creating custom exceptions and error handling
• Writing asynchronous code with async/await
• Implementing context managers
• Using Protocol for structural typing

Core Principles

• Type Safety First: Always use type hints for function signatures and class attributes
• Explicit is Better Than Implicit: Clear, readable code over clever tricks
• Fail Fast with Clear Errors: Raise exceptions early with descriptive messages
• Dataclasses for Data: Use dataclasses for data containers
• Protocol for Duck Typing: Use Protocol for structural subtyping
• Async for I/O: Use async/await for I/O-bound operations

Implementation Guidelines

Dataclass Pattern

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class User:
    """User data model."""

    id: int
    email: str
    name: str
    created_at: datetime = field(default_factory=datetime.now)
    is_active: bool = True
    metadata: dict = field(default_factory=dict)

    def __post_init__(self):
        """Validate data after initialization."""
        if not self.email or "@" not in self.email:
            raise ValueError(f"Invalid email: {self.email}")

@dataclass
class Order:
    """Order with validation."""

    id: int
    user_id: int
    items: list[str]
    total: float
    status: str = "pending"

    def __post_init__(self):
        """WHY: Ensure business rules are enforced at object creation."""
        if self.total < 0:
            raise ValueError("Total cannot be negative")
        if not self.items:
            raise ValueError("Order must have at least one item")

Type Hints and Protocol

from typing import Protocol, Optional, Sequence
from collections.abc import Iterator

# Protocol for structural typing
class Drawable(Protocol):
    """Protocol for objects that can be drawn."""

    def draw(self) -> str:
        """Return string representation."""
        ...

class Circle:
    """Circle implements Drawable protocol."""

    def __init__(self, radius: float):
        self.radius = radius

    def draw(self) -> str:
        return f"Circle(radius={self.radius})"

# Function using Protocol
def render(shapes: Sequence[Drawable]) -> list[str]:
    """Render multiple shapes.

    Args:
        shapes: Sequence of drawable objects

    Returns:
        List of rendered strings
    """
    return [shape.draw() for shape in shapes]

# Type hints for complex types
def process_items(
    items: Sequence[str],
    max_count: Optional[int] = None,
    *,  # Force keyword-only arguments after this
    case_sensitive: bool = True
) -> Iterator[str]:
    """Process items with optional filtering.

    Args:
        items: Sequence of items to process
        max_count: Optional maximum number of items to process
        case_sensitive: Whether to preserve case (keyword-only)

    Yields:
        Processed items
    """
    count = 0
    for item in items:
        if max_count is not None and count >= max_count:
            break

        processed = item if case_sensitive else item.lower()
        yield processed
        count += 1

Custom Exceptions and Error Handling

# Custom exception hierarchy
class ApplicationError(Exception):
    """Base exception for application errors."""

    def __init__(self, message: str, code: Optional[str] = None):
        super().__init__(message)
        self.message = message
        self.code = code

class ValidationError(ApplicationError):
    """Raised when validation fails."""

    def __init__(self, message: str, field: Optional[str] = None):
        super().__init__(message, code="VALIDATION_ERROR")
        self.field = field

class NotFoundError(ApplicationError):
    """Raised when resource is not found."""

    def __init__(self, resource: str, id: int | str):
        super().__init__(
            f"{resource} with id {id} not found",
            code="NOT_FOUND"
        )
        self.resource = resource
        self.id = id

# Usage with proper error handling
def create_user(email: str, name: str) -> User:
    """Create user with validation.

    Args:
        email: User email address
        name: User display name

    Returns:
        Created user object

    Raises:
        ValidationError: If email or name is invalid
    """
    # WHY: Validate early to fail fast with clear error
    if not email or "@" not in email:
        raise ValidationError("Invalid email format", field="email")

    if not name or len(name) < 2:
        raise ValidationError("Name must be at least 2 characters", field="name")

    return User(id=generate_id(), email=email, name=name)

# Error handling in caller
def register_user(email: str, name: str) -> dict:
    """Register new user with error handling."""
    try:
        user = create_user(email, name)
        return {"success": True, "user": user}
    except ValidationError as e:
        return {
            "success": False,
            "error": e.message,
            "field": e.field
        }
    except Exception as e:
        # WHY: Log unexpected errors for debugging
        logger.error(f"Unexpected error during registration: {e}")
        return {
            "success": False,
            "error": "An unexpected error occurred"
        }

Async/Await Pattern

import asyncio
from typing import Optional

# Async function
async def fetch_user(user_id: int) -> Optional[dict]:
    """Fetch user from database asynchronously.

    Args:
        user_id: User ID to fetch

    Returns:
        User data or None if not found
    """
    async with database.connection() as conn:
        result = await conn.fetchone(
            "SELECT * FROM users WHERE id = ?",
            (user_id,)
        )
        return dict(result) if result else None

# Async with multiple concurrent operations
async def fetch_user_with_orders(user_id: int) -> dict:
    """Fetch user and their orders concurrently.

    Args:
        user_id: User ID

    Returns:
        Dict with user and orders data
    """
    # WHY: Run queries concurrently to improve performance
    user_task = fetch_user(user_id)
    orders_task = fetch_orders(user_id)

    user, orders = await asyncio.gather(user_task, orders_task)

    if not user:
        raise NotFoundError("User", user_id)

    return {
        "user": user,
        "orders": orders
    }

# Async context manager
class DatabaseConnection:
    """Async database connection context manager."""

    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.conn: Optional[Connection] = None

    async def __aenter__(self) -> Connection:
        """Establish connection."""
        self.conn = await connect(self.connection_string)
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """WHY: Ensure connection is closed even on error."""
        if self.conn:
            await self.conn.close()

# Usage
async def query_users() -> list[dict]:
    """Query users using async context manager."""
    async with DatabaseConnection("postgresql://...") as conn:
        results = await conn.fetch("SELECT * FROM users")
        return [dict(row) for row in results]

Context Manager Pattern

from typing import Optional, TextIO
from pathlib import Path

# Context manager for file operations
class FileProcessor:
    """Process file with automatic cleanup."""

    def __init__(self, file_path: Path, mode: str = "r"):
        self.file_path = file_path
        self.mode = mode
        self.file: Optional[TextIO] = None

    def __enter__(self) -> TextIO:
        """Open file."""
        self.file = open(self.file_path, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        """WHY: Ensure file is closed even if error occurs."""
        if self.file:
            self.file.close()

# Usage
def process_log_file(file_path: Path) -> int:
    """Process log file and count lines."""
    with FileProcessor(file_path) as f:
        return sum(1 for _ in f)

Tools to Use

• Read: Read existing Python files
• Write: Create new Python modules
• Edit: Modify existing code
• Bash: Run Python code, tests, and linters

Bash Commands

# Run Python script
python script.py

# Type checking with mypy
mypy app/

# Linting and formatting with ruff (modern)
ruff check --fix .
ruff format .

# Traditional linting and formatting
black .
isort .
flake8 .

# Run tests
pytest tests/
pytest --cov=app tests/

Workflow

1. Understand Requirements: Clarify what needs to be implemented
2. Write Tests First: Use pytest-testing skill
3. Verify Tests Fail: Confirm tests fail correctly (Red)
4. Implement Code: Write implementation with type hints
5. Run Tests: Ensure tests pass (Green)
6. Run Type Checker: Validate with mypy
7. Run Linter/Formatter: Clean up code style
8. Refactor: Improve code quality
9. Commit: Create atomic commit

Related Skills

• pytest-testing: For writing unit and integration tests
• python-api-development: For FastAPI/Flask development
• pytest-api-testing: For testing API endpoints

Coding Standards

See Python Coding Standards

TDD Workflow

Follow Python TDD Workflow

Key Reminders

• Always use type hints for function signatures and class attributes
• Use dataclasses for data containers (avoid writing boilerplate)
• Create custom exceptions for domain-specific errors
• Fail fast with clear, descriptive error messages
• Use async/await for I/O-bound operations
• Use context managers for resource management
• Use Protocol for structural typing (duck typing with type safety)
• Write tests before implementation (TDD)
• Run mypy for type checking
• Use ruff or black+isort+flake8 for code quality
• Write comments explaining WHY, not WHAT