from __future__ import annotations from typing import TYPE_CHECKING from fastapi import Depends from sqlalchemy import Boolean, Integer, String from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select from sqlalchemy.orm import Mapped, mapped_column, relationship, selectinload from app.utils.db import Base, get_async_session if TYPE_CHECKING: from .integrations import UserIntegration from .profile import Profile class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True) username: Mapped[str] = mapped_column(String(20), unique=True, nullable=False) password: Mapped[str | None] = mapped_column(String(255), nullable=True) email: Mapped[str] = mapped_column(String(120), unique=True, nullable=False) google_id: Mapped[str | None] = mapped_column( String(255), unique=True, nullable=True ) description: Mapped[str | None] = mapped_column(String(250), nullable=True) premium: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) is_banned: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) is_moderator: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) token_version: Mapped[int] = mapped_column(Integer, default=0, nullable=False) profile: Mapped["Profile"] = relationship( "Profile", back_populates="user", uselist=False, lazy="selectin" ) integrations: Mapped["UserIntegration"] = relationship( "UserIntegration", back_populates="user", uselist=False, lazy="selectin" ) @classmethod async def get_user_by_email( cls, email: str, session: AsyncSession = Depends(get_async_session) ): result = await session.execute(select(cls).where(cls.email == email)) return result.scalars().first() @classmethod async def get_user_by_username( cls, username: str, session: AsyncSession = Depends(get_async_session) ): result = await session.execute( select(cls) .options( selectinload(cls.profile), selectinload(cls.integrations), ) .where(cls.username == username) ) return result.scalars().first() @classmethod async def get_user_by_id( cls, user_id: int, session: AsyncSession = Depends(get_async_session) ): result = await session.execute(select(cls).where(cls.id == user_id)) return result.scalars().first() @property def has_password(self) -> bool: return bool(self.password)