📦 chore(user/crud.py): add CRUD operations for User model in database

🔒 fix(user/crud.py): add validation to prevent duplicate usernames during user update
🔄 refactor(user/crud.py): update user attributes and last_login_at timestamp during user update
🐛 fix(user/crud.py): handle IntegrityError during user update and rollback transaction

src/backend/langflow/services/database/models/user/crud.py

@@ -0,0 +1,53 @@
+from datetime import datetime, timezone
+from typing import Union
+from uuid import UUID
+from fastapi import Depends, HTTPException
+from langflow.services.database.models.user.user import User, UserUpdate
+from langflow.services.utils import get_session
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import Session
+
+
+from sqlalchemy.orm.attributes import flag_modified
+
+
+def get_user_by_username(db: Session, username: str) -> Union[User, None]:
+    return db.query(User).filter(User.username == username).first()
+
+
+def get_user_by_id(db: Session, id: UUID) -> Union[User, None]:
+    return db.query(User).filter(User.id == id).first()
+
+
+def update_user(
+    user_id: UUID, user: UserUpdate, db: Session = Depends(get_session)
+) -> User:
+    user_db = get_user_by_id(db, user_id)
+    if not user_db:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    user_db_by_username = get_user_by_username(db, user.username)  # type: ignore
+    if user_db_by_username and user_db_by_username.id != user_id:
+        raise HTTPException(status_code=409, detail="Username already exists")
+
+    user_data = user.dict(exclude_unset=True)
+    for attr, value in user_data.items():
+        if hasattr(user_db, attr) and value is not None:
+            setattr(user_db, attr, value)
+
+    user_db.updated_at = datetime.now(timezone.utc)
+    flag_modified(user_db, "updated_at")
+
+    try:
+        db.commit()
+    except IntegrityError as e:
+        db.rollback()
+        raise HTTPException(status_code=400, detail=str(e)) from e
+
+    return user_db
+
+
+def update_user_last_login_at(user_id: UUID, db: Session = Depends(get_session)):
+    user_data = UserUpdate(last_login_at=datetime.now(timezone.utc))  # type: ignore
+
+    return update_user(user_id, user_data, db)