# from sqlmodel import SQLModel, Field from sqlalchemy import Column, String, DateTime, delete from sqlalchemy import update as sqlalchemy_update, delete as sqlalchemy_delete from sqlalchemy.future import select from app.db import Base, db from datetime import datetime from uuid import uuid4 class User(Base): __tablename__ = "users" id = Column(String, primary_key=True) username = Column(String) created_at = Column(DateTime, index=True, default=datetime.utcnow) def __repr__(self): return ( f"<{self.__class__.__name__}(" f"id={self.id}, " f"username={self.username}, " f")>" ) @classmethod async def create(cls, **kwargs): user = cls(id=str(uuid4()), **kwargs) db.add(user) try: await db.commit() except Exception: await db.rollback() raise return user @classmethod async def update(cls, id, **kwargs): query = ( sqlalchemy_update(cls) .where(cls.id == id) .values(**kwargs) .execution_options(synchronize_session="fetch") ) await db.execute(query) try: await db.commit() except Exception: await db.rollback() raise return await cls.get(id) @classmethod async def get(cls, id): query = select(cls).where(cls.id == id) users = await db.execute(query) (user,) = users.first() return user @classmethod async def get_all(cls): query = select(cls) users = await db.execute(query) users = users.scalars().all() return users @classmethod async def delete(cls, id): query = sqlalchemy_delete(cls).where(cls.id == id) await db.execute(query) try: await db.commit() except Exception: await db.rollback() raise return True # class UserBase(SQLModel): # username: str # full_name: str # email: str # hashed_password: str # disabled: bool # class User(UserBase, table=True): # id: int = Field(default=None, primary_key=True)