fastapi-expert by personamanagmentlayer/pcl
npx skills add https://github.com/personamanagmentlayer/pcl --skill fastapi-expert为 FastAPI 提供专家指导——这是一个现代、快速的 Python Web 框架,用于构建具有自动 OpenAPI 文档和类型安全性的 API。
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr, Field
from typing import List, Optional
import uvicorn
app = FastAPI(
title="My API",
description="Production-ready FastAPI",
version="1.0.0"
)
# Models
class UserCreate(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
class UserResponse(BaseModel):
id: int
email: str
name: str
class Config:
from_attributes = True
# Routes
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
# Create user in database
db_user = await db.users.create(**user.dict())
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
user = await db.users.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.get("/users", response_model=List[UserResponse])
async def list_users(skip: int = 0, limit: int = 100):
return await db.users.find_many(skip=skip, limit=limit)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
# Database dependency
async def get_db() -> AsyncSession:
async with async_session() as session:
yield session
# Auth dependency
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
raise credentials_exception
user = await db.get(User, user_id)
if user is None:
raise credentials_exception
return user
# Use dependencies
@app.get("/me", response_model=UserResponse)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
@app.post("/posts")
async def create_post(
post: PostCreate,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
):
db_post = Post(**post.dict(), user_id=current_user.id)
db.add(db_post)
await db.commit()
return db_post
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
@app.post("/token")
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db)
):
user = await authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=401, detail="Invalid credentials")
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, Boolean
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
# CRUD operations
async def get_user(db: AsyncSession, user_id: int):
return await db.get(User, user_id)
async def create_user(db: AsyncSession, user: UserCreate):
hashed_password = get_password_hash(user.password)
db_user = User(email=user.email, hashed_password=hashed_password)
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
from pydantic import BaseModel, validator, root_validator
from typing import Optional
from datetime import date
class PostCreate(BaseModel):
title: str = Field(..., min_length=5, max_length=200)
content: str = Field(..., min_length=10)
published: bool = False
tags: List[str] = Field(default_factory=list, max_items=10)
@validator('tags')
def validate_tags(cls, v):
return [tag.lower().strip() for tag in v]
@validator('title')
def validate_title(cls, v):
if any(word in v.lower() for word in ['spam', 'xxx']):
raise ValueError('Invalid content')
return v
from fastapi import BackgroundTasks
def send_email(email: str, message: str):
# Send email
print(f"Sending email to {email}: {message}")
@app.post("/send-notification")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(send_email, email, "Welcome!")
return {"message": "Notification sent"}
from fastapi.testclient import TestClient
import pytest
client = TestClient(app)
def test_create_user():
response = client.post(
"/users",
json={"email": "test@example.com", "name": "Test User"}
)
assert response.status_code == 201
assert response.json()["email"] == "test@example.com"
def test_get_user():
response = client.get("/users/1")
assert response.status_code == 200
assert "email" in response.json()
@pytest.mark.asyncio
async def test_async_endpoint():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/users")
assert response.status_code == 200
每周安装次数
78
仓库
GitHub 星标数
11
首次出现时间
2026年1月24日
安全审计
已安装于
opencode71
codex69
gemini-cli67
github-copilot63
cursor61
claude-code60
Expert guidance for FastAPI - modern, fast Python web framework for building APIs with automatic OpenAPI documentation and type safety.
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr, Field
from typing import List, Optional
import uvicorn
app = FastAPI(
title="My API",
description="Production-ready FastAPI",
version="1.0.0"
)
# Models
class UserCreate(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
class UserResponse(BaseModel):
id: int
email: str
name: str
class Config:
from_attributes = True
# Routes
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
# Create user in database
db_user = await db.users.create(**user.dict())
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
user = await db.users.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@app.get("/users", response_model=List[UserResponse])
async def list_users(skip: int = 0, limit: int = 100):
return await db.users.find_many(skip=skip, limit=limit)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
# Database dependency
async def get_db() -> AsyncSession:
async with async_session() as session:
yield session
# Auth dependency
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials"
)
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
raise credentials_exception
user = await db.get(User, user_id)
if user is None:
raise credentials_exception
return user
# Use dependencies
@app.get("/me", response_model=UserResponse)
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
@app.post("/posts")
async def create_post(
post: PostCreate,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
):
db_post = Post(**post.dict(), user_id=current_user.id)
db.add(db_post)
await db.commit()
return db_post
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
@app.post("/token")
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db)
):
user = await authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=401, detail="Invalid credentials")
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, Boolean
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
# CRUD operations
async def get_user(db: AsyncSession, user_id: int):
return await db.get(User, user_id)
async def create_user(db: AsyncSession, user: UserCreate):
hashed_password = get_password_hash(user.password)
db_user = User(email=user.email, hashed_password=hashed_password)
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
from pydantic import BaseModel, validator, root_validator
from typing import Optional
from datetime import date
class PostCreate(BaseModel):
title: str = Field(..., min_length=5, max_length=200)
content: str = Field(..., min_length=10)
published: bool = False
tags: List[str] = Field(default_factory=list, max_items=10)
@validator('tags')
def validate_tags(cls, v):
return [tag.lower().strip() for tag in v]
@validator('title')
def validate_title(cls, v):
if any(word in v.lower() for word in ['spam', 'xxx']):
raise ValueError('Invalid content')
return v
from fastapi import BackgroundTasks
def send_email(email: str, message: str):
# Send email
print(f"Sending email to {email}: {message}")
@app.post("/send-notification")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(send_email, email, "Welcome!")
return {"message": "Notification sent"}
from fastapi.testclient import TestClient
import pytest
client = TestClient(app)
def test_create_user():
response = client.post(
"/users",
json={"email": "test@example.com", "name": "Test User"}
)
assert response.status_code == 201
assert response.json()["email"] == "test@example.com"
def test_get_user():
response = client.get("/users/1")
assert response.status_code == 200
assert "email" in response.json()
@pytest.mark.asyncio
async def test_async_endpoint():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/users")
assert response.status_code == 200
Weekly Installs
78
Repository
GitHub Stars
11
First Seen
Jan 24, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
opencode71
codex69
gemini-cli67
github-copilot63
cursor61
claude-code60
agent-browser 浏览器自动化工具 - Vercel Labs 命令行网页操作与测试
163,300 周安装