agent-platform/services/auth-service/app/domain/repositories.py

from datetime import datetime

from core_domain import RoleAssignmentStatus, RoleStatus, UserStatus
from core_shared import JSONValue
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.db.models import Role, RoleAssignment, User


class UserRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def create(
        self,
        *,
        username: str,
        password_hash: str,
        display_name: str | None,
        email: str | None,
        metadata_json: dict[str, JSONValue]) -> User:
        entity = User(
            username=username,
            password_hash=password_hash,
            display_name=display_name,
            email=email,
            metadata_json=metadata_json)
        self.db.add(entity)
        self.db.commit()
        self.db.refresh(entity)
        return entity

    def list_all(self) -> list[User]:
        stmt = select(User).order_by(User.created_time.desc())
        return list(self.db.scalars(stmt))

    def get_by_id(self, *, user_id: str) -> User | None:
        return self.db.get(User, user_id)

    def get_by_username(self, *, username: str) -> User | None:
        stmt = select(User).where(User.username == username)
        return self.db.scalar(stmt)

    def touch_last_login_time(self, *, user_id: str) -> None:
        entity = self.db.get(User, user_id)
        if entity is None:
            return
        entity.last_login_time = datetime.utcnow()
        self.db.commit()
        self.db.refresh(entity)

    def update_status(self, *, user_id: str, status: UserStatus) -> User | None:
        entity = self.get_by_id(user_id=user_id)
        if entity is None:
            return None
        entity.status = status
        self.db.commit()
        self.db.refresh(entity)
        return entity


class RoleRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def create(
        self,
        *,
        code: str,
        name: str,
        description: str | None,
        permissions_json: list[str]) -> Role:
        entity = Role(
            code=code,
            name=name,
            description=description,
            permissions_json=permissions_json)
        self.db.add(entity)
        self.db.commit()
        self.db.refresh(entity)
        return entity

    def list_all(self) -> list[Role]:
        stmt = select(Role).order_by(Role.created_time.desc())
        return list(self.db.scalars(stmt))

    def get_by_id(self, *, role_id: str) -> Role | None:
        return self.db.get(Role, role_id)

    def update_status(self, *, role_id: str, status: RoleStatus) -> Role | None:
        entity = self.get_by_id(role_id=role_id)
        if entity is None:
            return None
        entity.status = status
        self.db.commit()
        self.db.refresh(entity)
        return entity


class RoleAssignmentRepository:
    def __init__(self, db: Session) -> None:
        self.db = db

    def create(
        self,
        *,
        user_id: str,
        role_id: str,
        scope_type: str | None,
        scope_id: str | None,
        expires_time: datetime | None) -> RoleAssignment:
        entity = RoleAssignment(
            user_id=user_id,
            role_id=role_id,
            scope_type=scope_type,
            scope_id=scope_id,
            expires_time=expires_time)
        self.db.add(entity)
        self.db.commit()
        self.db.refresh(entity)
        return entity

    def list_by_user(self, *, user_id: str) -> list[RoleAssignment]:
        stmt = (
            select(RoleAssignment)
            .where(RoleAssignment.user_id == user_id)
            .order_by(RoleAssignment.created_time.desc())
        )
        return list(self.db.scalars(stmt))

    def get_by_id(
        self,
        *,
        assignment_id: str) -> RoleAssignment | None:
        return self.db.get(RoleAssignment, assignment_id)

    def update_status(
        self,
        *,
        assignment_id: str,
        status: RoleAssignmentStatus) -> RoleAssignment | None:
        entity = self.get_by_id(assignment_id=assignment_id)
        if entity is None:
            return None
        entity.status = status
        self.db.commit()
        self.db.refresh(entity)
        return entity