app/model/transaction.py

from datetime import datetime
from typing import List
from sqlalchemy import ForeignKey
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
from app.schema.index import TransactionCreate


class Transaction(Base, BaseModel):
    __tablename__ = "transactions"
    transaction_date: Mapped[datetime]
    category: Mapped[str]
    name_description: Mapped[str]
    amount: Mapped[float]
    type: Mapped[str]

    user_id = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="transactions")

    @classmethod
    async def create(cls: "type[Transaction]", db: AsyncSession, **kwargs) -> "Transaction":
        query = sql.insert(cls).values(**kwargs).returning(cls.id)
        transactions = await db.execute(query)
        transaction = transactions.first()
        await db.commit()
        return transaction

    @classmethod
    async def bulk_create(cls: "type[Transaction]", db: AsyncSession, transactions: List[TransactionCreate]) -> None:
        query = sql.insert(cls).values(transactions)
        await db.execute(query)
        await db.commit()

    @classmethod
    async def update(cls: "type[Transaction]", db: AsyncSession, id: int, **kwargs) -> "Transaction":
        query = sql.update(cls).where(cls.id == id).values(**kwargs).execution_options(synchronize_session="fetch")
        transactions = await db.scalars(query)
        transaction = transactions.first()
        await db.commit()
        return transaction

    @classmethod
    async def get_by_user(cls: "type[Transaction]", db: AsyncSession, user_id: int) -> "List[Transaction]":
        query = sql.select(cls).where(cls.user_id == user_id)
        transactions = await db.scalars(query)
        return transactions