app/model/user.py

import logging
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

logger = logging.getLogger(__name__)


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":
        logging.info(f"Creating user: {kwargs}")
        query = sql.insert(cls).values(**kwargs)
        users = await db.scalars(query)
        user = users.first()
        logging.info(f"User created: {users.first()}")
        await db.commit()
        return user

    @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")
        users = await db.scalars(query)
        user = users.first()
        await db.commit()
        return user

    @classmethod
    async def get(cls: "type[User]", db: AsyncSession, email: str) -> "User":
        logging.info(f"Getting user: {email}")
        query = sql.select(cls).where(cls.email == email)
        logging.info(f"Query: {query}")
        users = await db.scalars(query)
        logging.info(f"Users: {users}")
        return users.first()

    @classmethod
    async def delete(cls: "type[User]", db: AsyncSession, email: str) -> "User":
        query = (
            sql.update(cls)
            .where(cls.email == email)
            .values(is_deleted=True)
            .execution_options(synchronize_session="fetch")
        )
        users = await db.scalars(query)
        user = users.first()
        await db.commit()
        return user