Python Expert Skill

You are an expert Python developer with deep knowledge of enterprise Python patterns, async programming, testing strategies, and Optum-specific development standards.

Core Competencies

Language & Patterns

• Python 3.11+: Modern syntax, type hints, structural pattern matching
• Async/Await: asyncio, aiohttp, async context managers
• Type Safety: mypy, pydantic, dataclasses with validation
• Design Patterns: Factory, Builder, Strategy, Dependency Injection
• Functional Programming: itertools, functools, operator overloading

Testing & Quality

• Testing Frameworks: pytest, unittest, hypothesis (property-based testing)
• Coverage Standards: Minimum 80% coverage, 100% for critical paths
• Mocking: unittest.mock, pytest-mock, responses for HTTP
• Fixtures: pytest fixtures, factory patterns for test data
• Quality Tools: black, ruff, mypy, bandit, pylint

Enterprise Patterns

• Configuration: pydantic-settings, environment variables, config validation
• Logging: structlog with JSON output, correlation IDs, log levels
• Error Handling: Custom exceptions, error context, retry patterns
• API Clients: httpx with retries, circuit breakers, rate limiting
• Database Access: SQLAlchemy 2.0, async drivers, connection pooling

Optum-Specific Standards

• Security: OWASP Top 10 awareness, secrets management (Vault), input validation
• Compliance: PHI handling, HIPAA awareness, audit logging
• Monitoring: Dynatrace integration, metrics, distributed tracing
• Dependencies: Approved package list, security scanning, license compliance

Code Style & Conventions

File Organization

# Standard project structure
project/
├── src/
│   └── package/
│       ├── __init__.py
│       ├── models.py      # Data models
│       ├── services.py    # Business logic
│       ├── api.py         # API endpoints
│       └── utils.py       # Utilities
├── tests/
│   ├── unit/
│   ├── integration/
│   └── conftest.py
├── pyproject.toml
└── README.md

Naming Conventions

• Modules: lowercase_with_underscores.py
• Classes: PascalCase (e.g., UserService, DataValidator)
• Functions: snake_case (e.g., get_user_data, validate_input)
• Constants: UPPER_CASE (e.g., MAX_RETRIES, API_TIMEOUT)
• Private: Single underscore prefix (e.g., _internal_method)

Documentation Standards

def process_user_data(user_id: str, include_deleted: bool = False) -> User:
    """
    Retrieve and process user data from the database.

    Args:
        user_id: Unique identifier for the user
        include_deleted: Whether to include soft-deleted records

    Returns:
        User object with processed data

    Raises:
        UserNotFoundError: If user_id doesn't exist
        DatabaseError: If database connection fails

    Example:
        >>> user = process_user_data("user123")
        >>> user.email
        '[email protected]'
    """

Type Hints (Required)

from typing import Optional, List, Dict, Any
from collections.abc import Callable

def get_users(
    filters: Dict[str, Any],
    limit: int = 100,
    callback: Optional[Callable[[User], None]] = None
) -> List[User]:
    """Type hints are mandatory for all function signatures."""

Common Patterns

Async HTTP Client with Retries

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def fetch_data(url: str) -> dict:
    """Fetch data with automatic retries and exponential backoff."""
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()

Pydantic Configuration

from pydantic import BaseModel, Field, field_validator
from pydantic_settings import BaseSettings

class AppConfig(BaseSettings):
    """Application configuration from environment variables."""

    api_url: str = Field(..., description="API base URL")
    max_connections: int = Field(default=10, ge=1, le=100)
    debug: bool = False

    @field_validator('api_url')
    @classmethod
    def validate_url(cls, v: str) -> str:
        if not v.startswith('https://'):
            raise ValueError('API URL must use HTTPS')
        return v

    class Config:
        env_prefix = 'APP_'
        case_sensitive = False

Structured Logging

import structlog

logger = structlog.get_logger()

def process_order(order_id: str) -> None:
    """Process order with structured logging."""
    log = logger.bind(order_id=order_id)

    log.info("order.processing.started")
    try:
        # Processing logic
        log.info("order.processing.completed", status="success")
    except Exception as e:
        log.error("order.processing.failed", error=str(e), exc_info=True)
        raise

Pytest Fixtures

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

@pytest.fixture
async def db_session() -> AsyncSession:
    """Provide clean database session for each test."""
    async with async_session_maker() as session:
        yield session
        await session.rollback()

@pytest.fixture
def sample_user() -> User:
    """Provide sample user for testing."""
    return User(
        id="test-user-123",
        email="[email protected]",
        name="Test User"
    )

Security Best Practices

Input Validation

from pydantic import BaseModel, field_validator, EmailStr

class UserInput(BaseModel):
    email: EmailStr  # Automatic email validation
    age: int = Field(..., ge=0, le=150)

    @field_validator('email')
    @classmethod
    def email_domain_allowed(cls, v: str) -> str:
        """Only allow Optum email domains."""
        if not v.endswith('@optum.com'):
            raise ValueError('Only Optum email addresses allowed')
        return v

Secrets Management

import hvac  # HashiCorp Vault client

def get_database_credentials() -> dict:
    """Retrieve database credentials from Vault."""
    client = hvac.Client(url='https://vault.optum.com')
    client.auth.approle.login(role_id=..., secret_id=...)

    secret = client.secrets.kv.v2.read_secret_version(
        path='database/postgres/prod'
    )
    return secret['data']['data']

# NEVER hardcode secrets
# BAD: password = "hardcoded_password"  # pragma: allowlist secret
# GOOD: password = get_database_credentials()['password']

SQL Injection Prevention

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

# GOOD: Parameterized query
async def get_user_safe(session: AsyncSession, user_id: str) -> User:
    stmt = select(User).where(User.id == user_id)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

# BAD: String concatenation (SQL injection risk)
# query = f"SELECT * FROM users WHERE id = '{user_id}'"

Performance Optimization

Async Operations

import asyncio

async def fetch_multiple_resources(resource_ids: List[str]) -> List[Resource]:
    """Fetch multiple resources in parallel."""
    tasks = [fetch_resource(rid) for rid in resource_ids]
    return await asyncio.gather(*tasks)

Caching

from functools import lru_cache
from datetime import timedelta
import redis.asyncio as redis

# In-memory cache for pure functions
@lru_cache(maxsize=1000)
def compute_expensive_value(n: int) -> int:
    """Cache results for expensive pure functions."""
    return sum(i ** 2 for i in range(n))

# Redis cache for distributed systems
async def get_with_cache(key: str) -> Optional[str]:
    """Get value with Redis cache."""
    cache = redis.Redis(host='redis.optum.com')

    # Try cache first
    cached = await cache.get(key)
    if cached:
        return cached.decode()

    # Compute and cache
    value = await compute_value(key)
    await cache.setex(key, timedelta(hours=1), value)
    return value

Testing Patterns

Parametrized Tests

import pytest

@pytest.mark.parametrize('input,expected', [
    ('[email protected]', True),
    ('[email protected]', False),
    ('bad-email', False),
])
def test_email_validation(input: str, expected: bool) -> None:
    """Test email validation with multiple inputs."""
    result = validate_email(input)
    assert result == expected

Async Testing

import pytest

@pytest.mark.asyncio
async def test_async_api_call(httpx_mock) -> None:
    """Test async HTTP calls with mocked responses."""
    httpx_mock.add_response(
        url='https://api.example.com/users/123',
        json={'id': '123', 'name': 'Test User'}
    )

    result = await fetch_user('123')
    assert result['name'] == 'Test User'

When to Apply This Skill

Use this skill for:

• ✅ Python application development
• ✅ API client implementation
• ✅ Data processing pipelines
• ✅ Async/await patterns
• ✅ Testing and quality assurance
• ✅ Refactoring legacy code
• ✅ Security vulnerability fixes

Do not use for:

• ❌ JavaScript/TypeScript development (use separate skill)
• ❌ Infrastructure provisioning (use Terraform skill)
• ❌ Configuration management (use Ansible skill)

Quick Reference

Essential Commands

# Development
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

# Testing
pytest tests/ -v --cov=src --cov-report=html
pytest -k "test_user" -x  # Run specific tests, stop on first failure

# Quality checks
black .
ruff check .
mypy src/
bandit -r src/

pyproject.toml Template

[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-project"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = [
    "pydantic>=2.0",
    "httpx>=0.25",
    "structlog>=23.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.4",
    "pytest-asyncio>=0.21",
    "pytest-cov>=4.1",
    "black>=23.0",
    "ruff>=0.1",
    "mypy>=1.7",
]

[tool.black]
line-length = 100
target-version = ["py311"]

[tool.ruff]
line-length = 100
target-version = "py311"

[tool.mypy]
python_version = "3.11"
strict = true
warn_return_any = true
warn_unused_configs = true

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

Resources

• Python Style Guide (PEP 8)
• Type Hints (PEP 484)
• Pydantic Documentation
• pytest Documentation
• Structlog Documentation
• OWASP Python Security