Python backend expert. PROACTIVELY use when working with Django, FastAPI, Flask, Python APIs. Triggers: python, django, fastapi, flask, async python
/plugin marketplace add nguyenthienthanh/aura-frog/plugin install aura-frog@aurafrogThis skill is limited to using the following tools:
Expert-level Python backend patterns for Django, FastAPI, Flask, and async programming.
This skill activates when:
django, fastapi, or flask in requirements.txt/pyproject.toml*.py files in API/backend context# ✅ GOOD - Type-safe validation
from pydantic import BaseModel, EmailStr, Field, field_validator
class UserCreate(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
password: str = Field(..., min_length=8)
@field_validator('name')
@classmethod
def name_must_not_be_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError('Name cannot be empty')
return v.strip()
model_config = {'str_strip_whitespace': True}
# ✅ GOOD - Dependencies for shared logic
from typing import Annotated
from fastapi import Depends, HTTPException
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
user = await verify_token(token, db)
if user is None:
raise HTTPException(status_code=401, detail="Invalid token")
return user
CurrentUser = Annotated[User, Depends(get_current_user)]
DB = Annotated[AsyncSession, Depends(get_db)]
@app.get("/me")
async def get_me(user: CurrentUser, db: DB):
return user
# ✅ GOOD - Non-blocking operations
from fastapi import BackgroundTasks
@app.post("/users/")
async def create_user(
user: UserCreate,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db)
):
db_user = await crud.create_user(db, user)
background_tasks.add_task(send_welcome_email, user.email)
return db_user
# ✅ GOOD - Custom exceptions
from fastapi import HTTPException
from fastapi.responses import JSONResponse
class AppException(Exception):
def __init__(self, code: str, message: str, status_code: int = 400):
self.code = code
self.message = message
self.status_code = status_code
@app.exception_handler(AppException)
async def app_exception_handler(request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={"code": exc.code, "message": exc.message}
)
# ✅ GOOD - Startup/shutdown handling
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
await database.connect()
yield
# Shutdown
await database.disconnect()
app = FastAPI(lifespan=lifespan)
# ❌ BAD - N+1 queries
users = User.objects.all()
for user in users:
print(user.profile.bio) # 1 query per user!
# ✅ GOOD - select_related for FK
users = User.objects.select_related('profile').all()
# ✅ GOOD - prefetch_related for M2M/reverse FK
posts = Post.objects.prefetch_related('tags', 'comments').all()
from django.db.models import F
# ❌ BAD - Race condition
article = Article.objects.get(pk=1)
article.views += 1
article.save()
# ✅ GOOD - Atomic update
Article.objects.filter(pk=1).update(views=F('views') + 1)
from django.db.models import Q
# ✅ GOOD - OR query
users = User.objects.filter(
Q(is_staff=True) | Q(is_superuser=True)
)
# ✅ GOOD - Combined conditions
users = User.objects.filter(
Q(email__endswith='@company.com') &
(Q(is_active=True) | Q(is_staff=True))
)
# ❌ BAD - Loads all objects
if len(User.objects.filter(email=email)) > 0:
...
# ✅ GOOD - Efficient existence check
if User.objects.filter(email=email).exists():
...
# ❌ BAD - N queries
for user_data in users_data:
User.objects.create(**user_data)
# ✅ GOOD - 1 query
User.objects.bulk_create([
User(**data) for data in users_data
])
from django.db import transaction
@transaction.atomic
def transfer_funds(from_account, to_account, amount):
from_account.balance = F('balance') - amount
from_account.save()
to_account.balance = F('balance') + amount
to_account.save()
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import ForeignKey
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(unique=True, index=True)
name: Mapped[str]
is_active: Mapped[bool] = mapped_column(default=True)
posts: Mapped[list["Post"]] = relationship(back_populates="author")
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
async def get_user(db: AsyncSession, user_id: int) -> User | None:
stmt = select(User).where(User.id == user_id)
result = await db.execute(stmt)
return result.scalar_one_or_none()
from sqlalchemy.orm import selectinload, joinedload
# ✅ GOOD - Prevent N+1
stmt = select(User).options(
selectinload(User.posts), # Separate query
joinedload(User.profile) # JOIN
)
from sqlalchemy import func
async def get_users_paginated(
db: AsyncSession,
page: int = 1,
per_page: int = 20
) -> tuple[list[User], int]:
# Count total
count_stmt = select(func.count()).select_from(User)
total = await db.scalar(count_stmt)
# Get page
stmt = select(User).offset((page - 1) * per_page).limit(per_page)
result = await db.execute(stmt)
users = result.scalars().all()
return users, total
import asyncio
# ✅ GOOD - Concurrent execution
async def get_dashboard_data(user_id: int) -> dict:
user, posts, notifications = await asyncio.gather(
get_user(user_id),
get_user_posts(user_id),
get_notifications(user_id)
)
return {"user": user, "posts": posts, "notifications": notifications}
# ✅ GOOD - Proper timeout
async def fetch_with_timeout(url: str, timeout: float = 5.0):
async with asyncio.timeout(timeout):
async with httpx.AsyncClient() as client:
return await client.get(url)
# ✅ GOOD - Semaphore for concurrency control
semaphore = asyncio.Semaphore(10) # Max 10 concurrent
async def fetch_limited(url: str):
async with semaphore:
return await fetch(url)
# ✅ GOOD - Structured concurrency
async def process_all(items: list[Item]):
async with asyncio.TaskGroup() as tg:
for item in items:
tg.create_task(process_item(item))
# All tasks completed or exception raised
# ✅ GOOD - Modern type hints (Python 3.10+)
def get_users(active: bool | None = None) -> list[User]:
...
# ✅ GOOD - TypeVar for generics
from typing import TypeVar
T = TypeVar('T', bound=BaseModel)
async def get_or_404(db: AsyncSession, model: type[T], id: int) -> T:
obj = await db.get(model, id)
if obj is None:
raise HTTPException(status_code=404)
return obj
# ✅ GOOD - Protocol for structural typing
from typing import Protocol
class Repository(Protocol):
async def get(self, id: int) -> Model | None: ...
async def create(self, data: dict) -> Model: ...
async def delete(self, id: int) -> bool: ...
import pytest
from httpx import AsyncClient
@pytest.fixture
async def client(app):
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def db_session():
async with async_session() as session:
yield session
await session.rollback()
@pytest.mark.parametrize("email,valid", [
("test@example.com", True),
("invalid", False),
("", False),
("a@b.c", True),
])
def test_validate_email(email: str, valid: bool):
result = validate_email(email)
assert result == valid
from factory import Factory, Faker
class UserFactory(Factory):
class Meta:
model = User
email = Faker('email')
name = Faker('name')
is_active = True
# Usage
user = UserFactory()
inactive_user = UserFactory(is_active=False)
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_create_user_sends_email():
with patch('app.services.send_email', new_callable=AsyncMock) as mock:
user = await create_user(UserCreate(email="test@example.com"))
mock.assert_called_once()
class AppError(Exception):
"""Base application error"""
def __init__(self, message: str, code: str = "UNKNOWN"):
self.message = message
self.code = code
super().__init__(message)
class NotFoundError(AppError):
def __init__(self, resource: str, id: int):
super().__init__(f"{resource} with id {id} not found", "NOT_FOUND")
class ValidationError(AppError):
def __init__(self, field: str, message: str):
super().__init__(message, f"VALIDATION_{field.upper()}")
from dataclasses import dataclass
from typing import Generic, TypeVar
T = TypeVar('T')
E = TypeVar('E')
@dataclass
class Ok(Generic[T]):
value: T
@dataclass
class Err(Generic[E]):
error: E
Result = Ok[T] | Err[E]
async def get_user(user_id: int) -> Result[User, str]:
user = await db.get(User, user_id)
if user is None:
return Err("User not found")
return Ok(user)
# Usage
match await get_user(123):
case Ok(user):
print(f"Found: {user.name}")
case Err(error):
print(f"Error: {error}")
checklist[12]{pattern,best_practice}:
FastAPI,Pydantic models + Depends injection
Django ORM,select_related/prefetch_related
SQLAlchemy,Mapped types + selectinload
Async,asyncio.gather for parallel
Typing,Modern 3.10+ syntax X | None
Testing,pytest fixtures + parametrize
Validation,Pydantic or Django Forms
Errors,Custom exception hierarchy
Transactions,atomic or db.transaction
Queries,F() objects for atomic updates
Bulk,bulk_create over loops
Cache,Redis or Django cache
Version: 1.3.0
This skill should be used when the user asks to "create an agent", "add an agent", "write a subagent", "agent frontmatter", "when to use description", "agent examples", "agent tools", "agent colors", "autonomous agent", or needs guidance on agent structure, system prompts, triggering conditions, or agent development best practices for Claude Code plugins.
This skill should be used when the user asks to "create a slash command", "add a command", "write a custom command", "define command arguments", "use command frontmatter", "organize commands", "create command with file references", "interactive command", "use AskUserQuestion in command", or needs guidance on slash command structure, YAML frontmatter fields, dynamic arguments, bash execution in commands, user interaction patterns, or command development best practices for Claude Code.
This skill should be used when the user asks to "create a hook", "add a PreToolUse/PostToolUse/Stop hook", "validate tool use", "implement prompt-based hooks", "use ${CLAUDE_PLUGIN_ROOT}", "set up event-driven automation", "block dangerous commands", or mentions hook events (PreToolUse, PostToolUse, Stop, SubagentStop, SessionStart, SessionEnd, UserPromptSubmit, PreCompact, Notification). Provides comprehensive guidance for creating and implementing Claude Code plugin hooks with focus on advanced prompt-based hooks API.