FastAPI

uv add fastapi[all] "uvicorn[standard]" sqlalchemy aiosqlite
uv add pydantic-settings email-validator "pwdlib[argon2]" pyjwt python-multipart

uv run uvicorn app.main:app --host 127.0.0.1 --port 8080 --use-colors --reload

# app/core/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
    
    # 数据库相关配置
    DB_HOST: str = "127.0.0.1"
    DB_PORT: int = 3306
    DB_USER: str = "root"
    DB_PASSWORD: str = "123456"
    DB_NAME: str = "gin"
 
    # jwt相关配置
    JWT_SECRET_KEY: str = "change-me-in-.env"
    JWT_ALGORITHM: str = "HS256"
    JWT_EXPIRE_MINUTES: int = 60 * 24
 
    @property
    def DATABASE_URL(self) -> str:
        return (
            f"mysql+aiomysql://{self.DB_USER}:{self.DB_PASSWORD}"
            f"@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
        )
 
 
@lru_cache
def get_settings() -> Settings:
    return Settings()
 
 
settings = get_settings()

# app/core/database.py
from collections.abc import AsyncGenerator
 
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
 
from app.core.config import get_settings
 
settings = get_settings()
 
 
# 创建连接
engine = create_async_engine( 
    settings.DATABASE_URL,
    echo=True, # 允许打印SQL日志
    pool_pre_ping=True,
    pool_recycle=3600,
)
 
# 创建会话
SessionLocal = async_sessionmaker( 
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
 
# ORM的基类
class Base(DeclarativeBase):
    pass
 
# 创建  依赖注入的入口
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session
 
 

# app/models/user.py
from datetime import datetime, timezone
 
from sqlalchemy import DateTime, String
from sqlalchemy.orm import Mapped, mapped_column
 
from app.core.database import Base
 
 
def utc_now() -> datetime:
    return datetime.now(timezone.utc)
 
 
class User(Base):
    __tablename__ = "users"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=utc_now,
    )

# app/core/security.py
from datetime import datetime, timedelta, timezone
 
import jwt
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
 
from app.core.config import get_settings
 
password_hash = PasswordHash.recommended()
 
 
def hash_password(password: str) -> str:
    return password_hash.hash(password)
 
 
def verify_password(password: str, hashed_password: str) -> bool:
    return password_hash.verify(password, hashed_password)
 
 
def create_access_token(user_id: int) -> str:
    settings = get_settings()
    expires_at = datetime.now(timezone.utc) + timedelta(
        minutes=settings.access_token_expire_minutes,
    )
    payload = {"sub": str(user_id), "exp": expires_at}
    return jwt.encode(
        payload,
        settings.jwt_secret,
        algorithm=settings.jwt_algorithm,
    )
 
 
def decode_access_token(token: str) -> dict:
    settings = get_settings()
    return jwt.decode(
        token,
        settings.jwt_secret,
        algorithms=[settings.jwt_algorithm],
    )

# app/api/deps.py
from typing import Annotated
 
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from sqlalchemy.ext.asyncio import AsyncSession
 
from app.core.database import get_db
from app.core.security import decode_access_token
from app.models.user import User
 
DbSession = Annotated[AsyncSession, Depends(get_db)]
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
 
 
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: DbSession,
) -> User:
    credentials_error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid authentication credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
 
    try:
        payload = decode_access_token(token)
        user_id = int(payload["sub"])
    except (InvalidTokenError, KeyError, TypeError, ValueError) as exc:
        raise credentials_error from exc
 
    user = await db.get(User, user_id)
    if user is None or not user.is_active:
        raise credentials_error
    return user
 
 
CurrentUser = Annotated[User, Depends(get_current_user)]

# 判断token是否合法
from typing import Any
 
import jwt
from fastapi import Depends, Header
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import settings
from app.core.response import Result
from app.db.session import get_db
from app.models.user import UserMapper
 
 
async def get_current_user(
    authorization: str = Header(),
    db: AsyncSession = Depends(get_db),
) -> Result[Any] | None | Any:
    if not authorization.startswith("Bearer "):
        return Result.fail(code=500,message="未提供有效的认证信息")
 
    token = authorization.removeprefix("Bearer ").strip()
 
    try:
        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
    except jwt.ExpiredSignatureError:
        return Result.fail(code=500, message="登录已过期,请重新登录")
    except jwt.InvalidTokenError:
        return Result.fail(code=500, message="无效的登录凭证")
 
    user_id = int(payload["sub"])
 
    stmt = select(UserMapper).where(UserMapper.id == user_id)
    result = await db.execute(stmt)
    user = result.scalar_one_or_none()
 
    if user is None:
        return Result.fail(code=500,message="用户不存在")
 
    return user

from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pwdlib.exceptions import UnknownHashError
 
class ParameterException(Exception):
    def __init__(self, message: str, code: int = 400):
        self.message = message
        self.code = code
 
 
async def parameter_exception_handler(request: Request, exc: ParameterException):
    return JSONResponse(
        status_code=exc.code,
        content={"code": exc.code, "message": exc.message},
    )
 
async def UnknownHashError_handler(request: Request, exc: UnknownHashError):
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "code": status.HTTP_500_INTERNAL_SERVER_ERROR,
            "message": "密码校验失败",
        }
    )
 
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "error_type": error["type"],
            "message": error["msg"],
        })
 
    return JSONResponse(
        content={
            "code": status.HTTP_422_UNPROCESSABLE_ENTITY,
            "message": "请求数据校验失败",
            "errors": errors,
        },
    )
 
 
exception_handlers = [
    (ParameterException, parameter_exception_handler),
    (RequestValidationError, validation_exception_handler),
    (UnknownHashError, UnknownHashError_handler),
]

from typing import Any, Generic, TypeVar
from pydantic import BaseModel
 
T = TypeVar("T")
 
 
class Result(BaseModel, Generic[T]):
    code: int = 200
    message: str = "success"
    data: T | None = None
 
    @classmethod
    def ok(cls, data: T | None = None, message: str = "success") -> "Result[T]":
        return cls(code=200, message=message, data=data)
 
    @classmethod
    def fail(cls, code: int = 500, message: str = "fail") -> "Result[Any]":
        return cls(code=code, message=message, data=None)
 
 

# app/main.py
import uvicorn
import logging
from fastapi import FastAPI
from app.api.v1.user import router as user_router
from app.core.errors import exception_handlers
 
logging.basicConfig(# type: ignore
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s',
    datefmt="%Y-%m-%d %H:%M:%S",
    filemode='a',
    encoding='utf-8'
)
def create_app():
    _app = FastAPI()
    _routers = [user_router]
    for exc_type, handler in exception_handlers:
        _app.add_exception_handler(exc_type, handler)
    for router in _routers:
        _app.include_router(router)
    return _app
 
 
app = create_app()
 
if __name__ == "__main__":
    uvicorn.run(
        app,
        port=3000,
        reload=False,
        host="127.0.0.1",
    )