85 lines
2.2 KiB
Python
85 lines
2.2 KiB
Python
# 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)
|