feat: Phase 1 后端骨架主体入库(补 backend/app 主体与 requirements.txt)
上轮 3 次 commit 只入库了修补文件,主体目录漏入库。本次补入 backend/app/、backend/requirements.txt、backend/scripts/__init__.py。同步在 .gitignore 加 homePC/(制片人家用机带过来的参考文件,不进库)。
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
"""
|
||||
应用配置 — 用 pydantic-settings 的 BaseSettings 读取 .env
|
||||
"""
|
||||
|
||||
from pydantic_settings import BaseSettings
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
DATABASE_URL: str
|
||||
SECRET_KEY: str = "change-me-to-a-random-string-in-production"
|
||||
SESSION_MAX_AGE: int = 86400 # 秒,默认 1 天
|
||||
|
||||
model_config = {
|
||||
"env_file": ".env",
|
||||
"env_file_encoding": "utf-8",
|
||||
}
|
||||
|
||||
|
||||
settings = Settings()
|
||||
@@ -0,0 +1,56 @@
|
||||
"""
|
||||
FastAPI 依赖注入工具 — 获取当前用户、角色鉴权
|
||||
"""
|
||||
|
||||
from fastapi import Depends, HTTPException, Request, status
|
||||
from sqlmodel import Session
|
||||
|
||||
from app.db.session import get_session
|
||||
from app.models.user import User, UserRole
|
||||
|
||||
|
||||
def get_current_user(
|
||||
request: Request,
|
||||
session: Session = Depends(get_session),
|
||||
) -> User:
|
||||
"""从 Session 中取 user_id,查库返回当前用户。
|
||||
|
||||
未登录或用户不存在 / 已停用时抛出 401。
|
||||
"""
|
||||
user_id = request.session.get("user_id")
|
||||
if user_id is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="请先登录",
|
||||
)
|
||||
user = session.get(User, user_id)
|
||||
if user is None or not user.is_active:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="用户不存在或已被停用",
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
def require_role(*allowed_roles: UserRole):
|
||||
"""返回一个 Depends 依赖,要求当前用户拥有指定的角色之一。
|
||||
|
||||
使用示例:
|
||||
@router.get("/admin-only")
|
||||
def admin_endpoint(user=Depends(require_role(UserRole.zhipianren))):
|
||||
...
|
||||
|
||||
@router.get("/staff")
|
||||
def staff_endpoint(user=Depends(require_role(UserRole.zhipianren, UserRole.zebian))):
|
||||
...
|
||||
"""
|
||||
|
||||
def check_role(user: User = Depends(get_current_user)) -> User:
|
||||
if user.role not in allowed_roles:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="权限不足,无法执行此操作",
|
||||
)
|
||||
return user
|
||||
|
||||
return check_role
|
||||
@@ -0,0 +1,17 @@
|
||||
"""
|
||||
认证安全工具 — passlib + bcrypt
|
||||
"""
|
||||
|
||||
from passlib.context import CryptContext
|
||||
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
|
||||
def hash_password(plain: str) -> str:
|
||||
"""用 bcrypt 生成密码哈希。"""
|
||||
return pwd_context.hash(plain)
|
||||
|
||||
|
||||
def verify_password(plain: str, hashed: str) -> bool:
|
||||
"""验证明文密码与哈希是否匹配。"""
|
||||
return pwd_context.verify(plain, hashed)
|
||||
Reference in New Issue
Block a user