forked from jiaoly/financial_system
97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
from __future__ import annotations
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..database.orm import UserORM
|
||
|
||
|
||
class UserRepository:
|
||
"""用户仓库：封装 users 表 CRUD，供认证服务调用。"""
|
||
|
||
def __init__(self, session: Session):
|
||
self.session = session
|
||
|
||
def create_user(
|
||
self,
|
||
*,
|
||
username: str,
|
||
password_hash: str,
|
||
role: str,
|
||
display_name: str = "",
|
||
avatar_url: str = "",
|
||
email: str = "",
|
||
phone: str = "",
|
||
department: str = "",
|
||
position_title: str = "",
|
||
is_active: bool = True,
|
||
) -> UserORM:
|
||
user = UserORM(
|
||
username=username,
|
||
password_hash=password_hash,
|
||
role=role,
|
||
display_name=display_name,
|
||
avatar_url=avatar_url,
|
||
email=email,
|
||
phone=phone,
|
||
department=department,
|
||
position_title=position_title,
|
||
is_active=is_active,
|
||
)
|
||
self.session.add(user)
|
||
self.session.commit()
|
||
self.session.refresh(user)
|
||
return user
|
||
|
||
def get_by_username(self, username: str) -> UserORM | None:
|
||
return self.session.query(UserORM).filter(UserORM.username == username).one_or_none()
|
||
|
||
def get_by_id(self, user_id: int) -> UserORM | None:
|
||
return self.session.query(UserORM).filter(UserORM.id == user_id).one_or_none()
|
||
|
||
def list_users(self) -> list[UserORM]:
|
||
return self.session.query(UserORM).order_by(UserORM.id.asc()).all()
|
||
|
||
def has_users(self) -> bool:
|
||
return self.session.query(UserORM.id).first() is not None
|
||
|
||
def update_profile(
|
||
self,
|
||
user_id: int,
|
||
*,
|
||
display_name: str | None = None,
|
||
avatar_url: str | None = None,
|
||
email: str | None = None,
|
||
phone: str | None = None,
|
||
department: str | None = None,
|
||
position_title: str | None = None,
|
||
) -> UserORM:
|
||
user = self.get_by_id(user_id)
|
||
if user is None:
|
||
raise ValueError("用户不存在")
|
||
|
||
values = {
|
||
"display_name": display_name,
|
||
"avatar_url": avatar_url,
|
||
"email": email,
|
||
"phone": phone,
|
||
"department": department,
|
||
"position_title": position_title,
|
||
}
|
||
for field, value in values.items():
|
||
if value is not None:
|
||
setattr(user, field, value)
|
||
|
||
self.session.commit()
|
||
self.session.refresh(user)
|
||
return user
|
||
|
||
def update_password_hash(self, user_id: int, password_hash: str) -> UserORM:
|
||
user = self.get_by_id(user_id)
|
||
if user is None:
|
||
raise ValueError("用户不存在")
|
||
|
||
user.password_hash = password_hash
|
||
self.session.commit()
|
||
self.session.refresh(user)
|
||
return user
|