In this tutorial, we’ll see how to implement session management in a FastAPI application using Upstash Redis. We’ll use cookies to store session IDs, while session data is maintained in Redis for its speed and expiration features.
Session: A session is a mechanism to store user-specific data (like authentication status) between requests. It allows the server to “remember” users as they interact with the application.
Cookie: A small piece of data stored in the client’s browser. In this tutorial, we’ll use cookies to store session IDs, which the server uses to fetch session details from Redis.
Let’s implement a simple FastAPI application that handles login, profile access, and logout using Redis for session management. We use sliding expiration by updating the session expiration time on every request. If a session is inactive for 15 minutes (900 seconds), it will automatically expire.
main.py
Copy
Ask AI
from fastapi import FastAPI, Response, Cookie, HTTPExceptionfrom pydantic import BaseModelfrom upstash_redis import Redisfrom dotenv import load_dotenvimport uuid# Load environment variablesload_dotenv()redis = Redis.from_env()app = FastAPI()SESSION_TIMEOUT_SECONDS = 900 # 15 minutes# Define the request body model for loginclass LoginRequest(BaseModel): username: str@app.post("/login/")async def login(request: LoginRequest, response: Response): session_id = str(uuid.uuid4()) redis.hset(f"session:{session_id}", values={"user": request.username, "status": "active"}) redis.expire(f"session:{session_id}", SESSION_TIMEOUT_SECONDS) response.set_cookie(key="session_id", value=session_id, httponly=True) return {"message": "Logged in successfully", "session_id": session_id}@app.get("/profile/")async def get_profile(session_id: str = Cookie(None)): if not session_id: raise HTTPException(status_code=403, detail="No session cookie found") session_data = redis.hgetall(f"session:{session_id}") if not session_data: response = Response() response.delete_cookie(key="session_id") # Clear the expired cookie raise HTTPException(status_code=404, detail="Session expired") # Update the session expiration time (sliding expiration) redis.expire(f"session:{session_id}", SESSION_TIMEOUT_SECONDS) return {"session_id": session_id, "session_data": session_data}@app.post("/logout/")async def logout(response: Response, session_id: str = Cookie(None)): if session_id: redis.delete(f"session:{session_id}") response.delete_cookie(key="session_id") return {"message": "Logged out successfully"}
Let’s test the implementation using the following script:
test_script.py
Copy
Ask AI
import requestsbase_url = "http://127.0.0.1:8000"# Test loginresponse = requests.post(f"{base_url}/login/", json={"username": "abdullah"})print("Login Response:", response.json())# In the browser, you don't need to set cookies manually. The browser will handle it automatically.session_cookie = response.cookies.get("session_id")# Test profileprofile_response = requests.get(f"{base_url}/profile/", cookies={"session_id": session_cookie})print("Access Profile Response:", profile_response.json())# Test logoutlogout_response = requests.post(f"{base_url}/logout/", cookies={"session_id": session_cookie})print("Logout Response:", logout_response.json())# Test profile after logoutprofile_after_logout_response = requests.get(f"{base_url}/profile/", cookies={"session_id": session_cookie})print("Access Profile After Logout Response:", profile_after_logout_response.text)
By combining FastAPI, cookies, and Upstash Redis, we’ve created a reliable session management system. With Redis’s speed and built-in expiration features, this approach ensures secure and efficient handling of user sessions.To learn more about Upstash Redis, visit the Upstash Redis Documentation.