Session Management for User

src/api/dependencies/authenticated.py

@@ -14,9 +14,9 @@ async def get_pub_user_dep(
         session_token (str, optional): Session Token. Defaults to Cookie(default=None, alias=SessionConfig.auto_cookie_name).
 
     Raises:
-        HTTPException: _description_
-        HTTPException: _description_
-        HTTPException: _description_
+        HTTPException: No Session Token
+        HTTPException: No Session Found
+        HTTPException: No User for Session Found
 
     Returns:
         User: User Dictionary
@@ -44,9 +44,9 @@ async def get_user_dep(
         session_token (str, optional): Session Token. Defaults to Cookie(default=None, alias=SessionConfig.auto_cookie_name).
 
     Raises:
-        HTTPException: _description_
-        HTTPException: _description_
-        HTTPException: _description_
+        HTTPException: No Session Token
+        HTTPException: No Session Found
+        HTTPException: No User for Session Found
 
     Returns:
         User: User Dictionary
@@ -74,9 +74,9 @@ async def get_dangerous_user_dep(
         session_token (str, optional): Session Token. Defaults to Cookie(default=None, alias=SessionConfig.auto_cookie_name).
 
     Raises:
-        HTTPException: _description_
-        HTTPException: _description_
-        HTTPException: _description_
+        HTTPException: No Session Token
+        HTTPException: No Session Found
+        HTTPException: No User for Session Found
 
     Returns:
         User: User Dictionary

src/api/main.py

@@ -7,6 +7,7 @@
 from api.profile import router as profileRouter
 from api.twofactor import router as twofactorRouter
 from api.oauth_providers import router as oauthRouter
+from api.sessions import router as sessionsRouter
 import logging
 from tools import SecurityConfig
 
@@ -44,7 +45,8 @@ async def up():
 app.include_router(router)
 app.include_router(signupRouter)
 app.include_router(loginRouter)
-app.include_router(internalRouter)
 app.include_router(profileRouter)
+app.include_router(sessionsRouter)
 app.include_router(twofactorRouter)
 app.include_router(oauthRouter)
+app.include_router(internalRouter)

src/api/model.py

@@ -1,9 +1,19 @@
-from pydantic import BaseModel, field_validator, EmailStr, SecretStr, ConfigDict
-from typing import Optional
+from pydantic import BaseModel, field_validator, EmailStr, SecretStr, ConfigDict, Field
+from typing import Optional, List
 import re
 import bcrypt
 
 
+class SessionDetailResponse(BaseModel):
+    id: str = Field(alias="_id")
+    device_information: dict
+    createdAt: str
+
+
+class SessionListResponseModel(BaseModel):
+    sessions: List[SessionDetailResponse]
+
+
 class ConfirmEmailRequest(BaseModel):
     code: int | str
 

src/api/sessions.py

@@ -0,0 +1,41 @@
+from fastapi import APIRouter, Depends, Response, HTTPException
+from api.dependencies.authenticated import get_user_dep
+from tools import bson_to_json
+from crud.sessions import get_user_sessions, delete_session, get_session_by_id
+from api.model import SessionListResponseModel
+
+router = APIRouter(
+    prefix="/sessions",
+    tags=["Sessions"],
+    dependencies=[Depends(get_user_dep)],
+)
+
+
+@router.get("", response_model=SessionListResponseModel)
+async def sessions_list(user: dict = Depends(get_user_dep)):
+    """
+    # Get Sessions
+
+    ## Description
+    This endpoint is used to get the sessions of the user.
+    """
+    sesss = get_user_sessions(user["_id"])
+    sesss = [bson_to_json(sess) for sess in sesss]
+    return {"sessions": sesss}
+
+
+@router.delete("/{session_id}")
+async def delete_other_session(session_id: str, user: dict = Depends(get_user_dep)):
+    """
+    # Delete Session
+
+    ## Description
+    This endpoint is used to delete a session.
+    """
+    sess_to_delete = get_session_by_id(session_id)
+    if not sess_to_delete:
+        raise HTTPException(status_code=404, detail="Session not found.")
+    if sess_to_delete["user_id"] != user["_id"]:
+        raise HTTPException(status_code=404, detail="Session not found.")
+    delete_session(sess_to_delete["session_token"])
+    return Response(status_code=204)

src/crud/sessions.py

@@ -80,3 +80,29 @@ def delete_session(session_token: str) -> bool:
         sessions_collection.find_one_and_delete({"session_token": session_token})
         is not None
     )
+
+
+def get_user_sessions(user_id: str) -> list:
+    """
+    Get all sessions for a user.
+
+    Args:
+        user_id (str): User ID
+
+    Returns:
+        list: List of Sessions
+    """
+    return list(sessions_collection.find({"user_id": user_id}))
+
+
+def get_session_by_id(session_id: str) -> dict:
+    """
+    Get a session by ID.
+
+    Args:
+        session_id (str): Session ID
+
+    Returns:
+        dict: Session Information
+    """
+    return sessions_collection.find_one({"_id": ObjectId(session_id)})

src/tools/db.py

@@ -50,7 +50,7 @@
 
 
 def bson_to_json(data: bson.BSON) -> dict:
-    """Convert BSON to JSON. Also converts the _id to a string.
+    """Convert BSON to JSON. From MongoDB to JSON.
 
     Args:
         data (bson.BSON): BSON Data
@@ -61,5 +61,11 @@ def bson_to_json(data: bson.BSON) -> dict:
     if not data:
         return None
     original_json = json.loads(bson.json_util.dumps(data))
-    original_json["_id"] = str(original_json["_id"]["$oid"])
+    # Iterate over the keys and convert the _id to a string
+    for key in original_json:
+        if isinstance(original_json[key], dict):
+            if list(original_json[key].keys())[0] == "$oid":
+                original_json[key] = str(original_json[key]["$oid"])
+            elif list(original_json[key].keys())[0] == "$date":
+                original_json[key] = str(original_json[key]["$date"])
     return original_json