python
Use when building FastAPI applications, implementing async endpoints, setting up Pydantic schemas, working with SQLAlchemy, or writing pytest tests for Python backend services.
ソース情報
- リポジトリ
- MadAppGang/claude-code
- ソースの最終更新活動
- 2026年1月20日 08:16
- 検出された SKILL.md の言語
- 英語
- スター
- 284
- フォーク
- 25
インストール方法
デフォルトでは、最初にソースを確認する Prompt が選択されています。直接コマンドに切り替えるか、ローカルコピーをダウンロードすることもできます。
ソースファイルを確認
インストールを決める前に、SKILL.md と SkillsMP に表示されている付属ファイルをお読みください。
SKILL.md を表示中
SKILL.md
ソースの指示 · 読み取り専用プレビュー- name
- python
- version
- 1.0.0
- description
- Use when building FastAPI applications, implementing async endpoints, setting up Pydantic schemas, working with SQLAlchemy, or writing pytest tests for Python backend services.
- keywords
- ["Python","FastAPI","Pydantic","SQLAlchemy","async","pytest","backend","API"]
- plugin
- dev
- updated
- 2026-01-20T00:00:00.000Z
# Python Backend Patterns
## Overview
Python patterns for building backend services with FastAPI.
## Project Structure
```
project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app
│ ├── config.py # Configuration
│ ├── dependencies.py # Dependency injection
│ ├── routers/ # API routes
│ │ ├── __init__.py
│ │ └── users.py
│ ├── services/ # Business logic
│ ├── repositories/ # Data access
│ ├── models/ # SQLAlchemy models
│ ├── schemas/ # Pydantic schemas
│ └── utils/ # Utilities
├── tests/ # Test files
├── migrations/ # Alembic migrations
├── pyproject.toml
└── requirements.txt
```
## FastAPI Application
### Main Application
```python
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.config import settings
from app.routers import users, auth
from app.database import engine, Base
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Shutdown
await engine.dispose()
app = FastAPI(
title="My API",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])
@app.get("/health")
async def health():
return {"status": "ok"}
```
### Configuration
```python
# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
database_url: str = "postgresql+asyncpg://localhost/app"
redis_url: str = "redis://localhost:6379"
secret_key: str = "your-secret-key"
access_token_expire_minutes: int = 30
cors_origins: list[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
@lru_cache
def get_settings() -> Settings:
return Settings()
settings = get_settings()
```
## Pydantic Schemas
```python
# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional
class UserBase(BaseModel):
name: str = Field(..., min_length=2, max_length=100)
email: EmailStr
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=2, max_length=100)
email: Optional[EmailStr] = None
class UserResponse(UserBase):
id: str
created_at: datetime
class Config:
from_attributes = True
class PaginatedResponse(BaseModel):
items: list[UserResponse]
total: int
page: int
page_size: int
```
## SQLAlchemy Models
```python
# app/models/user.py
from sqlalchemy import Column, String, DateTime, Boolean
from sqlalchemy.sql import func
from app.database import Base
import uuid
class User(Base):
__tablename__ = "users"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, nullable=False, index=True)
password_hash = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
```
## Async Database
```python
# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from app.config import settings
engine = create_async_engine(settings.database_url, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
async def get_db():
async with async_session() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
```
## Repository Pattern
```python
# app/repositories/user.py
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
class UserRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def find_by_id(self, id: str) -> User | None:
result = await self.session.execute(
select(User).where(User.id == id)
)
return result.scalar_one_or_none()
async def find_by_email(self, email: str) -> User | None:
result = await self.session.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()
async def find_all(self, page: int = 1, page_size: int = 20) -> tuple[list[User], int]:
offset = (page - 1) * page_size
# Get items
result = await self.session.execute(
select(User)
.order_by(User.created_at.desc())
.offset(offset)
.limit(page_size)
)
items = result.scalars().all()
# Get total count
count_result = await self.session.execute(
select(func.count()).select_from(User)
)
total = count_result.scalar()
return list(items), total
async def create(self, data: UserCreate, password_hash: str) -> User:
user = User(
name=data.name,
email=data.email,
password_hash=password_hash,
)
self.session.add(user)
await self.session.flush()
return user
async def update(self, user: User, data: UserUpdate) -> User:
for key, value in data.model_dump(exclude_unset=True).items():
setattr(user, key, value)
await self.session.flush()
return user
async def delete(self, user: User) -> None:
await self.session.delete(user)
```
## API Routes
```python
# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.repositories.user import UserRepository
from app.services.user import UserService
from app.schemas.user import UserCreate, UserUpdate, UserResponse, PaginatedResponse
from app.dependencies import get_current_user
router = APIRouter()
def get_user_service(db: AsyncSession = Depends(get_db)) -> UserService:
return UserService(UserRepository(db))
@router.get("", response_model=PaginatedResponse)
async def list_users(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
service: UserService = Depends(get_user_service),
):
users, total = await service.find_all(page, page_size)
return PaginatedResponse(
items=users,
total=total,
page=page,
page_size=page_size,
)
@router.get("/{id}", response_model=UserResponse)
async def get_user(
id: str,
service: UserService = Depends(get_user_service),
):
user = await service.find_by_id(id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
data: UserCreate,
service: UserService = Depends(get_user_service),
):
return await service.create(data)
@router.patch("/{id}", response_model=UserResponse)
async def update_user(
id: str,
data: UserUpdate,
service: UserService = Depends(get_user_service),
current_user: User = Depends(get_current_user),
):
user = await service.update(id, data)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
id: str,
service: UserService = Depends(get_user_service),
current_user: User = Depends(get_current_user),
):
success = await service.delete(id)
if not success:
raise HTTPException(status_code=404, detail="User not found")
```
## Authentication
```python
# app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from app.config import settings
from app.models.user import User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db),
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
repo = UserRepository(db)
user = await repo.find_by_id(user_id)
if user is None:
raise credentials_exception
return user
```
## Testing
```python
# tests/test_users.py
import pytest
from httpx import AsyncClient
from app.main import app
from app.database import get_db, async_session
@pytest.fixture
async def client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def db():
async with async_session() as session:
yield session
await session.rollback()
@pytest.mark.asyncio
async def test_list_users(client: AsyncClient):
response = await client.get("/api/users")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
response = await client.post("/api/users", json={
"name": "John Doe",
"email": "john@example.com",
"password": "password123",
})
assert response.status_code == 201
data = response.json()
assert data["name"] == "John Doe"
assert data["email"] == "john@example.com"
@pytest.mark.asyncio
async def test_create_user_validation(client: AsyncClient):
response = await client.post("/api/users", json={
"name": "J", # Too short
"email": "invalid", # Invalid email
})
assert response.status_code == 422
```
---
*Python FastAPI patterns for backend development*
GitHubで見る