提交
This commit is contained in:
zlwu committed
1 parent
c87732fcd5
commit
058811ae77
107 files changed
+16983
-12758
No files matched your search
+21
-165
@@ -1,177 +1,33 @@
|
||||
"""用户CRUD"""
|
||||
from typing import Optional, List, Tuple
|
||||
from sqlalchemy.orm import Session
|
||||
"""用户 CRUD（V0.3 sys_user）"""
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy import and_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.models import SysUser, SysUserHis
|
||||
from app.core.config import settings
|
||||
from app.models.models import SysUser
|
||||
from app.crud.base import CRUDBase
|
||||
from app.core.security import get_password_hash
|
||||
from app.core.utils import generate_random_password
|
||||
|
||||
# 运营端「重置密码」专用明文：仅服务端使用，禁止由 HTTP 请求传入
|
||||
DEFAULT_ADMIN_RESET_PLAIN_PASSWORD = "111111"
|
||||
|
||||
|
||||
class UserCRUD(CRUDBase[SysUser]):
|
||||
"""用户CRUD操作"""
|
||||
"""用户 CRUD：按租户 + 用户账号查询"""
|
||||
|
||||
def get_by_username(self, db: Session, username: str) -> Optional[SysUser]:
|
||||
"""根据用户名获取用户"""
|
||||
return db.query(SysUser).filter(
|
||||
and_(SysUser.username == username, SysUser.is_del == 0)
|
||||
).first()
|
||||
|
||||
def get_by_phone(self, db: Session, phone: str) -> Optional[SysUser]:
|
||||
"""根据手机号获取用户"""
|
||||
return db.query(SysUser).filter(
|
||||
and_(SysUser.phone == phone, SysUser.is_del == 0)
|
||||
).first()
|
||||
|
||||
def get_by_wx_mini_openid(self, db: Session, openid: str) -> Optional[SysUser]:
|
||||
"""根据微信小程序 openid 获取用户"""
|
||||
if not openid:
|
||||
def get_by_user_no(self, db: Session, user_no: str, tenant_no: Optional[str] = None) -> Optional[SysUser]:
|
||||
tenant = tenant_no or settings.DEFAULT_TENANT_NO
|
||||
un = (user_no or "").strip()
|
||||
if not un:
|
||||
return None
|
||||
return db.query(SysUser).filter(
|
||||
and_(SysUser.wx_mini_openid == openid, SysUser.is_del == 0)
|
||||
).first()
|
||||
|
||||
def get_multi_by_role(self, db: Session, role: str, skip: int = 0, limit: int = 100) -> List[SysUser]:
|
||||
"""根据角色获取用户列表"""
|
||||
return db.query(SysUser).filter(
|
||||
and_(SysUser.role == role, SysUser.is_del == 0)
|
||||
).offset(skip).limit(limit).all()
|
||||
|
||||
def get_multi_by_region(self, db: Session, region_id: str, skip: int = 0, limit: int = 100) -> List[SysUser]:
|
||||
"""根据区域获取用户列表"""
|
||||
return db.query(SysUser).filter(
|
||||
and_(SysUser.region_id == region_id, SysUser.is_del == 0)
|
||||
).offset(skip).limit(limit).all()
|
||||
|
||||
def get_multi_with_filters(
|
||||
self,
|
||||
db: Session,
|
||||
role: str = None,
|
||||
roles: Optional[List[str]] = None,
|
||||
region_id: str = None,
|
||||
keyword: str = None,
|
||||
is_active: bool = None,
|
||||
skip: int = 0,
|
||||
limit: int = 10
|
||||
) -> tuple[List[SysUser], int]:
|
||||
"""多条件查询用户"""
|
||||
query = db.query(SysUser).filter(SysUser.is_del == 0)
|
||||
|
||||
if role:
|
||||
query = query.filter(SysUser.role == role)
|
||||
elif roles:
|
||||
role_values = [str(x).strip() for x in roles if str(x).strip()]
|
||||
if role_values:
|
||||
query = query.filter(SysUser.role.in_(role_values))
|
||||
if region_id:
|
||||
query = query.filter(SysUser.region_id == region_id)
|
||||
if is_active is not None:
|
||||
query = query.filter(SysUser.is_active == is_active)
|
||||
if keyword:
|
||||
query = query.filter(
|
||||
(SysUser.username.like(f"%{keyword}%")) |
|
||||
(SysUser.real_name.like(f"%{keyword}%")) |
|
||||
(SysUser.phone.like(f"%{keyword}%"))
|
||||
return (
|
||||
db.query(SysUser)
|
||||
.filter(
|
||||
and_(
|
||||
SysUser.user_no == un,
|
||||
SysUser.tenant_no == tenant,
|
||||
SysUser.del_flag == 0,
|
||||
)
|
||||
)
|
||||
|
||||
total = query.count()
|
||||
users = query.order_by(SysUser.create_at.desc()).offset(skip).limit(limit).all()
|
||||
return users, total
|
||||
|
||||
def create_with_password(self, db: Session, obj_in: dict, creator: dict = None) -> Tuple[SysUser, Optional[str]]:
|
||||
"""创建用户。
|
||||
|
||||
若请求体包含非空 password（运营端表单），则使用该密码写入 password_hash；
|
||||
否则生成随机密码，并在返回值中给出明文 initialPassword（仅此种情况返回，供管理员告知用户）。
|
||||
"""
|
||||
obj_in = dict(obj_in)
|
||||
raw_pwd = obj_in.pop("password", None)
|
||||
plain: Optional[str] = None
|
||||
if isinstance(raw_pwd, str):
|
||||
plain = raw_pwd.strip()
|
||||
if plain:
|
||||
if len(plain) < 6:
|
||||
raise ValueError("密码至少6位")
|
||||
obj_in["password_hash"] = get_password_hash(plain)
|
||||
initial_reveal: Optional[str] = None
|
||||
else:
|
||||
initial_reveal = generate_random_password()
|
||||
obj_in["password_hash"] = get_password_hash(initial_reveal)
|
||||
obj_in["is_active"] = obj_in.get("is_active", True)
|
||||
user = self.create(db, obj_in, creator)
|
||||
return user, initial_reveal
|
||||
|
||||
def reset_password_to_admin_default(self, db: Session, id: str) -> bool:
|
||||
"""将密码重置为服务端固定默认（DEFAULT_ADMIN_RESET_PLAIN_PASSWORD），不由请求传入。"""
|
||||
user = self.get(db, id)
|
||||
if not user:
|
||||
return False
|
||||
plain = DEFAULT_ADMIN_RESET_PLAIN_PASSWORD
|
||||
if len(plain) < 6:
|
||||
raise ValueError("默认重置密码配置无效：至少6位")
|
||||
user.password_hash = get_password_hash(plain)
|
||||
db.commit()
|
||||
return True
|
||||
|
||||
def update_with_history(self, db: Session, id: str, obj_in: dict, updater: dict = None) -> Optional[SysUser]:
|
||||
"""更新用户并记录历史"""
|
||||
original = self.get(db, id)
|
||||
if not original:
|
||||
return None
|
||||
|
||||
his_data = {
|
||||
"id": original.id,
|
||||
"username": original.username,
|
||||
"password_hash": original.password_hash,
|
||||
"real_name": original.real_name,
|
||||
"phone": original.phone,
|
||||
"avatar_url": getattr(original, "avatar_url", None),
|
||||
"wx_mini_openid": getattr(original, "wx_mini_openid", None),
|
||||
"role": original.role,
|
||||
"region_id": original.region_id,
|
||||
"is_active": original.is_active,
|
||||
"create_at": original.create_at,
|
||||
"create_by": original.create_by,
|
||||
"create_name": original.create_name,
|
||||
"update_at": original.update_at,
|
||||
"update_by": original.update_by,
|
||||
}
|
||||
his = SysUserHis(**his_data)
|
||||
db.add(his)
|
||||
|
||||
return self.update(db, id, obj_in, updater)
|
||||
|
||||
def soft_delete_with_history(self, db: Session, id: str) -> bool:
|
||||
"""删除用户并记录历史"""
|
||||
original = self.get(db, id)
|
||||
if not original:
|
||||
return False
|
||||
|
||||
his_data = {
|
||||
"id": original.id,
|
||||
"username": original.username,
|
||||
"password_hash": original.password_hash,
|
||||
"real_name": original.real_name,
|
||||
"phone": original.phone,
|
||||
"avatar_url": getattr(original, "avatar_url", None),
|
||||
"wx_mini_openid": getattr(original, "wx_mini_openid", None),
|
||||
"role": original.role,
|
||||
"region_id": original.region_id,
|
||||
"is_active": original.is_active,
|
||||
"create_at": original.create_at,
|
||||
"create_by": original.create_by,
|
||||
"create_name": original.create_name,
|
||||
"update_at": original.update_at,
|
||||
"update_by": original.update_by,
|
||||
}
|
||||
his = SysUserHis(**his_data)
|
||||
db.add(his)
|
||||
|
||||
return self.delete(db, id)
|
||||
.first()
|
||||
)
|
||||
|
||||
|
||||
user_crud = UserCRUD(SysUser)
|
||||
Reference in new issue
Block a user