1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
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
)
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)
|
