Spaces:
Running
Running
File size: 2,041 Bytes
48ec4db |
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 |
from app.backend.controllers.base_controller import engine
from sqlalchemy.ext.asyncio import AsyncSession
from app.backend.models.base_model import Base
from app.backend.models.chats import Chat
from sqlalchemy.orm import relationship
from sqlalchemy import Column, String
from sqlalchemy.future import select
class User(Base):
'''
Base model for users table
'''
__tablename__ = "users"
id = Column("id", String, primary_key=True, unique=True)
language = Column("language", String, default="English", nullable=False)
theme = Column("theme", String, default="light", nullable=False)
chats = relationship("Chat", back_populates="user", lazy="selectin")
async def add_new_user(id: str) -> User:
async with AsyncSession(engine, expire_on_commit=False) as db:
new_user = User(id=id)
db.add(new_user)
await db.commit()
return new_user
async def find_user_by_id(id: str) -> User | None:
async with AsyncSession(engine) as db:
result = await db.execute(select(User).where(User.id == id))
return result.scalar_one_or_none()
async def update_user(user: User, language: str = None, theme: str = None) -> None:
async with AsyncSession(engine) as db:
user = await db.merge(user)
if language:
user.language = language
if theme:
user.theme = theme
await db.commit()
async def get_user_chats(id: str) -> list[Chat]:
async with AsyncSession(autoflush=False, bind=engine) as db:
result = await db.execute(
select(Chat).filter(Chat.user_id == id)
)
return result.scalars().all()
async def get_user_last_chat(user: User) -> Chat | None:
if user is None:
return None
async with AsyncSession(engine) as db:
result = await db.execute(
select(Chat)
.where(Chat.user_id == user.id)
.order_by(Chat.created_at.desc())
.limit(1)
)
chat = result.scalars().first()
return chat
|