Spaces:
Sleeping
Sleeping
File size: 3,822 Bytes
81ec8d6 f92858b d5684b3 c553fd3 d5684b3 81ec8d6 d5684b3 f92858b d5684b3 a4ef0c7 f92858b c553fd3 f92858b c553fd3 81ec8d6 f92858b c553fd3 81ec8d6 c553fd3 f92858b c553fd3 81ec8d6 f92858b 81ec8d6 f92858b 81ec8d6 c553fd3 f92858b 81ec8d6 f92858b c553fd3 f92858b 81ec8d6 |
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 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
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
from app.schema.index import UserCreate, UserUpdate
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")
income_statements = relationship("IncomeStatement", back_populates="user")
@classmethod
async def create(cls: "type[User]", db: AsyncSession, **kwargs: UserCreate) -> "User":
"""
Creates a new user in the database.
Args:
cls (type[User]): The class object representing the User model.
db (AsyncSession): The asynchronous session object for interacting with the database.
**kwargs (UserCreate): The keyword arguments representing the user's attributes.
Returns:
User
"""
print(f"Creating user: {kwargs}")
# query = sql.insert(cls).values(**kwargs)
user = cls(name=kwargs["name"], email=kwargs["email"], hashed_password=kwargs["hashed_password"])
db.add(user)
await db.commit()
await db.refresh(user)
return user
@classmethod
async def update(cls: "type[User]", db: AsyncSession, id: int, **kwargs: UserUpdate) -> "User":
"""
Updates a user in the database with the given ID and keyword arguments.
Args:
cls (type[User]): The class object representing the User model.
db (AsyncSession): The asynchronous session object for interacting with the database.
id (int): The ID of the user to update.
**kwargs (UserUpdate): The keyword arguments representing the user's attributes to update.
Returns:
User
"""
query = sql.update(cls).where(cls.id == id).values(**kwargs).execution_options(synchronize_session="fetch")
results = await db.execute(query)
user = results.fetchone()
await db.commit()
return user
@classmethod
async def get(cls: "type[User]", db: AsyncSession, email: str) -> "User":
"""
Retrieves a user from the database based on their email.
Args:
cls (type[User]): The class object representing the User model.
db (AsyncSession): The asynchronous session object for interacting with the database.
email (str): The email of the user to retrieve.
Returns:
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":
"""
Deletes a user from the database based on their email.
Args:
cls (type[User]): The class object representing the User model.
db (AsyncSession): The asynchronous session object for interacting with the database.
email (str): The email of the user to delete.
Returns:
User
"""
query = (
sql.update(cls)
.where(cls.email == email)
.values(is_deleted=True)
.execution_options(synchronize_session="fetch")
)
result = await db.execute(query)
user = result.fetchone()
await db.commit()
return user
|