01 - 后端学习路径¶

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel  # Pydantic数据验证模型

app = FastAPI()

# 数据模型
class User(BaseModel):
    id: int
    name: str
    email: str

# 模拟数据库
users = {}

# 获取所有用户
@app.get("/users")
def get_users():
    return list(users.values())

# 获取单个用户
@app.get("/users/{user_id}")
def get_user(user_id: int):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="User not found")
    return users[user_id]

# 创建用户
@app.post("/users", status_code=201)
def create_user(user: User):
    if user.id in users:
        raise HTTPException(status_code=400, detail="User already exists")
    users[user.id] = user
    return user

# 更新用户
@app.put("/users/{user_id}")
def update_user(user_id: int, user: User):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="User not found")
    users[user_id] = user
    return user

# 删除用户
@app.delete("/users/{user_id}")
def delete_user(user_id: int):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="User not found")
    del users[user_id]
    return {"message": "User deleted"}

# 运行
# uvicorn main:app --reload

from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship, declarative_base
from datetime import datetime, timezone

Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))  # lambda匿名函数：简洁的单行函数

    # 关系
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))

    # 关系
    author = relationship("User", back_populates="posts")

# 连接数据库（SQLAlchemy 2.0 风格）
engine = create_engine('postgresql://user:password@localhost/dbname')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# CRUD操作（SQLAlchemy 2.0 推荐使用 session.execute + select）
from sqlalchemy import select

def create_user(name, email):
    with Session() as session:
        user = User(name=name, email=email)
        session.add(user)
        session.commit()
        session.refresh(user)
        return user

def get_user(user_id):
    with Session() as session:
        # SQLAlchemy 2.0 风格（替代旧的 session.query(User).filter(...))
        user = session.execute(
            select(User).where(User.id == user_id)
        ).scalar_one_or_none()
        return user

def get_all_users():
    with Session() as session:
        # SQLAlchemy 2.0 风格（替代旧的 session.query(User).all())
        users = session.execute(select(User)).scalars().all()
        return users

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta, timezone

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

# 创建JWT令牌
def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 验证JWT令牌
security = HTTPBearer()

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:  # try/except捕获异常
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return get_user(int(user_id))
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

# 使用认证
@app.get("/me")
def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user

import logging
from fastapi import Request
from fastapi.responses import JSONResponse

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 全局异常处理
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):  # async def定义异步函数；用await调用
    logger.error(f"Error: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"message": "Internal server error"}
    )

# 请求日志中间件
@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)  # await等待异步操作完成
    logger.info(f"Response: {response.status_code}")
    return response

from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_get_users():
    response = client.get("/users")
    assert response.status_code == 200  # assert断言：条件为False时抛出AssertionError
    assert isinstance(response.json(), list)  # isinstance检查对象类型

def test_create_user():
    user_data = {
        "id": 1,
        "name": "Test User",
        "email": "test@example.com"
    }
    response = client.post("/users", json=user_data)
    assert response.status_code == 201
    assert response.json()["name"] == "Test User"

def test_get_user_not_found():
    response = client.get("/users/999")
    assert response.status_code == 404