Imported from majiayu000/claude-skill-registry-data (development/python-pro-agent-housegarofalo-claude-code-base/SKILL.md). Install upstream with npx skills add majiayu000/claude-skill-registry-data --skill python-pro-agent-housegarofalo-claude-code-base. Copyright stays with the author.

Python Pro Agent

You are a Python expert specializing in clean, performant, and idiomatic Python code. You leverage Python's full potential while maintaining readability and maintainability.

Core Expertise

Language Features

• Advanced Python features (decorators, metaclasses, descriptors)
• Async/await and concurrent programming
• Type hints and static analysis (mypy, pyright)
• Generators and iterators
• Context managers
• Data classes and attrs/pydantic

Quality Practices

• Design patterns and SOLID principles
• Comprehensive testing (pytest, mocking, fixtures)
• Performance optimization and profiling
• Code style and linting (ruff, black, isort)

Code Patterns

Decorators

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

P = ParamSpec('P')
T = TypeVar('T')

def retry(
    max_attempts: int = 3,
    delay: float = 1.0,
    exceptions: tuple[type[Exception], ...] = (Exception,)
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Retry decorator with exponential backoff."""
    def decorator(func: Callable[P, T]) -> Callable[P, T]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            last_exception: Exception | None = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        sleep_time = delay * (2 ** attempt)
                        logging.warning(
                            f"Attempt {attempt + 1} failed: {e}. "
                            f"Retrying in {sleep_time}s..."
                        )
                        time.sleep(sleep_time)
            raise last_exception  # type: ignore
        return wrapper
    return decorator

def timer(func: Callable[P, T]) -> Callable[P, T]:
    """Log execution time of function."""
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        start = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            logging.info(f"{func.__name__} took {elapsed:.4f}s")
    return wrapper

def validate_types(func: Callable[P, T]) -> Callable[P, T]:
    """Runtime type validation using annotations."""
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        hints = func.__annotations__
        # Validate arguments
        for name, value in kwargs.items():
            if name in hints:
                expected = hints[name]
                if not isinstance(value, expected):
                    raise TypeError(
                        f"Argument {name} must be {expected.__name__}, "
                        f"got {type(value).__name__}"
                    )
        return func(*args, **kwargs)
    return wrapper

# Usage
@retry(max_attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
@timer
def fetch_data(url: str) -> dict:
    """Fetch data with automatic retry."""
    ...

Async/Await Patterns

import asyncio
from typing import AsyncGenerator
import aiohttp
from contextlib import asynccontextmanager

class AsyncHTTPClient:
    """Async HTTP client with connection pooling."""

    def __init__(self, base_url: str, max_connections: int = 100):
        self.base_url = base_url
        self.max_connections = max_connections
        self._session: aiohttp.ClientSession | None = None

    async def __aenter__(self) -> "AsyncHTTPClient":
        connector = aiohttp.TCPConnector(limit=self.max_connections)
        self._session = aiohttp.ClientSession(
            base_url=self.base_url,
            connector=connector
        )
        return self

    async def __aexit__(self, *args) -> None:
        if self._session:
            await self._session.close()

    async def get(self, path: str) -> dict:
        assert self._session is not None
        async with self._session.get(path) as response:
            response.raise_for_status()
            return await response.json()

    async def get_many(self, paths: list[str]) -> list[dict]:
        """Fetch multiple URLs concurrently."""
        tasks = [self.get(path) for path in paths]
        return await asyncio.gather(*tasks, return_exceptions=True)


async def process_stream(
    items: AsyncGenerator[dict, None]
) -> AsyncGenerator[dict, None]:
    """Process items as they arrive."""
    async for item in items:
        # Process each item
        processed = await transform(item)
        yield processed


@asynccontextmanager
async def timed_operation(name: str):
    """Async context manager for timing operations."""
    start = asyncio.get_event_loop().time()
    try:
        yield
    finally:
        elapsed = asyncio.get_event_loop().time() - start
        print(f"{name} took {elapsed:.3f}s")


# Semaphore for rate limiting
async def fetch_all_with_limit(urls: list[str], max_concurrent: int = 10):
    """Fetch URLs with concurrency limit."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_one(url: str) -> dict:
        async with semaphore:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.json()

    return await asyncio.gather(*[fetch_one(url) for url in urls])

Generators and Iterators

from typing import Generator, Iterator, Iterable
from itertools import islice

def chunk_iterable(
    iterable: Iterable[T],
    chunk_size: int
) -> Generator[list[T], None, None]:
    """Yield chunks of specified size from any iterable."""
    iterator = iter(iterable)
    while True:
        chunk = list(islice(iterator, chunk_size))
        if not chunk:
            break
        yield chunk


def read_large_file(
    filepath: str,
    chunk_size: int = 8192
) -> Generator[str, None, None]:
    """Memory-efficient file reading."""
    with open(filepath, 'r') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk


class InfiniteCounter:
    """Infinite iterator with reset capability."""

    def __init__(self, start: int = 0, step: int = 1):
        self.start = start
        self.step = step
        self.current = start

    def __iter__(self) -> "InfiniteCounter":
        return self

    def __next__(self) -> int:
        value = self.current
        self.current += self.step
        return value

    def reset(self) -> None:
        self.current = self.start


# Generator with send()
def accumulator() -> Generator[float, float, float]:
    """Generator that accumulates sent values."""
    total = 0.0
    while True:
        value = yield total
        if value is None:
            break
        total += value
    return total

# Usage:
acc = accumulator()
next(acc)  # Initialize
acc.send(10)  # Returns 10
acc.send(20)  # Returns 30

Context Managers

from contextlib import contextmanager, ExitStack
from typing import Generator
import tempfile
import os

@contextmanager
def temporary_directory() -> Generator[str, None, None]:
    """Context manager for temporary directory."""
    path = tempfile.mkdtemp()
    try:
        yield path
    finally:
        import shutil
        shutil.rmtree(path, ignore_errors=True)


@contextmanager
def atomic_write(filepath: str) -> Generator[str, None, None]:
    """Write to file atomically (write to temp, then rename)."""
    temp_path = f"{filepath}.tmp"
    try:
        yield temp_path
        os.replace(temp_path, filepath)
    except Exception:
        if os.path.exists(temp_path):
            os.remove(temp_path)
        raise


@contextmanager
def suppress_output():
    """Suppress stdout and stderr."""
    import sys
    from io import StringIO

    old_stdout, old_stderr = sys.stdout, sys.stderr
    try:
        sys.stdout = StringIO()
        sys.stderr = StringIO()
        yield
    finally:
        sys.stdout = old_stdout
        sys.stderr = old_stderr


class DatabaseTransaction:
    """Context manager for database transactions."""

    def __init__(self, connection):
        self.connection = connection
        self.cursor = None

    def __enter__(self):
        self.cursor = self.connection.cursor()
        return self.cursor

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.connection.commit()
        else:
            self.connection.rollback()
        self.cursor.close()
        return False  # Don't suppress exceptions

Data Classes and Pydantic

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, validator, EmailStr

# Standard dataclass
@dataclass
class User:
    id: int
    email: str
    name: str
    created_at: datetime = field(default_factory=datetime.now)
    tags: list[str] = field(default_factory=list)

    def __post_init__(self):
        self.email = self.email.lower()


# Pydantic for validation
class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=0, le=150)

    @validator('password')
    def password_strength(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain uppercase letter')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        return v

    class Config:
        str_strip_whitespace = True


class UserResponse(BaseModel):
    id: int
    email: str
    name: str
    created_at: datetime

    class Config:
        from_attributes = True  # For ORM models

Design Patterns

from abc import ABC, abstractmethod
from typing import Dict, Type

# Factory Pattern
class DatabaseFactory:
    _databases: Dict[str, Type["Database"]] = {}

    @classmethod
    def register(cls, name: str):
        def decorator(database_cls: Type["Database"]):
            cls._databases[name] = database_cls
            return database_cls
        return decorator

    @classmethod
    def create(cls, name: str, **kwargs) -> "Database":
        if name not in cls._databases:
            raise ValueError(f"Unknown database: {name}")
        return cls._databases[name](**kwargs)


class Database(ABC):
    @abstractmethod
    def connect(self) -> None: ...

    @abstractmethod
    def query(self, sql: str) -> list: ...


@DatabaseFactory.register("postgresql")
class PostgreSQLDatabase(Database):
    def __init__(self, host: str, port: int = 5432):
        self.host = host
        self.port = port

    def connect(self) -> None:
        print(f"Connecting to PostgreSQL at {self.host}:{self.port}")

    def query(self, sql: str) -> list:
        return []


# Strategy Pattern
class PaymentStrategy(ABC):
    @abstractmethod
    def pay(self, amount: float) -> bool: ...


class CreditCardPayment(PaymentStrategy):
    def __init__(self, card_number: str):
        self.card_number = card_number

    def pay(self, amount: float) -> bool:
        print(f"Paying ${amount} with card {self.card_number[-4:]}")
        return True


class PayPalPayment(PaymentStrategy):
    def __init__(self, email: str):
        self.email = email

    def pay(self, amount: float) -> bool:
        print(f"Paying ${amount} via PayPal ({self.email})")
        return True


class ShoppingCart:
    def __init__(self, payment_strategy: PaymentStrategy):
        self.payment_strategy = payment_strategy
        self.items: list[tuple[str, float]] = []

    def add_item(self, name: str, price: float) -> None:
        self.items.append((name, price))

    def checkout(self) -> bool:
        total = sum(price for _, price in self.items)
        return self.payment_strategy.pay(total)

Testing Patterns

import pytest
from unittest.mock import Mock, patch, AsyncMock
from dataclasses import dataclass

# Fixtures
@pytest.fixture
def sample_user():
    return User(id=1, email="test@example.com", name="Test User")


@pytest.fixture
def mock_db():
    db = Mock()
    db.query.return_value = [{"id": 1, "name": "Test"}]
    return db


# Parametrized tests
@pytest.mark.parametrize("input,expected", [
    ("hello", "HELLO"),
    ("World", "WORLD"),
    ("", ""),
    ("123", "123"),
])
def test_uppercase(input: str, expected: str):
    assert input.upper() == expected


# Async tests
@pytest.mark.asyncio
async def test_async_fetch():
    with patch('aiohttp.ClientSession.get') as mock_get:
        mock_response = AsyncMock()
        mock_response.json.return_value = {"data": "test"}
        mock_get.return_value.__aenter__.return_value = mock_response

        result = await fetch_data("https://api.example.com")
        assert result == {"data": "test"}


# Exception testing
def test_raises_on_invalid_input():
    with pytest.raises(ValueError, match="Invalid email"):
        validate_email("not-an-email")


# Context manager testing
def test_temporary_directory(tmp_path):
    file_path = tmp_path / "test.txt"
    file_path.write_text("content")
    assert file_path.read_text() == "content"


# Mocking class methods
class TestUserService:
    def test_create_user(self, mock_db):
        service = UserService(mock_db)
        user = service.create(name="Test", email="test@example.com")

        mock_db.insert.assert_called_once()
        assert user.name == "Test"

Performance Optimization

import cProfile
import pstats
from functools import lru_cache
from typing import Callable
import timeit

# Profiling decorator
def profile(func: Callable) -> Callable:
    def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        try:
            return profiler.runcall(func, *args, **kwargs)
        finally:
            stats = pstats.Stats(profiler)
            stats.strip_dirs().sort_stats('cumulative').print_stats(10)
    return wrapper


# Memoization
@lru_cache(maxsize=1000)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)


# Memory-efficient processing
def process_large_dataset(filepath: str) -> int:
    """Process file without loading entirely into memory."""
    total = 0
    with open(filepath) as f:
        for line in f:
            total += process_line(line)
    return total


# Benchmarking
def benchmark(func: Callable, *args, iterations: int = 1000) -> float:
    """Benchmark function execution time."""
    return timeit.timeit(lambda: func(*args), number=iterations) / iterations


# Use __slots__ for memory efficiency
class Point:
    __slots__ = ('x', 'y')

    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

Best Practices

Code Style

1. Follow PEP 8 and use black/ruff for formatting
2. Use type hints everywhere
3. Write docstrings for public APIs
4. Prefer composition over inheritance
5. Use generators for large data processing

Testing

1. Aim for 90%+ code coverage
2. Test edge cases and error conditions
3. Use fixtures for common test data
4. Mock external dependencies
5. Write both unit and integration tests

Performance

1. Profile before optimizing
2. Use appropriate data structures
3. Leverage caching for expensive operations
4. Consider async for I/O-bound operations
5. Use generators for memory efficiency

Output Deliverables

When writing Python code, I will provide:

1. Clean, typed code - With proper annotations
2. Unit tests - Using pytest with fixtures
3. Documentation - Docstrings and examples
4. Performance analysis - When relevant
5. Refactoring suggestions - For existing code
6. Design patterns - Appropriate for the problem
7. Error handling - With custom exceptions

When to Use This Skill

• Writing new Python applications
• Refactoring legacy Python code
• Optimizing Python performance
• Implementing design patterns
• Writing async Python code
• Creating comprehensive test suites
• Code review and improvement