Spaces:
Sleeping
Sleeping
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") | |
def __str__(self) -> str: | |
return f"{self.transaction_date}, {self.category}, {self.name_description}, {self.amount}, {self.type}" | |
async def create(cls: "type[Transaction]", db: AsyncSession, **kwargs) -> "Transaction": | |
query = sql.insert(cls).values(**kwargs) | |
transactions = await db.execute(query) | |
transaction = transactions.first() | |
await db.commit() | |
return transaction | |
async def bulk_create(cls: "type[Transaction]", db: AsyncSession, transactions: List[TransactionCreate]) -> None: | |
values = [transaction.model_dump() for transaction in transactions] | |
query = sql.insert(cls).values(values) | |
await db.execute(query) | |
await db.commit() | |
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 | |
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 | |
async def get_by_user_between_dates( | |
cls: "type[Transaction]", db: AsyncSession, user_id: int, start_date: datetime, end_date: datetime | |
) -> "List[Transaction]": | |
query = sql.select(cls).where(cls.user_id == user_id).where(cls.transaction_date.between(start_date, end_date)) | |
transactions = await db.scalars(query) | |
return transactions.all() | |