Dmitry Chumak
b3c5c032cf
worked on async alembic migrations, User model. Current issue: web_1 | INFO: 172.20.0.1:62958 - "POST /users/ HTTP/1.1" 500 Internal Server Error web_1 | ERROR: Exception in ASGI application web_1 | Traceback (most recent call last): ... web_1 | response web_1 | value is not a valid dict (type=type_error.dict) when trying to create new user
84 lines
2.1 KiB
Python
84 lines
2.1 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
|
|
|
|
@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)
|