Expert Python development specialist for backend APIs, async/await optimization, Django/Flask/FastAPI frameworks, type hints, packaging, and performance profiling. Use when building Python backend services, optimizing async code, implementing REST/GraphQL APIs, or requiring Python-specific best practices. Handles dependency management, virtual environments, testing with pytest, and deployment patterns.
Inherits all available tools
Additional assets for this skill
This skill inherits all available tools. When active, it can use any tool Claude has access to.
name: python-specialist description: Expert Python development specialist for backend APIs, async/await optimization, Django/Flask/FastAPI frameworks, type hints, packaging, and performance profiling. Use when building Python backend services, optimizing async code, implementing REST/GraphQL APIs, or requiring Python-specific best practices. Handles dependency management, virtual environments, testing with pytest, and deployment patterns. category: Language Specialists complexity: Medium triggers:
Expert Python development for modern backend systems, API development, and high-performance applications.
This skill provides comprehensive Python expertise across frameworks, async patterns, type safety, and production deployment. It ensures Python code follows best practices, leverages modern features (Python 3.10+), and achieves optimal performance.
Activate this skill when:
Required Knowledge:
Required Tools:
Agent Assignments:
backend-dev: Primary Python implementationcoder: General coding and refactoringtester: pytest test suite creationcode-analyzer: Code quality and connascence analysisperf-analyzer: Performance optimizationStep 1: Initialize Project Structure
Create a production-ready FastAPI project with proper organization:
# Create project structure
mkdir -p my_api/{app,tests,alembic}
cd my_api
# Initialize virtual environment
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
# Install dependencies
pip install fastapi uvicorn[standard] pydantic pydantic-settings sqlalchemy alembic pytest pytest-asyncio httpx
Step 2: Define Data Models with Pydantic
Create type-safe models with validation:
# app/models.py
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional
from datetime import datetime
class UserBase(BaseModel):
email: str = Field(..., description="User email address")
username: str = Field(..., min_length=3, max_length=50)
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserResponse(UserBase):
id: int
created_at: datetime
model_config = ConfigDict(from_attributes=True)
Step 3: Implement API Routes with Dependency Injection
# app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Annotated
from .models import UserCreate, UserResponse
from .dependencies import get_db, get_current_user
app = FastAPI(title="My API", version="1.0.0")
@app.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
user: UserCreate,
db: Annotated[AsyncSession, Depends(get_db)]
) -> UserResponse:
"""Create a new user with email validation."""
# Implementation
return user_response
@app.get("/users/me", response_model=UserResponse)
async def read_current_user(
current_user: Annotated[User, Depends(get_current_user)]
) -> UserResponse:
"""Get current authenticated user."""
return current_user
Step 4: Add Database Integration with SQLAlchemy
# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
SQLALCHEMY_DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(SQLALCHEMY_DATABASE_URL, echo=True)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
async def get_db():
async with async_session_maker() as session:
yield session
Step 5: Write Comprehensive Tests
# tests/test_users.py
import pytest
from httpx import AsyncClient
from app.main import app
@pytest.mark.asyncio
async def test_create_user():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.post("/users", json={
"email": "test@example.com",
"username": "testuser",
"password": "securepass123"
})
assert response.status_code == 201
assert response.json()["email"] == "test@example.com"
Step 1: Identify Blocking Operations
Analyze code for blocking I/O that should be async:
# ❌ BLOCKING (avoid)
import requests
def fetch_data(url: str) -> dict:
response = requests.get(url)
return response.json()
# ✅ ASYNC (preferred)
import httpx
async def fetch_data(url: str) -> dict:
async with httpx.AsyncClient() as client:
response = await client.get(url)
return response.json()
Step 2: Use asyncio.gather for Concurrent Operations
import asyncio
from typing import List
async def fetch_multiple(urls: List[str]) -> List[dict]:
"""Fetch multiple URLs concurrently."""
tasks = [fetch_data(url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if not isinstance(r, Exception)]
Step 3: Handle CPU-Bound Tasks Properly
For CPU-intensive work, use ProcessPoolExecutor to avoid blocking the event loop:
import asyncio
from concurrent.futures import ProcessPoolExecutor
from typing import Any
async def cpu_intensive_task(data: Any) -> Any:
"""Run CPU-bound task in separate process."""
loop = asyncio.get_event_loop()
with ProcessPoolExecutor() as executor:
result = await loop.run_in_executor(executor, blocking_cpu_function, data)
return result
Step 1: Add Type Hints to Functions
from typing import List, Dict, Optional, Union, TypeVar, Generic
T = TypeVar('T')
def process_items(
items: List[Dict[str, Any]],
filter_key: str,
default: Optional[T] = None
) -> List[T]:
"""Process items with type-safe filtering."""
results: List[T] = []
for item in items:
value = item.get(filter_key, default)
if value is not None:
results.append(value)
return results
Step 2: Configure mypy Strict Mode
# mypy.ini
[mypy]
python_version = 3.10
warn_return_any = True
warn_unused_configs = True
disallow_untyped_defs = True
disallow_any_generics = True
strict_optional = True
warn_redundant_casts = True
warn_unused_ignores = True
warn_no_return = True
check_untyped_defs = True
Step 3: Run Type Checking
mypy app/ tests/
Step 1: Profile with cProfile
import cProfile
import pstats
from pstats import SortKey
def profile_function(func, *args, **kwargs):
"""Profile function execution."""
profiler = cProfile.Profile()
profiler.enable()
result = func(*args, **kwargs)
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats(SortKey.CUMULATIVE)
stats.print_stats(20) # Top 20 functions
return result
Step 2: Memory Profiling
from memory_profiler import profile
@profile
def memory_intensive_function():
"""Function to profile memory usage."""
large_list = [i for i in range(1000000)]
return sum(large_list)
Step 3: Optimize with Numba for Numerical Code
from numba import jit
import numpy as np
@jit(nopython=True)
def fast_computation(array: np.ndarray) -> float:
"""JIT-compiled numerical computation."""
total = 0.0
for i in range(array.size):
total += array[i] ** 2
return total
1. Use Virtual Environments Always
venv for simple projects, poetry for complex ones2. Follow PEP 8 Style Guide
black for automatic formattingflake8 or ruff for linting3. Write Docstrings
4. Handle Errors Explicitly
# ✅ GOOD: Specific exception handling
try:
result = risky_operation()
except ValueError as e:
logger.error(f"Invalid value: {e}")
raise HTTPException(status_code=400, detail=str(e))
except DatabaseError as e:
logger.error(f"Database error: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
# ❌ BAD: Bare except
try:
result = risky_operation()
except:
pass # Silently fails
5. Use Context Managers for Resources
# ✅ GOOD: Automatic cleanup
async with httpx.AsyncClient() as client:
response = await client.get(url)
# ❌ BAD: Manual cleanup (error-prone)
client = httpx.AsyncClient()
response = await client.get(url)
client.close() # Might not execute if error occurs
Code Quality:
Performance:
Security:
backend-dev Agent:
# Pre-task: Setup Python environment
npx claude-flow@alpha hooks pre-task --description "Python FastAPI implementation"
python -m venv .venv && source .venv/bin/activate
# During: Track file changes
npx claude-flow@alpha hooks post-edit --file "app/main.py" --memory-key "python-api/fastapi-routes"
# Post-task: Run tests
pytest tests/ --cov=app --cov-report=html
npx claude-flow@alpha hooks post-task --task-id "python-api-implementation"
tester Agent:
# Pre-task: Prepare test environment
npx claude-flow@alpha hooks pre-task --description "Python test suite creation"
# Execute: Run comprehensive tests
pytest tests/ -v --cov=app --cov-report=term-missing --cov-fail-under=90
# Post-task: Report coverage
npx claude-flow@alpha hooks post-task --task-id "python-testing"
Pattern 1: Dependency Injection with FastAPI
from typing import Annotated
from fastapi import Depends
async def get_service() -> MyService:
return MyService()
@app.get("/items")
async def get_items(
service: Annotated[MyService, Depends(get_service)]
):
return await service.get_all()
Pattern 2: Background Tasks
from fastapi import BackgroundTasks
async def send_email(email: str):
# Send email asynchronously
pass
@app.post("/register")
async def register(user: UserCreate, background_tasks: BackgroundTasks):
# Create user
background_tasks.add_task(send_email, user.email)
return {"status": "registered"}
Pattern 3: Middleware for Logging
from fastapi import Request
import time
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
duration = time.time() - start_time
logger.info(f"{request.method} {request.url.path} - {duration:.2f}s")
return response
Issue: Import errors with relative imports
Solution: Ensure __init__.py files exist in all package directories and use absolute imports from package root
Issue: asyncio.run() RuntimeError "Event loop is already running"
Solution: Use await inside async functions, not asyncio.run() which creates a new event loop
Issue: SQLAlchemy "greenlet_spawn has not been called" error
Solution: Ensure using asyncpg driver for PostgreSQL and AsyncSession for all database operations
Issue: mypy errors with third-party libraries
Solution: Install type stubs (pip install types-requests) or ignore with # type: ignore comment
backend-dev: General backend development patternssql-database-specialist: Database optimizationtesting-quality: Advanced testing strategiesdocker-containerization: Containerizing Python appsopentelemetry-observability: Adding tracing to Python servicesDevelopment:
Code Quality:
Performance:
mcp__flow-nexus__sandbox_create with template: "python" for isolated testingmcp__flow-nexus__sandbox_execute for running Python codemcp__connascence-analyzer__analyze_file for detecting code quality issuesmcp__memory-mcp__memory_store for persisting API patterns and decisionsDevelopment Speed:
Code Quality:
Performance:
Skill Version: 1.0.0 Last Updated: 2025-11-02 Maintained By: python-specialist agent