Change userManager to modern sqlmodel syntax

app/controllers/userManager.py

@@ -1,53 +1,54 @@
-from fastapi import APIRouter, Depends, HTTPException
-from sqlmodel import Session
+from fastapi import APIRouter, HTTPException, Depends
+from sqlmodel import Session, select
 from typing import List
 
-from ..model.models import UserResponse, UserCreate, UserDB
+from ..model.models import UserResponse, UserCreate, UserDB, UserUpdate
 from ..services.database import engine
 
 user_router = APIRouter(tags=["users"])
 
-@user_router.post("/users/", response_model=UserResponse)
-def create_user(user: UserCreate):
+def get_session():
     with Session(engine) as db:
-        db_user = User(**user.dict())
-        db.add(db_user)
-        db.commit()
-        db.refresh(db_user)
-        return db_user
+        yield db
+
+@user_router.post("/users/", response_model=UserResponse)
+def create_user(*, db: Session = Depends(get_session), user: UserCreate):
+    print("creating user with data ", user)
+    db_user = UserDB.model_validate(user)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
 
 @user_router.get("/users/", response_model=List[UserResponse])
-def read_users():
-    with Session(engine) as db:
-        users = db.query(User).all()
-        return users
+def read_users(*, db: Session = Depends(get_session)):
+    users = db.exec(select(UserDB)).all()
+    return users
 
 @user_router.get("/users/{user_id}", response_model=UserResponse)
-def read_user(user_id: int):
-    with Session(engine) as db:
-        db_user = db.query(User).filter(User.id == user_id).first()
-        if db_user is None:
-            raise HTTPException(status_code=404, detail="User not found")
-        return db_user
+def read_user(*, db: Session = Depends(get_session), user_id: int):
+    db_user = db.get(UserDB, user_id)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    return db_user
 
-@user_router.put("/users/{user_id}", response_model=UserResponse)
-def update_user(user_id: int, user: UserCreate):
-    with Session(engine) as db:
-        db_user = db.query(User).filter(User.id == user_id).first()
-        if db_user is None:
-            raise HTTPException(status_code=404, detail="User not found")
-        for key, value in user.dict().items():
-            setattr(db_user, key, value)
-        db.commit()
-        db.refresh(db_user)
-        return db_user
+@user_router.patch("/users/{user_id}", response_model=UserResponse)
+def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserUpdate):
+    db_user = db.get(UserDB, user_id)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    user_data = user.model_dump(exclude_unset=True)
+    db_user.sqlmodel_update(user_data)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
 
 @user_router.delete("/users/{user_id}")
-def delete_user(user_id: int):
-    with Session(engine) as db:
-        db_user = db.query(User).filter(User.id == user_id).first()
-        if db_user is None:
-            raise HTTPException(status_code=404, detail="User not found")
-        db.delete(db_user)
-        db.commit()
-        return {"message": "User deleted successfully"}
+def delete_user(*, db: Session = Depends(get_session), user_id: int):
+    db_user = db.get(UserDB, user_id)
+    if db_user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    db.delete(db_user)
+    db.commit()
+    return {"message": "User deleted successfully"}

app/model/models.py

@@ -20,35 +20,40 @@ class UserDB(UserBase, table=True):
     id: int | None = Field(default=None, primary_key=True)
     password: str
 
+class UserUpdate(Base):
+    name: str | None = None
+    email: str | None = None
+    is_admin: bool | None = None
+    password: str | None = None
+
 #### Special
 class AaGroupLink(Base, table=True):
-    group_id: int | None = Field(default=None, foreign_key="group.id", primary_key=True)
-    accessauth_id: int | None = Field(default=None, foreign_key="accessauthorization.id", primary_key=True)
+    group_id: int | None = Field(default=None, foreign_key="groupdb.id", primary_key=True)
+    accessauth_id: int | None = Field(default=None, foreign_key="accessauthorizationdb.id", primary_key=True)
 
 #### Group
 class GroupBase(Base):
     name: str = Field(index=True)
-    
 
 class GroupDB(GroupBase, table=True):
     id: int | None = Field(default=None, primary_key=True)
     cards: List["Card"] = Relationship(back_populates="group")
-    accessauths: List["AccessAuthorizationBase"] = Relationship(back_populates="groups", link_model=AaGroupLink)
+    accessauths: List["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink)
 
 class GroupResponse(GroupBase):
     id: int
     cards: List["Card"]
-    accessauths: List["AccessAuthorizationBase"]
+    accessauths: List["AccessAuthorizationDB"]
 
 #### AccessAuthorization
 class AccessAuthorizationBase(Base):
     name: str = Field(index=True)
     is_active: bool
-    groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
-    timetables: List["Timetable"] = Relationship(back_populates="accessauth")
 
 class AccessAuthorizationDB(Base, table=True):
     id: int | None = Field(default=None, primary_key=True)
+    groups: List["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink)
+    timetables: List["Timetable"] = Relationship(back_populates="accessauth")
 
 class AccessAuthorizationCreate(AccessAuthorizationBase):
     pass