Spaces:
Sleeping
Sleeping
File size: 2,063 Bytes
5753020 d5684b3 5753020 d5684b3 5753020 | 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 | from sqlalchemy.orm import relationship, Mapped, mapped_column
from sqlalchemy.sql import expression as sql
from sqlalchemy.ext.asyncio import AsyncSession
from app.model.base import BaseModel
from app.engine.postgresdb import Base
class User(Base, BaseModel):
__tablename__ = "users"
name: Mapped[str]
email: Mapped[str]
hashed_password: Mapped[str]
is_deleted: Mapped[bool] = mapped_column(default=False)
transactions = relationship("Transaction", back_populates="user")
@classmethod
async def create(cls: "type[User]", db: AsyncSession, **kwargs) -> "User":
query = sql.insert(cls).values(**kwargs).returning(cls.id)
users = await db.execute(query)
await db.commit()
return users.first()
@classmethod
async def update(cls: "type[User]", db: AsyncSession, id: int, **kwargs) -> "User":
query = (
sql.update(cls)
.where(cls.id == id)
.values(**kwargs)
.execution_options(synchronize_session="fetch")
.returning(cls.id)
)
users = await db.execute(query)
await db.commit()
return users.first()
@classmethod
async def get(cls: "type[User]", db: AsyncSession, id: int) -> "User":
query = sql.select(cls).where(cls.id == id)
users = await db.execute(query)
(user,) = users.first()
return user
@classmethod
async def get(cls: "type[User]", db: AsyncSession, email: str) -> "User":
query = sql.select(cls).where(cls.email == email)
users = await db.execute(query)
(user,) = users.first()
return user
@classmethod
async def delete(cls: "type[User]", db: AsyncSession, id: int) -> "User":
query = (
sql.update(cls)
.where(cls.id == id)
.values(is_deleted=True)
.execution_options(synchronize_session="fetch")
.returning(cls.id)
)
users = await db.execute(query)
await db.commit()
return users.first()
|