120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlmodel import Field, Relationship, SQLModel, select
|
|
|
|
from role.models import Roles
|
|
|
|
|
|
class Users(SQLModel, table=True):
|
|
__tablename__ = "users"
|
|
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
name: str
|
|
email: str = Field(unique=True)
|
|
role_id: int | None = Field(nullable=True, foreign_key="roles.id")
|
|
role: Roles | None = Relationship(back_populates="users")
|
|
password: str
|
|
created_at: datetime = Field(default_factory=datetime.now)
|
|
updated_at: datetime = Field(default_factory=datetime.now)
|
|
is_active: bool = Field(default=False)
|
|
is_deleted: bool = Field(default=False)
|
|
|
|
@classmethod
|
|
def _search_filter(cls, search: str):
|
|
pattern = f"%{search}%"
|
|
return or_(
|
|
cls.name.ilike(pattern),
|
|
cls.email.ilike(pattern),
|
|
)
|
|
|
|
@staticmethod
|
|
def _as_uuid(record_id: str) -> uuid.UUID | None:
|
|
try:
|
|
return uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
|
|
@classmethod
|
|
async def get_users(
|
|
cls, session: AsyncSession, top: int | None, skip: int, search: str | None
|
|
):
|
|
statement = (
|
|
select(cls)
|
|
.options(selectinload(cls.role))
|
|
.where(cls.is_deleted == False)
|
|
.order_by(cls.created_at.desc())
|
|
)
|
|
if search:
|
|
statement = statement.where(cls._search_filter(search))
|
|
if skip:
|
|
statement = statement.offset(skip)
|
|
if top is not None:
|
|
statement = statement.limit(top)
|
|
result = await session.execute(statement)
|
|
return result.scalars().all()
|
|
|
|
@classmethod
|
|
async def get_user_by_id(cls, session: AsyncSession, record_id: str):
|
|
uid = cls._as_uuid(record_id)
|
|
if uid is None:
|
|
return None
|
|
statement = select(cls).options(selectinload(cls.role)).where(cls.id == uid)
|
|
result = await session.execute(statement)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_user_by_email(cls, session: AsyncSession, email: str):
|
|
statement = select(cls).options(selectinload(cls.role)).where(cls.email == email)
|
|
result = await session.execute(statement)
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def count_users(cls, session: AsyncSession, search: str | None):
|
|
statement = (
|
|
select(func.count())
|
|
.select_from(cls)
|
|
.where(cls.is_deleted == False) # noqa: E712
|
|
)
|
|
if search:
|
|
statement = statement.where(cls._search_filter(search))
|
|
result = await session.execute(statement)
|
|
return result.scalar_one()
|
|
|
|
@classmethod
|
|
async def insert_user(cls, session: AsyncSession, fields: dict):
|
|
"""`fields["password"]` is expected to be hashed already — see users.plugins."""
|
|
user = cls(**fields)
|
|
session.add(user)
|
|
await session.commit()
|
|
return await cls.get_user_by_id(session, user.id)
|
|
|
|
@classmethod
|
|
async def update_user(cls, session: AsyncSession, record_id: str, fields: dict):
|
|
user = await cls.get_user_by_id(session, record_id)
|
|
if not user:
|
|
return None
|
|
for key, value in fields.items():
|
|
setattr(user, key, value)
|
|
user.updated_at = datetime.now()
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return await cls.get_user_by_id(session, user.id)
|
|
|
|
@classmethod
|
|
async def soft_delete_user(cls, session: AsyncSession, record_id: str):
|
|
user = await cls.get_user_by_id(session, record_id)
|
|
if not user:
|
|
return None
|
|
user.is_deleted = True
|
|
user.is_active = False
|
|
user.updated_at = datetime.now()
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return user
|