yigityildiz0 Python Expert
Specialized expertise in Python programming, providing deep guidance on type annotations and static analysis, exception handling idioms, async/await concurrency, data modeling with dataclasses and Pydantic, testing with pytest, performance optimization, and modern packaging with pyproject.toml.
When to Use This Skill
Use this skill for:
- Adding type annotations and configuring mypy strict mode
- Designing custom exception hierarchies and error chains
- Implementing async/await patterns with asyncio
- Modeling data with dataclasses and Pydantic
- Writing pytest test suites with fixtures and parametrize
- Profiling and optimizing Python performance
- Setting up Python project structure and packaging
Trigger phrases: "python", "asyncio", "pydantic", "pytest", "mypy", "type hints", "dataclass", "python packaging", "python performance"
What This Skill Does
Provides Python expertise including:
- Type System: Annotations, generics, protocols, mypy strict
- Error Handling: Custom exceptions, error chains, context managers
- Async/Await: asyncio event loop, task groups, async generators
- Data Modeling: dataclasses, Pydantic BaseModel, validation
- Testing: pytest fixtures, parametrize, monkeypatch, coverage
- Performance: Profiling, generators, slots, caching, comprehensions
- Packaging: pyproject.toml, src layout, virtual environments
Instructions
Step 1: Master Type Annotations and mypy-Strict Patterns
Enable Postponed Evaluation and Modern Syntax:
from __future__ import annotations
from typing import TypeVar, Generic, Protocol
# X | Y union syntax (works at runtime with __future__ annotations)
def fetch_user(user_id: int) -> User | None:
"""Return user or None if not found."""
row = db.query("SELECT * FROM users WHERE id = %s", (user_id,))
if row is None:
return None
return User.from_row(row)
# Avoid Optional[X] and Union[X, Y] in new code
# Bad: Optional[str]
# Good: str | None
def parse_header(raw: str) -> str | None:
parts = raw.split(":", maxsplit=1)
return parts[1].strip() if len(parts) == 2 else None
TypeVar and Generic Containers:
from typing import TypeVar, Generic
T = TypeVar("T")
class Stack(Generic[T]):
"""Type-safe generic stack."""
def __init__(self) -> None:
self._items: list[T] = []
def push(self, item: T) -> None:
self._items.append(item)
def pop(self) -> T:
if not self._items:
raise IndexError("pop from empty stack")
return self._items.pop()
def peek(self) -> T:
if not self._items:
raise IndexError("peek at empty stack")
return self._items[-1]
def __len__(self) -> int:
return len(self._items)
# Usage: mypy enforces type consistency
stack: Stack[int] = Stack()
stack.push(42)
stack.push("oops") # mypy error: Argument 1 has incompatible type "str"
Protocol for Structural Subtyping:
from typing import Protocol, runtime_checkable
@runtime_checkable
class Serializable(Protocol):
def to_dict(self) -> dict[str, object]: ...
class User:
def __init__(self, name: str, age: int) -> None:
self.name = name
self.age = age
def to_dict(self) -> dict[str, object]:
return {"name": self.name, "age": self.age}
# User satisfies Serializable without inheriting from it
def serialize(obj: Serializable) -> str:
import json
return json.dumps(obj.to_dict())
# Runtime check also works
assert isinstance(User("Alice", 30), Serializable)
mypy Strict Configuration (pyproject.toml):
[tool.mypy]
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_any_generics = true
check_untyped_defs = true
Step 2: Handle Exceptions Idiomatically
Custom Exception Hierarchy:
class AppError(Exception):
"""Base exception for the application."""
def __init__(self, message: str, code: str = "UNKNOWN") -> None:
super().__init__(message)
self.code = code
class NotFoundError(AppError):
"""Raised when a resource is not found."""
def __init__(self, resource: str, identifier: object) -> None:
super().__init__(
f"{resource} with id {identifier!r} not found",
code="NOT_FOUND",
)
self.resource = resource
self.identifier = identifier
class ValidationError(AppError):
"""Raised when input validation fails."""
def __init__(self, field: str, reason: str) -> None:
super().__init__(
f"Validation failed for {field!r}: {reason}",
code="VALIDATION_ERROR",
)
self.field = field
self.reason = reason
Error Chaining with raise-from:
import logging
logger = logging.getLogger(__name__)
def load_config(path: str) -> dict[str, object]:
try:
with open(path) as f:
import json
return json.load(f)
except FileNotFoundError as exc:
raise AppError(f"Config file missing: {path}") from exc
except json.JSONDecodeError as exc:
raise AppError(f"Invalid JSON in {path}: line {exc.lineno}") from exc
# The original exception is preserved in __cause__
# Tracebacks show: "The above exception was the direct cause of..."
Context Managers for Resource Cleanup:
from contextlib import contextmanager
from typing import Iterator
@contextmanager
def temporary_directory(prefix: str = "tmp") -> Iterator[str]:
"""Create a temp directory and clean it up on exit."""
import tempfile
import shutil
path = tempfile.mkdtemp(prefix=prefix)
try:
yield path
finally:
shutil.rmtree(path, ignore_errors=True)
# Usage
with temporary_directory("build_") as build_dir:
compile_assets(build_dir)
# Directory is cleaned up even if an exception occurs
# Class-based context manager for database transactions
class Transaction:
def __init__(self, conn: Connection) -> None:
self._conn = conn
def __enter__(self) -> Transaction:
self._conn.begin()
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: object,
) -> bool:
if exc_type is not None:
self._conn.rollback()
logger.error("Transaction rolled back: %s", exc_val)
return False # Re-raise the exception
self._conn.commit()
return False
Step 3: Implement Async/Await Patterns
Basic asyncio Patterns:
import asyncio
async def fetch_url(session: aiohttp.ClientSession, url: str) -> str:
"""Fetch a single URL with timeout."""
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
resp.raise_for_status()
return await resp.text()
async def fetch_all(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, return_exceptions=False)
# Entry point
async def main() -> None:
urls = [
"https://api.example.com/users",
"https://api.example.com/orders",
"https://api.example.com/products",
]
results = await fetch_all(urls)
for url, body in zip(urls, results):
print(f"{url}: {len(body)} bytes")
asyncio.run(main())
Task Groups (Python 3.11+):
async def process_batch(items: list[Item]) -> list[Result]:
"""Process items concurrently with structured cancellation."""
results: list[Result] = []
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(process_item(item)) for item in items]
# All tasks completed successfully if we reach here.
# If any task raises, the group cancels all remaining tasks
# and raises an ExceptionGroup.
results = [task.result() for task in tasks]
return results
Async Context Managers and Generators:
from contextlib import asynccontextmanager
from typing import AsyncIterator
@asynccontextmanager
async def managed_connection(dsn: str) -> AsyncIterator[Connection]:
"""Acquire a database connection and release it on exit."""
conn = await asyncpg.connect(dsn)
try:
yield conn
finally:
await conn.close()
# Async generator for streaming results
async def stream_rows(query: str) -> AsyncIterator[dict[str, object]]:
"""Yield rows one at a time without loading all into memory."""
async with managed_connection(DATABASE_URL) as conn:
async with conn.transaction():
async for record in conn.cursor(query):
yield dict(record)
# Consuming an async generator
async def export_csv(query: str, path: str) -> int:
count = 0
with open(path, "w") as f:
async for row in stream_rows(query):
f.write(",".join(str(v) for v in row.values()) + "\n")
count += 1
return count
Semaphore for Rate Limiting:
async def fetch_with_limit(
urls: list[str],
max_concurrent: int = 10,
) -> list[str]:
"""Fetch URLs with a concurrency limit."""
semaphore = asyncio.Semaphore(max_concurrent)
async def limited_fetch(session: aiohttp.ClientSession, url: str) -> str:
async with semaphore:
return await fetch_url(session, url)
async with aiohttp.ClientSession() as session:
tasks = [limited_fetch(session, url) for url in urls]
return await asyncio.gather(*tasks)
Step 4: Model Data with Dataclasses and Pydantic
Dataclass Fundamentals:
from dataclasses import dataclass, field
from datetime import datetime
@dataclass(frozen=True)
class Money:
"""Immutable value object for monetary amounts."""
amount: int # Store as cents to avoid float issues
currency: str = "USD"
def __post_init__(self) -> None:
if self.amount < 0:
raise ValueError("Amount cannot be negative")
def __add__(self, other: Money) -> Money:
if self.currency != other.currency:
raise ValueError(f"Cannot add {self.currency} and {other.currency}")
return Money(self.amount + other.amount, self.currency)
@dataclass
class Order:
"""Mutable entity with computed defaults."""
customer_id: int
items: list[str] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
total: Money = field(default_factory=lambda: Money(0))
def add_item(self, name: str, price: Money) -> None:
self.items.append(name)
self.total = self.total + price
Dataclass with Slots (Python 3.10+):
@dataclass(slots=True)
class Point:
"""Memory-efficient dataclass using __slots__."""
x: float
y: float
z: float = 0.0
def distance_to(self, other: Point) -> float:
return (
(self.x - other.x) ** 2
+ (self.y - other.y) ** 2
+ (self.z - other.z) ** 2
) ** 0.5
Pydantic BaseModel for Validation and Serialization:
from pydantic import BaseModel, Field, field_validator, model_validator
class CreateUserRequest(BaseModel):
"""Validated API request model."""
username: str = Field(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$")
email: str = Field(max_length=255)
age: int = Field(ge=13, le=150)
tags: list[str] = Field(default_factory=list, max_length=10)
@field_validator("email")
@classmethod
def validate_email(cls, v: str) -> str:
if "@" not in v:
raise ValueError("Invalid email address")
return v.lower()
@model_validator(mode="after")
def check_consistency(self) -> CreateUserRequest:
if self.age < 18 and "admin" in self.tags:
raise ValueError("Users under 18 cannot be admins")
return self
# Parse and validate from dict (e.g., request.json())
user = CreateUserRequest.model_validate({"username": "alice", "email": "ALICE@example.com", "age": 25})
print(user.email) # alice@example.com (lowercased by validator)
# Serialize to dict or JSON
user.model_dump()
user.model_dump_json()
Pydantic Settings for Configuration:
from pydantic_settings import BaseSettings
class AppSettings(BaseSettings):
"""Load settings from environment variables."""
database_url: str
redis_url: str = "redis://localhost:6379"
debug: bool = False
max_connections: int = 10
model_config = {"env_prefix": "APP_"}
# Reads APP_DATABASE_URL, APP_REDIS_URL, etc. from environment
settings = AppSettings()
Step 5: Write Thorough Tests with pytest
Basic Test Structure and Parametrize:
import pytest
from myapp.calculator import add, divide
def test_add_positive_numbers() -> None:
assert add(2, 3) == 5
def test_divide_by_zero_raises() -> None:
with pytest.raises(ZeroDivisionError, match="division by zero"):
divide(10, 0)
@pytest.mark.parametrize(
("a", "b", "expected"),
[
(2, 3, 5),
(-1, -1, -2),
(0, 0, 0),
(-1, 1, 0),
(100, 200, 300),
],
ids=["positive", "negative", "zeros", "mixed", "large"],
)
def test_add_parametrized(a: int, b: int, expected: int) -> None:
assert add(a, b) == expected
Fixtures and conftest.py:
# conftest.py
import pytest
from myapp.database import Database
@pytest.fixture
def db(tmp_path: object) -> Database:
"""Create a fresh test database for each test."""
db = Database(f"sqlite:///{tmp_path}/test.db")
db.create_tables()
yield db
db.close()
@pytest.fixture
def sample_user(db: Database) -> User:
"""Insert and return a sample user."""
return db.create_user(name="Alice", email="alice@example.com")
# test_users.py
def test_user_creation(db: Database) -> None:
user = db.create_user(name="Bob", email="bob@example.com")
assert user.name == "Bob"
assert user.id is not None
def test_user_lookup(db: Database, sample_user: User) -> None:
found = db.get_user(sample_user.id)
assert found is not None
assert found.email == "alice@example.com"
Monkeypatch for Isolation:
def test_config_reads_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("APP_DATABASE_URL", "sqlite:///test.db")
monkeypatch.setenv("APP_DEBUG", "true")
settings = AppSettings()
assert settings.database_url == "sqlite:///test.db"
assert settings.debug is True
def test_fetch_retries_on_failure(monkeypatch: pytest.MonkeyPatch) -> None:
call_count = 0
def mock_get(url: str, timeout: int = 30) -> object:
nonlocal call_count
call_count += 1
if call_count < 3:
raise ConnectionError("network error")
return MockResponse(status_code=200, body="ok")
monkeypatch.setattr("myapp.client.requests.get", mock_get)
result = fetch_with_retry("https://api.example.com/data")
assert result == "ok"
assert call_count == 3
Async Test Support:
import pytest
@pytest.mark.asyncio
async def test_async_fetch() -> None:
async with aiohttp.ClientSession() as session:
result = await fetch_url(session, "https://httpbin.org/get")
assert len(result) > 0
# pytest.ini or pyproject.toml:
# [tool.pytest.ini_options]
# asyncio_mode = "auto"
Coverage Configuration (pyproject.toml):
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--strict-markers -q"
[tool.coverage.run]
source = ["src"]
branch = true
[tool.coverage.report]
fail_under = 80
show_missing = true
exclude_lines = [
"pragma: no cover",
"if TYPE_CHECKING:",
"if __name__",
]
Step 6: Optimize Performance
Profiling Before Optimizing:
import cProfile
import pstats
def profile_function(func, *args, **kwargs):
"""Profile a function and print the top 20 calls by cumulative time."""
profiler = cProfile.Profile()
profiler.enable()
result = func(*args, **kwargs)
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative")
stats.print_stats(20)
return result
# Or use the decorator approach
# python -m cProfile -s cumulative myapp/main.py
Generators for Memory Efficiency:
# Bad: loads everything into memory
def read_all_lines(path: str) -> list[str]:
with open(path) as f:
return f.readlines() # Entire file in memory
# Good: yields one line at a time
def read_lines(path: str) -> Iterator[str]:
with open(path) as f:
for line in f:
yield line.strip()
# Generator expression (lazy evaluation)
total = sum(len(line) for line in read_lines("large_file.txt"))
# itertools for composable lazy pipelines
import itertools
def process_large_dataset(path: str) -> Iterator[dict[str, str]]:
lines = read_lines(path)
non_empty = (line for line in lines if line)
batches = itertools.batched(non_empty, 1000) # Python 3.12+
for batch in batches:
yield from process_batch(batch)
Slots for Memory-Efficient Classes:
class HeavyObject:
"""Default: uses __dict__ for attribute storage."""
def __init__(self, x: int, y: int, z: int) -> None:
self.x = x
self.y = y
self.z = z
class LightObject:
"""Slots: fixed attribute set, ~40% less memory per instance."""
__slots__ = ("x", "y", "z")
def __init__(self, x: int, y: int, z: int) -> None:
self.x = x
self.y = y
self.z = z
# Memory comparison (approximate):
# HeavyObject: ~152 bytes per instance
# LightObject: ~88 bytes per instance
Caching with lru_cache and cache:
from functools import lru_cache, cache
@lru_cache(maxsize=256)
def fibonacci(n: int) -> int:
"""Memoized Fibonacci: O(n) instead of O(2^n)."""
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
# Unbounded cache (Python 3.9+)
@cache
def expensive_lookup(key: str) -> dict[str, object]:
return database.query(key)
# Cache with TTL using third-party library
from cachetools import TTLCache
api_cache: TTLCache[str, dict[str, object]] = TTLCache(maxsize=1000, ttl=300)
Comprehensions Over Loops:
# List comprehension (faster than append loop)
squares = [x * x for x in range(1000)]
# Dict comprehension
index = {user.id: user for user in users}
# Set comprehension for deduplication
unique_domains = {email.split("@")[1] for email in emails}
# Conditional comprehension
active_users = [u for u in users if u.is_active and u.last_login is not None]
# Avoid nested comprehensions beyond 2 levels; use a function instead
# Bad:
matrix = [[row[i] for row in data] for i in range(cols)]
# Better:
def transpose(data: list[list[int]]) -> list[list[int]]:
return [list(col) for col in zip(*data)]
Step 7: Structure Projects and Packaging
Modern pyproject.toml Configuration:
[project]
name = "myapp"
version = "1.2.0"
description = "A production application"
requires-python = ">=3.11"
license = {text = "MIT"}
dependencies = [
"pydantic>=2.0,<3.0",
"httpx>=0.25",
"asyncpg>=0.29",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-cov>=5.0",
"pytest-asyncio>=0.23",
"mypy>=1.8",
"ruff>=0.3",
]
[project.scripts]
myapp = "myapp.cli:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.ruff]
line-length = 88
target-version = "py311"
[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM", "TCH"]
src Layout (Recommended):
myapp/
pyproject.toml
src/
myapp/
__init__.py
cli.py
config.py
models/
__init__.py
user.py
services/
__init__.py
auth.py
utils/
__init__.py
logging.py
tests/
unit/
test_models.py
test_services.py
integration/
test_database.py
conftest.py
Virtual Environment Management with uv:
# Create virtual environment
uv venv
# Install project with dev dependencies
uv pip install -e ".[dev]"
# Add a new dependency
uv pip install httpx
# Sync from lock file (reproducible installs)
uv pip sync requirements.lock
# Generate lock file
uv pip compile pyproject.toml -o requirements.lock
Package Entry Points and CLI:
# src/myapp/cli.py
from __future__ import annotations
import argparse
import sys
def main(argv: list[str] | None = None) -> int:
"""Application entry point."""
parser = argparse.ArgumentParser(description="My Application")
parser.add_argument("--config", default="config.toml", help="Config file path")
parser.add_argument("--verbose", action="store_true", help="Enable verbose output")
subparsers = parser.add_subparsers(dest="command", required=True)
serve_parser = subparsers.add_parser("serve", help="Start the server")
serve_parser.add_argument("--port", type=int, default=8080)
migrate_parser = subparsers.add_parser("migrate", help="Run database migrations")
migrate_parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args(argv)
if args.command == "serve":
return run_server(args.port, args.config)
elif args.command == "migrate":
return run_migrations(args.config, dry_run=args.dry_run)
return 0
if __name__ == "__main__":
sys.exit(main())
Best Practices
- Type everything - Run mypy strict in CI; zero type errors is the target
- Raise specific exceptions - Never raise bare Exception; define a project hierarchy
- Use context managers - For any resource that needs cleanup (files, connections, locks)
- Prefer composition - Inherit only from abstract base classes or Protocol
- Immutable by default - Use frozen dataclasses and tuples unless mutation is required
- Profile first - Measure before optimizing; use cProfile or py-spy
- Test at boundaries - Mock external services, not internal functions
- Pin dependencies - Exact versions in production; ranges only in library pyproject.toml
Common Patterns
Pattern 1: Repository Pattern with Protocol
from __future__ import annotations
from typing import Protocol
class UserRepository(Protocol):
def get(self, user_id: int) -> User | None: ...
def save(self, user: User) -> User: ...
def delete(self, user_id: int) -> bool: ...
class PostgresUserRepository:
def __init__(self, conn: asyncpg.Connection) -> None:
self._conn = conn
async def get(self, user_id: int) -> User | None:
row = await self._conn.fetchrow(
"SELECT * FROM users WHERE id = $1", user_id
)
return User(**dict(row)) if row else None
async def save(self, user: User) -> User:
row = await self._conn.fetchrow(
"INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *",
user.name,
user.email,
)
return User(**dict(row))
async def delete(self, user_id: int) -> bool:
result = await self._conn.execute(
"DELETE FROM users WHERE id = $1", user_id
)
return result == "DELETE 1"
class InMemoryUserRepository:
"""Test double that satisfies the same Protocol."""
def __init__(self) -> None:
self._store: dict[int, User] = {}
self._next_id = 1
def get(self, user_id: int) -> User | None:
return self._store.get(user_id)
def save(self, user: User) -> User:
user.id = self._next_id
self._store[self._next_id] = user
self._next_id += 1
return user
def delete(self, user_id: int) -> bool:
return self._store.pop(user_id, None) is not None
Pattern 2: Retry with Exponential Backoff
from __future__ import annotations
import asyncio
import logging
from typing import TypeVar
T = TypeVar("T")
logger = logging.getLogger(__name__)
async def retry(
func,
*args,
max_attempts: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
retryable: tuple[type[Exception], ...] = (ConnectionError, TimeoutError),
) -> T:
"""Retry an async callable with exponential backoff."""
last_exc: Exception | None = None
for attempt in range(1, max_attempts + 1):
try:
return await func(*args)
except retryable as exc:
last_exc = exc
if attempt == max_attempts:
break
delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
logger.warning(
"Attempt %d/%d failed (%s), retrying in %.1fs",
attempt,
max_attempts,
exc,
delay,
)
await asyncio.sleep(delay)
raise last_exc # type: ignore[misc]
# Usage
result = await retry(fetch_url, session, url, max_attempts=5)
Quality Checklist
- All functions have type annotations (parameters and return)
- mypy strict passes with zero errors
- Custom exceptions inherit from a project base class
- Context managers used for resource cleanup
- Async code uses TaskGroup or gather for concurrency
- Pydantic models validate all external input
- pytest coverage at or above 80%
- No mutable default arguments in function signatures
- ruff check and ruff format pass
- pyproject.toml defines all project metadata
Related Skills
performance-testing- Python profiling and benchmarkscicd-architect- Python CI/CD pipelinescode-quality- Python code standardsapi-designer- FastAPI and async web services
Version: 1.0.0 Last Updated: March 2026 Based on: awesome-codex-subagents python-pro patterns
Iterative Refinement Strategy
This skill is optimized for an iterative approach:
- Execute: Perform the core steps defined above.
- Review: Critically analyze the output (coverage, quality, completeness).
- Refine: If targets are not met, repeat the specific implementation steps with improved context.
- Loop: Continue until the definition of done is satisfied.