#!/usr/bin/env python3 """ Redis session store demo server. Run this file and visit http://localhost:8080 to create, inspect, and destroy Redis-backed sessions in a local browser session. """ from __future__ import annotations import argparse from html import escape from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json from pathlib import Path import sys from urllib.parse import parse_qs, urlparse sys.path.insert(0, str(Path(__file__).resolve().parent)) try: import redis from session_store import RedisSessionStore except ImportError as exc: print(f"Error: {exc}") print("Make sure the 'redis' package is installed: pip install redis") sys.exit(1) class SessionDemoHandler(BaseHTTPRequestHandler): """Serve a small interactive page that uses Redis for session storage.""" session_store: RedisSessionStore | None = None session_cookie_name = "sid" def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path in {"/", "/index.html"}: self._render_index() return if parsed.path == "/session": self._handle_session_fetch() return self.send_error(404) def do_POST(self) -> None: parsed = urlparse(self.path) if parsed.path == "/login": self._handle_login() return if parsed.path == "/increment": self._handle_increment() return if parsed.path == "/ttl": self._handle_ttl_update() return if parsed.path == "/logout": self._handle_logout() return self.send_error(404) def _render_index(self) -> None: session_id = self._get_session_id_from_cookie() session = self._load_session(session_id) if session_id else None self._send_html(self._html_page(session_id, session)) def _handle_session_fetch(self) -> None: session_id = self._get_session_id_from_cookie() if not session_id: self._send_json({"authenticated": False, "session": None}, 200) return session = self._load_session(session_id) if session is None: self._send_json( {"authenticated": False, "session": None}, 200, clear_session_cookie=True, ) return self._send_json( { "authenticated": True, "session_id": session_id, "session": session, "configured_ttl": self.session_store.get_configured_ttl(session_id), "ttl": self.session_store.get_ttl(session_id), }, 200, ) def _handle_login(self) -> None: params = self._read_form_data() username = params.get("username", ["Guest"])[0].strip() or "Guest" ttl = self._parse_ttl_value(params.get("ttl", ["60"])[0]) if ttl is None: self._send_json({"error": "TTL must be a whole number greater than 0."}, 400) return session_id = self.session_store.create_session( { "username": username, "page_views": "1", }, ttl=ttl, ) session = self.session_store.get_session(session_id) self.send_response(200) self.send_header("Content-Type", "application/json") self._set_session_cookie(session_id) self.end_headers() self.wfile.write( json.dumps( { "authenticated": True, "session_id": session_id, "session": session, "configured_ttl": self.session_store.get_configured_ttl(session_id), "ttl": self.session_store.get_ttl(session_id), } ).encode("utf-8") ) def _handle_increment(self) -> None: session_id = self._get_session_id_from_cookie() if not session_id: self._send_json({"error": "No active session"}, 401) return new_value = self.session_store.increment_field(session_id, "page_views") if new_value is None: self._send_json( {"error": "Session expired"}, 401, clear_session_cookie=True, ) return session = self.session_store.get_session(session_id) if session is None: self._send_json( {"error": "Session expired"}, 401, clear_session_cookie=True, ) return self._send_json( { "authenticated": True, "session_id": session_id, "session": session, "configured_ttl": self.session_store.get_configured_ttl(session_id), "ttl": self.session_store.get_ttl(session_id), "page_views": new_value, }, 200, ) def _handle_ttl_update(self) -> None: session_id = self._get_session_id_from_cookie() if not session_id: self._send_json({"error": "No active session"}, 401) return params = self._read_form_data() ttl = self._parse_ttl_value(params.get("ttl", [""])[0]) if ttl is None: self._send_json({"error": "TTL must be a whole number greater than 0."}, 400) return if not self.session_store.set_session_ttl(session_id, ttl): self._send_json( {"error": "Session expired"}, 401, clear_session_cookie=True, ) return session = self.session_store.get_session(session_id) if session is None: self._send_json( {"error": "Session expired"}, 401, clear_session_cookie=True, ) return self._send_json( { "authenticated": True, "session_id": session_id, "session": session, "configured_ttl": self.session_store.get_configured_ttl(session_id), "ttl": self.session_store.get_ttl(session_id), }, 200, ) def _handle_logout(self) -> None: session_id = self._get_session_id_from_cookie() if session_id: self.session_store.delete_session(session_id) self.send_response(200) self.send_header("Content-Type", "application/json") self._clear_session_cookie() self.end_headers() self.wfile.write(json.dumps({"authenticated": False}).encode("utf-8")) def _read_form_data(self) -> dict[str, list[str]]: content_length = int(self.headers.get("Content-Length", "0")) raw_body = self.rfile.read(content_length).decode("utf-8") return parse_qs(raw_body) def _get_session_id_from_cookie(self) -> str | None: raw_cookie = self.headers.get("Cookie") if not raw_cookie: return None cookie = SimpleCookie() cookie.load(raw_cookie) morsel = cookie.get(self.session_cookie_name) if morsel is None: return None return morsel.value def _parse_ttl_value(self, raw_ttl: str) -> int | None: try: ttl = int(raw_ttl) except (TypeError, ValueError): return None if ttl < 1: return None return ttl def _set_session_cookie(self, session_id: str) -> None: cookie = SimpleCookie() cookie[self.session_cookie_name] = session_id cookie[self.session_cookie_name]["path"] = "/" cookie[self.session_cookie_name]["httponly"] = True cookie[self.session_cookie_name]["samesite"] = "Lax" self.send_header("Set-Cookie", cookie.output(header="").strip()) def _clear_session_cookie(self) -> None: cookie = SimpleCookie() cookie[self.session_cookie_name] = "" cookie[self.session_cookie_name]["path"] = "/" cookie[self.session_cookie_name]["expires"] = "Thu, 01 Jan 1970 00:00:00 GMT" cookie[self.session_cookie_name]["max-age"] = 0 self.send_header("Set-Cookie", cookie.output(header="").strip()) def _load_session(self, session_id: str | None) -> dict[str, str] | None: if not session_id: return None session = self.session_store.get_session(session_id) if session is None: return None session["ttl"] = str(self.session_store.get_ttl(session_id)) configured_ttl = self.session_store.get_configured_ttl(session_id) if configured_ttl is not None: session["session_ttl"] = str(configured_ttl) return session def _send_html(self, html: str, status: int = 200) -> None: self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(html.encode("utf-8")) def _send_json( self, payload: dict, status: int, clear_session_cookie: bool = False, ) -> None: self.send_response(status) self.send_header("Content-Type", "application/json") if clear_session_cookie: self._clear_session_cookie() self.end_headers() self.wfile.write(json.dumps(payload).encode("utf-8")) def _html_page( self, session_id: str | None, session: dict[str, str] | None, ) -> str: session_html = self._session_html(session_id, session) return f""" Redis Session Store Demo
redis-py + Python standard library HTTP server

Redis Session Store Demo

Start a session, refresh it by interacting with the page, and watch Redis hold the server-side session data while the browser keeps only an opaque cookie.

Start a session

Try a short TTL like 10 or 15 seconds to watch the session expire, then interact with the page to see the expiration refresh.

Current session

{session_html}
""" def _session_html( self, session_id: str | None, session: dict[str, str] | None, ) -> str: if not session_id or session is None: return ( "

No active session.

" "

Create one to store state in Redis and receive a cookie-backed session ID.

" ) safe_session_id = escape(session_id) safe_username = escape(session.get("username", "")) safe_page_views = escape(session.get("page_views", "0")) safe_created = escape(session.get("created_at", "")) safe_accessed = escape(session.get("last_accessed_at", "")) safe_configured_ttl = escape(session.get("session_ttl", "")) safe_ttl = escape(session.get("ttl", "")) return f"""
Session ID
{safe_session_id}
Username
{safe_username}
Page views
{safe_page_views}
Configured TTL
{safe_configured_ttl} seconds
Created
{safe_created}
Last accessed
{safe_accessed}
TTL
{safe_ttl} seconds
""" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the Redis session store demo server.") parser.add_argument("--host", default="127.0.0.1", help="HTTP bind host") parser.add_argument("--port", type=int, default=8080, help="HTTP bind port") parser.add_argument("--redis-host", default="localhost", help="Redis host") parser.add_argument("--redis-port", type=int, default=6379, help="Redis port") parser.add_argument("--ttl", type=int, default=1800, help="Session TTL in seconds") return parser.parse_args() def main() -> None: args = parse_args() redis_client = redis.Redis( host=args.redis_host, port=args.redis_port, decode_responses=True, ) SessionDemoHandler.session_store = RedisSessionStore( redis_client=redis_client, ttl=args.ttl, ) server = ThreadingHTTPServer((args.host, args.port), SessionDemoHandler) print(f"Redis session store demo server listening on http://{args.host}:{args.port}") print(f"Using Redis at {args.redis_host}:{args.redis_port} with TTL {args.ttl}s") server.serve_forever() if __name__ == "__main__": main()