"""用户与角色数据访问。 本层只 flush 不 commit —— 事务边界由 services 层控制。 """ from datetime import datetime from sqlalchemy import select from sqlalchemy.orm import Session from app.models import User def get_by_username(session: Session, username: str) -> User | None: """按用户名查询用户。不存在返回 None。""" return session.execute(select(User).where(User.username == username)).scalar_one_or_none() def get_by_id(session: Session, user_id: int) -> User | None: """按主键查询用户。不存在返回 None。""" return session.get(User, user_id) def get_role_codes(user: User) -> list[str]: """取用户的角色编码列表。 依赖 `User.user_roles` 的 `lazy="selectin"` 预加载(见模型定义), 故此处遍历不会触发 N+1 查询。 """ return [link.role.role_code for link in user.user_roles if link.role is not None] def touch_last_login(session: Session, user: User, moment: datetime) -> None: """更新最后登录时间。只 flush,提交由 service 决定。""" user.last_login_time = moment session.flush()