HR-ATS-Portal/backend/users/models.py

import uuid
from datetime import datetime
from typing import TYPE_CHECKING,List,Optional

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
from job.job_post.models import JobPosts

if TYPE_CHECKING:  # runtime import would cycle: inbox.models imports this module
    from inbox.models import Inbox
    from job.candidate.models import Feedback, Notes

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",
    sa_relationship_kwargs={"lazy": "selectin"}
    )
    # selectin, not joined: this is a one-to-many, so a joined load would repeat the
    # user row once per post. Without an explicit strategy the default is a lazy load,
    # which raises MissingGreenlet the moment anything touches it under asyncio.
    job_posts: List[JobPosts] = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    inbox: List["Inbox"] = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    feedback: List["Feedback"] = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    notes: List["Notes"] = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"lazy": "selectin", "foreign_keys": "[Notes.user_id]"},
    )
    authored_notes: List["Notes"] = Relationship(
        back_populates="author",
        sa_relationship_kwargs={"lazy": "selectin", "foreign_keys": "[Notes.created_by]"},
    )

    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
    async def get_user_id(cls, session: AsyncSession, user_id: str):
        uid = cls._as_uuid(user_id)
        if uid is None:
            return None
        statement = select(cls).where(cls.id == uid)
        result = await session.execute(statement)
        return result.scalars().first()

    @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


import job.candidate.models as _candidate_models  # noqa: E402, F401