jackyliu
c7f8e69d61
- Implemented Pydantic models for Kubernetes cluster state representation in `cluster.py`. - Created a `Resource` class for converting JSON/dict to Python objects in `resource.py`. - Established user models and services for user management, including password hashing and JWT token generation. - Developed application orchestration services for managing Kubernetes applications, including installation and uninstallation. - Added cluster service for retrieving cluster status and health reports. - Introduced node service for fetching node resource details and health status. - Implemented user service for handling user authentication and management.
80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel, constr, EmailStr, validator
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ocdp.database import get_db
|
|
from ocdp.services.user import user_service
|
|
from ocdp.services.user import user_exceptions
|
|
|
|
# 创建一个 API 路由器
|
|
router = APIRouter()
|
|
|
|
ALLOWED_PASSWORD_CHARS = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789@#$%&*_-")
|
|
class RegisterUserRequest(BaseModel):
|
|
"""
|
|
用户注册的请求体 (Request Body)
|
|
"""
|
|
username: str
|
|
password: str
|
|
email: EmailStr
|
|
|
|
@validator('password')
|
|
def validate_password(cls, v):
|
|
if len(v) < 8 or len(v) > 32:
|
|
raise ValueError('密码长度应在8~32位')
|
|
if not any(c.isalpha() for c in v):
|
|
raise ValueError('密码必须包含字母')
|
|
if not any(c.isdigit() for c in v):
|
|
raise ValueError('密码必须包含数字')
|
|
if any(c.isspace() for c in v):
|
|
raise ValueError('密码不能包含空格')
|
|
if any(c not in ALLOWED_PASSWORD_CHARS for c in v):
|
|
raise ValueError('密码包含非法字符')
|
|
return v
|
|
|
|
class RegisterUserResponse(BaseModel):
|
|
"""
|
|
成功注册后返回的用户信息 (Response Body)
|
|
不包含密码等敏感数据
|
|
"""
|
|
id: int
|
|
username: str
|
|
email: EmailStr
|
|
|
|
class Config:
|
|
# Pydantic V2 推荐的用法
|
|
from_attributes = True
|
|
# Pydantic V1 的旧用法
|
|
# orm_mode = True
|
|
|
|
@router.post("/", response_model=RegisterUserResponse, status_code=status.HTTP_201_CREATED)
|
|
def register_user(
|
|
user_in: RegisterUserRequest,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
注册一个新用户.
|
|
|
|
- **username**: 用户的唯一名称.
|
|
- **email**: 用户的唯一邮箱.
|
|
- **password**: 用户的密码.
|
|
"""
|
|
try:
|
|
# 调用 service 层的函数来创建用户
|
|
# user_in.dict() 将 Pydantic 模型转换为字典
|
|
created_user = user_service.create_user(
|
|
username=user_in.username,
|
|
password=user_in.password,
|
|
email=user_in.email,
|
|
db=db
|
|
)
|
|
return created_user
|
|
except user_exceptions.UserAlreadyExistsError as e:
|
|
# 捕获 service 层抛出的特定异常
|
|
# 返回 400 错误，并附带清晰的错误信息
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e),
|
|
)
|