Add user and team models with auth routes and admin endpoints

src/models/team.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel
+
+
+class TeamBase(BaseModel):
+    name: str
+
+
+class Team(TeamBase):
+    id: int

src/models/user.py

@@ -0,0 +1,54 @@
+from pydantic import BaseModel, Field
+from datetime import datetime
+
+
+class UserBase(BaseModel):
+    username: str = Field(description="Username of the user", examples=["Kalle"])
+
+
+class User(UserBase):
+    id: int = Field(description="ID of the user", examples=[1])
+    team_id: int = Field(description="ID of the team", examples=[1])
+    is_active: bool = Field(
+        default=True, description="Is the user active?", examples=[True]
+    )
+    avatar_id: int = Field(default=0, description="ID of the avatar", examples=[0])
+    is_admin: bool = Field(
+        default=False, description="Is the user an admin?", examples=[False]
+    )
+    last_login: datetime = Field(
+        description="Last login timestamp",
+        examples=["2025-01-01T00:00:00.000Z"],
+    )
+    created_at: datetime = Field(
+        description="Creation timestamp", examples=["2025-01-01T00:00:00.000Z"]
+    )
+    updated_at: datetime = Field(
+        description="Last update timestamp", examples=["2025-01-01T00:00:00.000Z"]
+    )
+    
+    @property
+    def has_logged_in(self) -> bool:
+        """Returns True if the user has ever logged in, False if last_login is the epoch date."""
+        epoch = datetime(1970, 1, 1, 0, 0, 0)
+        return self.last_login and self.last_login > epoch
+
+
+class CreateUser(UserBase):
+    password: str = Field(description="Password of the user", examples=["sn1rbul4"])
+    team_id: int = Field(description="ID of the team", examples=[1])
+    is_admin: bool = Field(
+        default=False, description="Is the user an admin?", examples=[False]
+    )
+
+
+class UserLogin(UserBase):
+    password: str = Field(description="Password of the user", examples=["sn1rbul4"])
+
+
+class UserChangePassword(BaseModel):
+    id: int = Field(description="ID of the user", examples=[1])
+    old_password: str = Field(
+        description="Old password of the user", examples=["sn1rbul4"]
+    )
+    password: str = Field(description="Password of the user", examples=["p1p4l1"])

src/routes/admin.py

@@ -1,10 +1,9 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, Body, HTTPException
-# from sqlmodel import Session, select
 
-# from models.user import User
+from models.user import CreateUser
-# from models.group import Group
+from models.team import TeamBase, Team
-
+from lib.db import get_connection
-# from ..app import get_session
+from lib.helpers import hash_password
 
 router = APIRouter(prefix="/admin", tags=["Admin"])
 
@@ -18,8 +17,41 @@ router = APIRouter(prefix="/admin", tags=["Admin"])
 #     # Implementation will be added with authentication logic
 #     return {"is_active": is_active}
 
-# @router.post("/users")
+
-# async def create_user(username: str, password: str, group_id: int, session: Session = Depends(get_session)):
+@router.post("/teams")
+async def create_team(team: TeamBase):
+    with get_connection() as conn:
+        conn.execute("INSERT INTO Team (name) VALUES (?)", (team.name,))
+    return {"message": "Team created successfully"}
+
+
+@router.get("/teams", response_model=list[Team])
+async def list_teams():
+    with get_connection() as conn:
+        teams = conn.execute("SELECT * FROM Team").fetchdf()
+    return teams.to_dict(orient="records")
+
+
+@router.post("/users")
+async def create_user(user: CreateUser):
+    with get_connection() as conn:
+        conn.execute(
+            "INSERT INTO User (username, hashed_password, team_id, is_admin) VALUES (?, ?, ?, ?)",
+            (
+                user.username,
+                hash_password(user.password),
+                user.team_id,
+                user.is_admin,
+            ),
+        )
+    return {"message": "User created successfully"}
+
+
+@router.post("/users/")
+async def disable_user(user_id: int = Body(..., embed=True)):
+    raise HTTPException(status_code=405, detail="Method not yet implemented")
+
+
 #     """
 #     Add a new user
 #     Only accessible to admin users

src/routes/auth.py

@@ -1,7 +1,37 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException
+
+from lib.db import get_connection
+from models.user import User, UserLogin
+from lib.helpers import verify_password
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
-@router.post("/token")
+
-async def login():
+@router.post("/token", response_model=User)
-    return {"message": "Login successful"}
+async def login(user_login: UserLogin):
+    with get_connection() as conn:
+        user_df = conn.execute(
+            'SELECT * FROM "User" WHERE username = ?', (user_login.username,)
+        ).fetchdf()
+
+    if user_df.empty:
+        raise HTTPException(status_code=401, detail="Invalid username or password")
+
+    user_data = user_df.to_dict(orient="records")[0]
+    if not verify_password(user_login.password, user_data["hashed_password"]):
+        raise HTTPException(status_code=401, detail="Invalid username or password")
+
+    # Update the login timestamp in the database
+    with get_connection() as conn:
+        conn.execute(
+            'UPDATE "User" SET last_login = CURRENT_TIMESTAMP WHERE id = ?',
+            (user_data["id"],),
+        )
+        
+        # Fetch the updated user data with the new last_login timestamp
+        updated_user = conn.execute(
+            'SELECT * FROM "User" WHERE id = ?', 
+            (user_data["id"],)
+        ).fetchdf().to_dict(orient="records")[0]
+    
+    return User(**updated_user)

src/routes/users.py

@@ -1,27 +1,36 @@
 from fastapi import APIRouter
+from fastapi import HTTPException
+
+from models.user import User, UserChangePassword
+from lib.db import get_connection
 
 router = APIRouter(prefix="/users", tags=["users"])
 
 
-@router.get("/")
+@router.get("/", response_model=list[User])
 async def list_users():
-    return {"users": []}
+    with get_connection() as conn:
+        users = conn.execute('SELECT * FROM "User"').fetchdf()
+
+    # Convert DataFrame to dict
+    users_dict = users.to_dict(orient="records")
+    return users_dict
 
 
-@router.get("/{user_id}")
+@router.get("/{user_id}", response_model=User)
 async def get_user(user_id: int):
-    return {"user_id": user_id}
+    with get_connection() as conn:
+        user_df = conn.execute(
+            'SELECT * FROM "User" WHERE id = ?', (user_id,)
+        ).fetchdf()
+
+    if user_df.empty:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    user_data = user_df.to_dict(orient="records")[0]
+    return User(**user_data)
 
 
-# @router.post("/")
+@router.patch("/{user_id}")
-# async def create_user(user: User):
+async def change_password(user: UserChangePassword):
-#     return {"user": user}
+    raise HTTPException(status_code=401, detail="Not implemented")
-
-# @router.patch("/{user_id}")
-# async def update_user(user_id: int, user: User):
-#     return {"user_id": user_id, "user": user}
-
-
-@router.delete("/{user_id}")
-async def delete_user(user_id: int):
-    return {"user_id": user_id}