forked from Kexing/AI4SE_Practices
470 lines
14 KiB
Python
470 lines
14 KiB
Python
"""用户管理系统核心功能模块"""
|
||
|
||
import hashlib
|
||
from typing import Dict, List, Optional, Tuple, Union
|
||
from datetime import datetime, timedelta
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import Column, Integer, String, Boolean, DateTime
|
||
from sqlalchemy.ext.declarative import declarative_base
|
||
from fastapi import HTTPException, status
|
||
from pydantic import BaseModel, Field, validator
|
||
|
||
# SQLAlchemy 模型基类
|
||
Base = declarative_base()
|
||
|
||
|
||
class DBUser(Base):
|
||
"""数据库用户模型"""
|
||
__tablename__ = "users"
|
||
|
||
id = Column(Integer, primary_key=True, index=True)
|
||
username = Column(String(32), unique=True, index=True)
|
||
email = Column(String(255), unique=True, index=True)
|
||
password_hash = Column(String(128))
|
||
role = Column(String(32), default="user")
|
||
created_at = Column(DateTime, default=datetime.utcnow)
|
||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
||
is_active = Column(Boolean, default=True)
|
||
|
||
|
||
class UserCreate(BaseModel):
|
||
"""
|
||
用户创建模型
|
||
|
||
Attributes:
|
||
username: str - 用户名，长度3-32个字符
|
||
email: str - 邮箱地址
|
||
password: str - 密码，至少8个字符，包含字母和数字
|
||
role: str - 用户角色，默认为'user'
|
||
"""
|
||
username: str
|
||
email: str
|
||
password: str
|
||
role: str = "user"
|
||
|
||
@validator('username')
|
||
def username_length(cls, v: str) -> str:
|
||
"""验证用户名长度"""
|
||
if len(v) < 3 or len(v) > 32:
|
||
raise ValueError('用户名长度必须在3-32个字符之间')
|
||
return v
|
||
|
||
@validator('password')
|
||
def password_strength(cls, v: str) -> str:
|
||
"""验证密码强度"""
|
||
if len(v) < 8:
|
||
raise ValueError('密码长度必须至少8个字符')
|
||
if not any(c.isalpha() for c in v):
|
||
raise ValueError('密码必须包含字母')
|
||
if not any(c.isdigit() for c in v):
|
||
raise ValueError('密码必须包含数字')
|
||
return v
|
||
|
||
|
||
class User(BaseModel):
|
||
"""
|
||
用户信息模型
|
||
|
||
Attributes:
|
||
id: int - 用户ID
|
||
username: str - 用户名
|
||
email: str - 邮箱地址
|
||
role: str - 用户角色
|
||
created_at: datetime - 创建时间
|
||
updated_at: datetime - 更新时间
|
||
is_active: bool - 用户是否激活
|
||
"""
|
||
id: int
|
||
username: str
|
||
email: str
|
||
role: str
|
||
created_at: datetime
|
||
updated_at: datetime
|
||
is_active: bool
|
||
|
||
class Config:
|
||
orm_mode = True
|
||
|
||
|
||
def hash_password(password: str) -> str:
|
||
"""
|
||
对密码进行哈希处理
|
||
|
||
Args:
|
||
password: str - 原始密码字符串
|
||
|
||
Returns:
|
||
str - 哈希后的密码字符串
|
||
|
||
Examples:
|
||
>>> hash_password("secure_password123")
|
||
"$2b$12$XaBcDeFgHiJkLmNoPqRsTuVwXyZaBcDeFgHiJkLmNoPqRsTuVwXyZ"
|
||
|
||
Raises:
|
||
ValueError: 当密码为空时抛出
|
||
|
||
Complexity:
|
||
时间复杂度: O(n)，其中n是密码长度
|
||
空间复杂度: O(1)
|
||
"""
|
||
if not password:
|
||
raise ValueError("密码不能为空")
|
||
|
||
# 使用bcrypt进行密码哈希 (此处仅为示例，实际实现会使用专门的bcrypt库)
|
||
salt = hashlib.sha256(password.encode()).hexdigest()[:29]
|
||
hashed = hashlib.sha256(f"{password}{salt}".encode()).hexdigest()
|
||
return f"$2b$12${salt}{hashed[:31]}"
|
||
|
||
|
||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||
"""
|
||
验证密码是否正确
|
||
|
||
Args:
|
||
plain_password: str - 原始密码字符串
|
||
hashed_password: str - 存储的哈希密码字符串
|
||
|
||
Returns:
|
||
bool - 密码是否匹配
|
||
|
||
Examples:
|
||
>>> verify_password("secure_password123", "$2b$12$XaBcDeFgHiJkLmNoPqRsTuVwXyZaBcDeFgHiJkLmNoPqRsTuVwXyZ")
|
||
True
|
||
|
||
Raises:
|
||
ValueError: 当任何参数为空时抛出
|
||
|
||
Complexity:
|
||
时间复杂度: O(n)，其中n是密码长度
|
||
空间复杂度: O(1)
|
||
"""
|
||
if not plain_password or not hashed_password:
|
||
raise ValueError("密码参数不能为空")
|
||
|
||
# 从哈希密码中提取盐值 (实际实现会使用专门的bcrypt库)
|
||
salt = hashed_password[7:36] # 提取盐值部分
|
||
test_hash = hash_password(plain_password)
|
||
return hashed_password == test_hash
|
||
|
||
|
||
def create_user(db: Session, user_data: UserCreate) -> User:
|
||
"""
|
||
创建新用户
|
||
|
||
Args:
|
||
db: Session - 数据库会话对象
|
||
user_data: UserCreate - 用户创建数据模型
|
||
|
||
Returns:
|
||
User - 创建的用户对象
|
||
|
||
Examples:
|
||
>>> from sqlalchemy.orm import Session
|
||
>>> from models import User as DBUser
|
||
>>> db = Session()
|
||
>>> user_data = UserCreate(username="testuser", email="test@example.com", password="secure123")
|
||
>>> new_user = create_user(db, user_data)
|
||
>>> new_user.username
|
||
"testuser"
|
||
|
||
Raises:
|
||
HTTPException:
|
||
- status_code=400: 当用户名或邮箱已存在时
|
||
- status_code=500: 当创建用户失败时
|
||
|
||
Complexity:
|
||
时间复杂度: O(1) - 数据库插入操作
|
||
空间复杂度: O(1)
|
||
"""
|
||
try:
|
||
# 检查用户名是否已存在
|
||
if db.query(DBUser).filter(DBUser.username == user_data.username).count() > 0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="用户名已存在"
|
||
)
|
||
|
||
# 检查邮箱是否已存在
|
||
if db.query(DBUser).filter(DBUser.email == user_data.email).count() > 0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="邮箱已存在"
|
||
)
|
||
|
||
# 创建用户对象
|
||
hashed_pwd = hash_password(user_data.password)
|
||
db_user = DBUser(
|
||
username=user_data.username,
|
||
email=user_data.email,
|
||
password_hash=hashed_pwd,
|
||
role=user_data.role,
|
||
created_at=datetime.utcnow(),
|
||
updated_at=datetime.utcnow(),
|
||
is_active=True
|
||
)
|
||
|
||
# 插入数据库
|
||
db.add(db_user)
|
||
db.commit()
|
||
db.refresh(db_user)
|
||
|
||
return User.from_orm(db_user)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail=f"创建用户失败: {str(e)}"
|
||
)
|
||
|
||
|
||
def get_user_by_id(db: Session, user_id: int) -> Optional[User]:
|
||
"""
|
||
根据ID获取用户
|
||
|
||
Args:
|
||
db: Session - 数据库会话对象
|
||
user_id: int - 用户ID
|
||
|
||
Returns:
|
||
Optional[User] - 用户对象，如果不存在则返回None
|
||
|
||
Examples:
|
||
>>> db = Session()
|
||
>>> user = get_user_by_id(db, 1)
|
||
>>> if user:
|
||
... print(user.username)
|
||
|
||
Raises:
|
||
ValueError: 当user_id小于等于0时抛出
|
||
|
||
Complexity:
|
||
时间复杂度: O(1) - 主键查询
|
||
空间复杂度: O(1)
|
||
"""
|
||
if user_id <= 0:
|
||
raise ValueError("用户ID必须大于0")
|
||
|
||
db_user = db.query(DBUser).filter(DBUser.id == user_id).first()
|
||
if db_user:
|
||
return User.from_orm(db_user)
|
||
return None
|
||
|
||
|
||
def get_user_by_username(db: Session, username: str) -> Optional[User]:
|
||
"""
|
||
根据用户名获取用户
|
||
|
||
Args:
|
||
db: Session - 数据库会话对象
|
||
username: str - 用户名
|
||
|
||
Returns:
|
||
Optional[User] - 用户对象，如果不存在则返回None
|
||
|
||
Examples:
|
||
>>> db = Session()
|
||
>>> user = get_user_by_username(db, "admin")
|
||
>>> if user:
|
||
... print(user.email)
|
||
|
||
Raises:
|
||
ValueError: 当用户名为空时抛出
|
||
|
||
Complexity:
|
||
时间复杂度: O(1) - 假设username有索引
|
||
空间复杂度: O(1)
|
||
"""
|
||
if not username:
|
||
raise ValueError("用户名不能为空")
|
||
|
||
db_user = db.query(DBUser).filter(DBUser.username == username).first()
|
||
if db_user:
|
||
return User.from_orm(db_user)
|
||
return None
|
||
|
||
|
||
def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
|
||
"""
|
||
获取用户列表
|
||
|
||
Args:
|
||
db: Session - 数据库会话对象
|
||
skip: int - 跳过的记录数，默认为0
|
||
limit: int - 返回的最大记录数，默认为100
|
||
|
||
Returns:
|
||
List[User] - 用户对象列表
|
||
|
||
Examples:
|
||
>>> db = Session()
|
||
>>> # 获取前10个用户
|
||
>>> users = get_users(db, limit=10)
|
||
>>> len(users)
|
||
10
|
||
>>>
|
||
>>> # 获取第11-20个用户
|
||
>>> users = get_users(db, skip=10, limit=10)
|
||
|
||
Raises:
|
||
ValueError:
|
||
- 当skip小于0时抛出
|
||
- 当limit小于等于0或大于1000时抛出
|
||
|
||
Complexity:
|
||
时间复杂度: O(n)，其中n是limit值
|
||
空间复杂度: O(n)，存储返回的用户列表
|
||
"""
|
||
if skip < 0:
|
||
raise ValueError("跳过的记录数不能小于0")
|
||
if limit <= 0 or limit > 1000:
|
||
raise ValueError("返回的记录数必须在1-1000之间")
|
||
|
||
db_users = db.query(DBUser).offset(skip).limit(limit).all()
|
||
return [User.from_orm(user) for user in db_users]
|
||
|
||
|
||
def update_user(db: Session, user_id: int, user_update: Dict[str, Union[str, bool]]) -> Optional[User]:
|
||
"""
|
||
更新用户信息
|
||
|
||
Args:
|
||
db: Session - 数据库会话对象
|
||
user_id: int - 用户ID
|
||
user_update: Dict[str, Union[str, bool]] - 更新的字段和值
|
||
|
||
Returns:
|
||
Optional[User] - 更新后的用户对象，如果用户不存在则返回None
|
||
|
||
Examples:
|
||
>>> db = Session()
|
||
>>> # 更新用户邮箱
|
||
>>> user = update_user(db, 1, {"email": "newemail@example.com"})
|
||
>>> if user:
|
||
... print(user.email)
|
||
"newemail@example.com"
|
||
>>>
|
||
>>> # 更新用户状态
|
||
>>> user = update_user(db, 1, {"is_active": False})
|
||
|
||
Raises:
|
||
HTTPException:
|
||
- status_code=400: 当更新的邮箱已被其他用户使用时
|
||
- status_code=400: 当更新的用户名已被其他用户使用时
|
||
ValueError:
|
||
- 当user_id小于等于0时抛出
|
||
- 当更新内容为空时抛出
|
||
|
||
Complexity:
|
||
时间复杂度: O(1) - 数据库更新操作
|
||
空间复杂度: O(1)
|
||
"""
|
||
if user_id <= 0:
|
||
raise ValueError("用户ID必须大于0")
|
||
if not user_update:
|
||
raise ValueError("更新内容不能为空")
|
||
|
||
try:
|
||
# 获取用户
|
||
db_user = db.query(DBUser).filter(DBUser.id == user_id).first()
|
||
if not db_user:
|
||
return None
|
||
|
||
# 检查更新内容
|
||
if "email" in user_update and user_update["email"] != db_user.email:
|
||
if db.query(DBUser).filter(DBUser.email == user_update["email"],
|
||
DBUser.id != user_id).count() > 0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="邮箱已被使用"
|
||
)
|
||
|
||
if "username" in user_update and user_update["username"] != db_user.username:
|
||
if db.query(DBUser).filter(DBUser.username == user_update["username"],
|
||
DBUser.id != user_id).count() > 0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="用户名已被使用"
|
||
)
|
||
|
||
# 更新密码（特殊处理）
|
||
if "password" in user_update:
|
||
user_update["password_hash"] = hash_password(user_update.pop("password"))
|
||
|
||
# 更新用户信息
|
||
user_update["updated_at"] = datetime.utcnow()
|
||
for field, value in user_update.items():
|
||
setattr(db_user, field, value)
|
||
|
||
db.commit()
|
||
db.refresh(db_user)
|
||
|
||
return User.from_orm(db_user)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail=f"更新用户失败: {str(e)}"
|
||
)
|
||
|
||
|
||
def delete_user(db: Session, user_id: int) -> bool:
|
||
"""
|
||
删除用户
|
||
|
||
Args:
|
||
db: Session - 数据库会话对象
|
||
user_id: int - 用户ID
|
||
|
||
Returns:
|
||
bool - 用户是否被成功删除
|
||
|
||
Examples:
|
||
>>> db = Session()
|
||
>>> # 删除用户ID为5的用户
|
||
>>> result = delete_user(db, 5)
|
||
>>> print(result)
|
||
True
|
||
|
||
Raises:
|
||
ValueError: 当user_id小于等于0时抛出
|
||
|
||
Complexity:
|
||
时间复杂度: O(1) - 数据库删除操作
|
||
空间复杂度: O(1)
|
||
"""
|
||
if user_id <= 0:
|
||
raise ValueError("用户ID必须大于0")
|
||
|
||
try:
|
||
# 查找并删除用户
|
||
db_user = db.query(DBUser).filter(DBUser.id == user_id).first()
|
||
if not db_user:
|
||
return False
|
||
|
||
db.delete(db_user)
|
||
db.commit()
|
||
return True
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail=f"删除用户失败: {str(e)}"
|
||
)
|
||
|
||
|
||
# 仅在直接运行时执行测试代码
|
||
if __name__ == "__main__":
|
||
print("用户管理系统核心功能模块加载成功")
|
||
|
||
# 测试密码哈希功能
|
||
test_password = "secure123"
|
||
hashed = hash_password(test_password)
|
||
print(f"密码哈希测试: {test_password} -> {hashed[:30]}...")
|
||
|
||
# 测试密码验证
|
||
is_valid = verify_password(test_password, hashed)
|
||
print(f"密码验证测试: {is_valid}") |