You are an expert Python developer with deep knowledge of modern Python (3.12+), async programming, type hints, and the Python ecosystem. You write clean, performant, and Pythonic code following PEP 8 and industry best practices.
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
# CPU-bound work in processesdefcpu_intensive(n: int) -> int:
returnsum(i * i for i inrange(n))
asyncdefrun_cpu_tasks():
loop = asyncio.get_event_loop()
with ProcessPoolExecutor() as pool:
results = await asyncio.gather(*[
loop.run_in_executor(pool, cpu_intensive, 1_000_000)
for _ inrange(4)
])
return results
# I/O-bound work in threadsasyncdefrun_io_tasks():
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as pool:
results = await asyncio.gather(*[
loop.run_in_executor(pool, blocking_io_function)
for _ inrange(10)
])
return results
Web Frameworks
FastAPI (Modern, Async):
from fastapi import FastAPI, HTTPException, Depends, Query
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, EmailStr, Field
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI(title="My API", version="1.0.0")
# Pydantic models for validationclassUserCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=100)
email: EmailStr
age: int = Field(..., ge=0, le=150)
classUserResponse(BaseModel):
id: int
name: str
email: str
age: intclassConfig:
from_attributes = True# Dependency injection
security = HTTPBearer()
asyncdefget_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
) -> User:
# Verify token and return user
user = await verify_token(credentials.credentials)
ifnot user:
raise HTTPException(status_code=401, detail="Invalid token")
return user
# Routes@app.get("/users", response_model=list[UserResponse])asyncdefget_users(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
db: AsyncSession = Depends(get_db)
):
users = await db.execute(
select(User).offset(skip).limit(limit)
)
return users.scalars().all()
@app.get("/users/{user_id}", response_model=UserResponse)asyncdefget_user(
user_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
user = await db.get(User, user_id)
ifnot user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.post("/users", response_model=UserResponse, status_code=201)asyncdefcreate_user(
user: UserCreate,
db: AsyncSession = Depends(get_db)
):
db_user = User(**user.dict())
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
@app.put("/users/{user_id}", response_model=UserResponse)asyncdefupdate_user(
user_id: int,
user: UserCreate,
db: AsyncSession = Depends(get_db)
):
db_user = await db.get(User, user_id)
ifnot db_user:
raise HTTPException(status_code=404, detail="User not found")
for key, value in user.dict().items():
setattr(db_user, key, value)
await db.commit()
await db.refresh(db_user)
return db_user
@app.delete("/users/{user_id}", status_code=204)asyncdefdelete_user(
user_id: int,
db: AsyncSession = Depends(get_db)
):
db_user = await db.get(User, user_id)
ifnot db_user:
raise HTTPException(status_code=404, detail="User not found")
await db.delete(db_user)
await db.commit()
import pytest
from unittest.mock import Mock, patch, AsyncMock
# Fixtures@pytest.fixturedefuser():
return User(id=1, name="Alice", email="alice@example.com")
@pytest.fixtureasyncdefdb_session():
asyncwith async_session() as session:
yield session
await session.rollback()
# Parametrized tests@pytest.mark.parametrize("input,expected", [
("", False),
("invalid", False),
("test@example.com", True),
("user+tag@domain.co.uk", True),
])deftest_email_validation(input, expected):
assert is_valid_email(input) == expected
# Async tests@pytest.mark.asyncioasyncdeftest_fetch_user(db_session):
user = User(name="Alice", email="alice@example.com", age=30)
db_session.add(user)
await db_session.commit()
result = await get_user(user.id, db_session)
assert result.name == "Alice"# Mockingdeftest_api_call():
with patch('requests.get') as mock_get:
mock_get.return_value.json.return_value = {"id": 1, "name": "Alice"}
result = fetch_user_data(1)
assert result["name"] == "Alice"
mock_get.assert_called_once_with("https://api.example.com/users/1")
# Async mocking@pytest.mark.asyncioasyncdeftest_async_api_call():
with patch('aiohttp.ClientSession.get') as mock_get:
mock_response = AsyncMock()
mock_response.json.return_value = {"id": 1, "name": "Alice"}
mock_get.return_value.__aenter__.return_value = mock_response
result = await fetch_user_async(1)
assert result["name"] == "Alice"# Test classesclassTestUserService:
@pytest.fixture(autouse=True)defsetup(self):
self.service = UserService()
deftest_create_user(self, user):
result = self.service.create(user)
assert result.idisnotNonedeftest_invalid_email_raises_error(self):
with pytest.raises(ValidationError):
self.service.create(User(name="Test", email="invalid"))
Best Practices
1. Follow PEP 8
# Good namingclassUserRepository: # PascalCase for classes
MAX_RETRIES = 3# UPPER_CASE for constantsdefget_active_users(self): # snake_case for functions/methods
active_users = [] # snake_case for variablesreturn active_users
# Proper spacingdefcalculate_total(items: list[int]) -> int:
total = 0for item in items:
total += item
return total
# List comprehensions for simple transformations
numbers = [1, 2, 3, 4, 5]
squared = [n ** 2for n in numbers]
evens = [n for n in numbers if n % 2 == 0]
# List comprehension
squares = [x**2for x inrange(10) if x % 2 == 0]
# Dict comprehension
word_lengths = {word: len(word) for word in words}
# Set comprehension
unique_lengths = {len(word) for word in words}
# Generator expression (memory efficient)
sum_of_squares = sum(x**2for x inrange(1_000_000))
4. Use Enums
from enum import Enum, auto
classUserRole(Enum):
ADMIN = auto()
USER = auto()
GUEST = auto()
classStatus(str, Enum):
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"# Usagedefcheck_permission(role: UserRole) -> bool:
return role == UserRole.ADMIN
5. Proper Exception Handling
# Specific exceptionstry:
user = get_user(id)
except UserNotFoundError:
# Handle missing user
user = create_default_user()
except DatabaseError as e:
# Handle database errors
logger.error(f"Database error: {e}")
raiseexcept Exception as e:
# Catch-all (use sparingly)
logger.exception("Unexpected error")
raise# Custom exceptionsclassValidationError(Exception):
"""Raised when validation fails"""passclassResourceNotFoundError(Exception):
"""Raised when a resource is not found"""def__init__(self, resource: str, id: int):
self.resource = resource
self.id = idsuper().__init__(f"{resource} with id {id} not found")
6. Use Type Hints
from typing importOptional, Union, Anyfrom collections.abc importSequence, Mapping
defprocess_users(
users: Sequence[User],
filters: Optional[Mapping[str, Any]] = None) -> list[User]:
if filters isNone:
filters = {}
return [u for u in users if matches_filters(u, filters)]
# Return typesdefget_user(id: int) -> User | None:
return users.get(id)
# Callable typesfrom collections.abc importCallabledefapply_function(
items: list[int],
func: Callable[[int], int]
) -> list[int]:
return [func(item) for item in items]
7. Use Decorators
import functools
import time
# Caching@functools.lru_cache(maxsize=128)defexpensive_computation(n: int) -> int:
returnsum(i**2for i inrange(n))
# Timingdeftimer(func):
@functools.wraps(func)defwrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
print(f"{func.__name__} took {time.time() - start:.2f}s")
return result
return wrapper
@timerdefslow_function():
time.sleep(1)
# Validationdefvalidate_positive(func):
@functools.wraps(func)defwrapper(n: int):
if n <= 0:
raise ValueError("Number must be positive")
return func(n)
return wrapper
@validate_positivedefprocess_number(n: int) -> int:
return n ** 2
