python-expert

Python Expert

Expert guidance for writing clean, maintainable, and idiomatic Python code following best practices and modern Python standards.

When to Use

• Writing new Python functions, classes, or modules
• Refactoring existing code for better structure
• Applying design patterns and best practices
• Adding type hints and documentation
• Improving code readability and maintainability
• Implementing error handling and logging
• Writing tests and test fixtures
• Using async/await and concurrent programming
• Optimizing Python code performance

Core Principles

1. Code Structure and Organization

Module Organization:

"""
Module docstring describing purpose and usage.

This module handles user authentication and session management.
"""

# Standard library imports
import os
import sys
from datetime import datetime
from typing import Optional, List, Dict

# Third-party imports
import requests
from sqlalchemy import create_engine

# Local application imports
from .models import User
from .exceptions import AuthenticationError

# Constants
MAX_LOGIN_ATTEMPTS = 3
SESSION_TIMEOUT = 3600

# Module-level code
logger = logging.getLogger(__name__)

Class Structure:

class UserManager:
    """Manage user operations with clean, single-responsibility methods.
    
    Attributes:
        db_connection: Database connection instance
        cache: Optional cache for user data
    """
    
    def __init__(self, db_connection, cache=None):
        """Initialize the UserManager.
        
        Args:
            db_connection: Active database connection
            cache: Optional cache instance for performance
        """
        self.db_connection = db_connection
        self.cache = cache
        self._session = None
    
    def get_user(self, user_id: int) -> Optional[User]:
        """Retrieve user by ID with caching support."""
        if self.cache:
            cached_user = self.cache.get(f"user:{user_id}")
            if cached_user:
                return cached_user
        
        user = self.db_connection.query(User).filter_by(id=user_id).first()
        
        if user and self.cache:
            self.cache.set(f"user:{user_id}", user, timeout=300)
        
        return user
    
    def __repr__(self):
        return f"UserManager(cache={'enabled' if self.cache else 'disabled'})"

2. Type Hints and Annotations

Use type hints for better code clarity and IDE support:

from typing import List, Dict, Optional, Union, Tuple, Callable, TypeVar, Generic

# Function signatures
def process_data(
    items: List[str],
    threshold: int = 10,
    callback: Optional[Callable[[str], None]] = None
) -> Dict[str, int]:
    """Process items and return frequency count."""
    result: Dict[str, int] = {}
    
    for item in items:
        result[item] = result.get(item, 0) + 1
        if callback:
            callback(item)
    
    return result

# Complex types
UserData = Dict[str, Union[str, int, List[str]]]

def get_user_data(user_id: int) -> Optional[UserData]:
    """Fetch user data with structured typing."""
    pass

# Generic types
T = TypeVar('T')

class Repository(Generic[T]):
    """Generic repository pattern."""
    
    def find_by_id(self, id: int) -> Optional[T]:
        """Find entity by ID."""
        pass
    
    def save(self, entity: T) -> T:
        """Save entity to repository."""
        pass

3. Python Idioms and Best Practices

List Comprehensions:

# Good - clear and efficient
squares = [x**2 for x in range(10)]
even_squares = [x**2 for x in range(10) if x % 2 == 0]

# Dictionary comprehensions
word_lengths = {word: len(word) for word in words}

# Set comprehensions
unique_lengths = {len(word) for word in words}

Context Managers:

# File handling
with open('data.txt', 'r') as f:
    data = f.read()

# Custom context manager
from contextlib import contextmanager

@contextmanager
def database_transaction(connection):
    """Ensure transaction is committed or rolled back."""
    try:
        yield connection
        connection.commit()
    except Exception:
        connection.rollback()
        raise
    
# Usage
with database_transaction(db_conn) as conn:
    conn.execute("INSERT INTO users VALUES (...)")

Generators for Memory Efficiency:

def read_large_file(file_path: str):
    """Read large file line by line without loading into memory."""
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()

def fibonacci(n: int):
    """Generate Fibonacci sequence up to n terms."""
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

# Usage
for num in fibonacci(10):
    print(num)

Unpacking and Multiple Assignment:

# Tuple unpacking
first, *middle, last = [1, 2, 3, 4, 5]

# Dictionary unpacking
defaults = {'timeout': 30, 'retries': 3}
config = {'timeout': 60, 'endpoint': '/api'}
final_config = {**defaults, **config}

# Function arguments
def process(name, age, city='Unknown'):
    pass

user_data = {'name': 'Alice', 'age': 30, 'city': 'NYC'}
process(**user_data)

4. Error Handling

Specific Exception Handling:

def safe_divide(a: float, b: float) -> Optional[float]:
    """Safely divide two numbers with proper error handling."""
    try:
        result = a / b
    except ZeroDivisionError:
        logger.error(f"Cannot divide {a} by zero")
        return None
    except TypeError as e:
        logger.error(f"Invalid types for division: {e}")
        raise
    else:
        return result
    finally:
        logger.debug("Division operation completed")

# Custom exceptions
class ValidationError(Exception):
    """Raised when data validation fails."""
    pass

class DataNotFoundError(Exception):
    """Raised when requested data doesn't exist."""
    
    def __init__(self, entity_type: str, entity_id: int):
        self.entity_type = entity_type
        self.entity_id = entity_id
        super().__init__(f"{entity_type} with ID {entity_id} not found")

# Usage
def get_user(user_id: int) -> User:
    user = db.query(User).get(user_id)
    if not user:
        raise DataNotFoundError("User", user_id)
    return user

5. Decorators and Higher-Order Functions

Common Decorator Patterns:

from functools import wraps
import time

def timer(func):
    """Measure function execution time."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        logger.info(f"{func.__name__} took {end - start:.4f}s")
        return result
    return wrapper

def retry(max_attempts: int = 3, delay: float = 1.0):
    """Retry function on failure."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    logger.warning(f"Attempt {attempt + 1} failed: {e}")
                    time.sleep(delay)
        return wrapper
    return decorator

def validate_types(**type_hints):
    """Validate function argument types."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Bind arguments to parameter names
            bound = inspect.signature(func).bind(*args, **kwargs)
            bound.apply_defaults()
            
            for param_name, expected_type in type_hints.items():
                if param_name in bound.arguments:
                    value = bound.arguments[param_name]
                    if not isinstance(value, expected_type):
                        raise TypeError(
                            f"{param_name} must be {expected_type.__name__}"
                        )
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Usage
@timer
@retry(max_attempts=3, delay=2.0)
def fetch_data(url: str):
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

6. Data Classes and Named Tuples

Modern Data Structures:

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

@dataclass
class User:
    """User data structure with automatic methods."""
    id: int
    username: str
    email: str
    created_at: datetime = field(default_factory=datetime.now)
    tags: List[str] = field(default_factory=list)
    
    def __post_init__(self):
        """Validate data after initialization."""
        if '@' not in self.email:
            raise ValueError(f"Invalid email: {self.email}")

# Frozen dataclass (immutable)
@dataclass(frozen=True)
class Point:
    x: float
    y: float
    
    def distance_from_origin(self) -> float:
        return (self.x**2 + self.y**2) ** 0.5

# Named tuples for simple structures
from collections import namedtuple

Coordinate = namedtuple('Coordinate', ['latitude', 'longitude'])
location = Coordinate(40.7128, -74.0060)

7. Async/Await for Concurrent Operations

Asynchronous Programming:

import asyncio
from typing import List

async def fetch_url(session, url: str) -> str:
    """Fetch single URL asynchronously."""
    async with session.get(url) as response:
        return await response.text()

async def fetch_all_urls(urls: List[str]) -> List[str]:
    """Fetch multiple URLs concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def process_with_timeout(coro, timeout: float):
    """Execute coroutine with timeout."""
    try:
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        logger.error(f"Operatio

---

*Content truncated.*