added session manager, but in local dict

app/api.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI, File, UploadFile, Depends, HTTPException, Security
+from fastapi import FastAPI, File, UploadFile, Depends, HTTPException, Security, Response, Cookie
 from fastapi.responses import FileResponse, PlainTextResponse, StreamingResponse
 from fastapi.security import APIKeyHeader
 from sqlalchemy import exists
@@ -6,6 +6,7 @@ import hashlib
 from ftplib import FTP
 from io import BytesIO
 from . import db
+from .session_manager import SessionManager
 from dotenv import load_dotenv
 import os
 import hmac
@@ -33,13 +34,40 @@ def compute_hash(data: bytes, algorithm="sha256") -> str:
     h.update(data)
     return h.hexdigest()
 
-app = FastAPI()
+def set_cookie(response: Response, name: str, value: str, max_age: int) -> None:
+    response.set_cookie(
+        key=name,
+        value=value,
+        httponly=True,
+        secure=True,
+        samesite="Strict",
+        max_age=max_age,
+    )
 
+TOKEN_TTL = 60*15
+session_manager = SessionManager(TOKEN_TTL)
+
+app = FastAPI()
 
 @app.get("/")
 async def root():
     return {"message": "hiii from sfs"}
 
+
+@app.post("/login")
+def login(response: Response):
+    user_id = "user123"
+    token = session_manager.create(user_id)
+    set_cookie(response, "token", token, TOKEN_TTL)
+    return {"message": "logged in"}
+
+def get_current_user(token: str = Cookie(None)) -> str:
+    return session_manager.validate(token)
+
+@app.get("/protected")
+def protected_route(user: str = Depends(get_current_user)):
+    return {"message": f"Hello {user}, you are authenticated!"}
+
 @app.post("/file")
 async def save_file(file: UploadFile = File(...), api_key: str = Depends(verify_api_key)):
     contents = await file.read()

app/session_manager.py

@@ -0,0 +1,35 @@
+from fastapi import HTTPException
+from datetime import datetime, timedelta
+import secrets
+from typing import Dict, Optional
+
+class SessionManager:
+    def __init__(self, ttl: int):
+        self.ttl = ttl
+        self._tokens: Dict[str, Dict[str, datetime]] = {}
+
+    def create(self, user_id: str) -> str:
+        token = secrets.token_urlsafe(32)
+        self._tokens[token] = {
+            "user": user_id,
+            "expires": datetime.utcnow() + timedelta(seconds=self.ttl),
+        }
+        return token
+
+    def validate(self, token: Optional[str]) -> str:
+        self.cleanup()
+        if not token or token not in self._tokens:
+            raise HTTPException(status_code=401, detail="Not authenticated")
+
+        token_data = self._tokens[token]
+        if token_data["expires"] < datetime.utcnow():
+            del self._tokens[token]
+            raise HTTPException(status_code=401, detail="Session expired")
+
+        return token_data["user"]
+
+    def cleanup(self) -> None:
+        now = datetime.utcnow()
+        expired = [t for t, data in self._tokens.items() if data["expires"] < now]
+        for t in expired:
+            del self._tokens[t]