patx/micropie
Improve the MongoDB session backend example.
Commit 6f5f840 · patx · 2025-12-04T10:30:22-05:00
Comments
No comments yet.
Diff
diff --git a/examples/sessions/motor_backend.py b/examples/sessions/motor_backend.py
index ff93d35..ffa050a 100644
--- a/examples/sessions/motor_backend.py
+++ b/examples/sessions/motor_backend.py
@@ -5,36 +5,82 @@ with MicroPie.
This application increments a visit counter stored in a MongoDB collection for sessions.
"""
-from micropie import App, SessionBackend
+from micropie import App, SessionBackend, SESSION_TIMEOUT
import motor.motor_asyncio
import uuid
from datetime import datetime, timedelta
+from typing import Dict, Any
+
class MotorSessionBackend(SessionBackend):
- def __init__(self, mongo_uri: str, db_name: str, collection_name: str = "sessions"):
+ def __init__(
+ self,
+ mongo_uri: str,
+ db_name: str,
+ collection_name: str = "sessions",
+ default_timeout: int = SESSION_TIMEOUT,
+ ) -> None:
+ """
+ A simple MongoDB-backed session store for MicroPie using Motor.
+
+ NOTE:
+ - Suitable for real deployments as a starting point.
+ - You should still tune indexes, replica set, etc. for your environment.
+ """
self.client = motor.motor_asyncio.AsyncIOMotorClient(mongo_uri)
self.db = self.client[db_name]
self.collection = self.db[collection_name]
+ self.default_timeout = default_timeout
+
+ async def load(self, session_id: str) -> Dict[str, Any]:
+ """
+ Load session data from MongoDB.
+
+ - If no session_id is provided, returns an empty dict.
+ - If the session is expired, deletes it and returns empty.
+ """
+ if not session_id:
+ return {}
- async def load(self, session_id: str) -> dict:
- """Load session data from MongoDB; if expired delete and return empty."""
doc = await self.collection.find_one({"_id": session_id})
if not doc:
return {}
- if "expires_at" in doc and datetime.utcnow() > doc["expires_at"]:
+
+ expires_at = doc.get("expires_at")
+ if expires_at is not None and datetime.utcnow() > expires_at:
+ # Expired: clean up and treat as no session
await self.collection.delete_one({"_id": session_id})
return {}
- return doc.get("data", {})
- async def save(self, session_id: str, data: dict, timeout: int) -> None:
- """Save session data into MongoDB with an expiration time."""
- expires_at = datetime.utcnow() + timedelta(seconds=timeout)
+ return doc.get("data", {}) or {}
+
+ async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
+ """
+ Save session data into MongoDB with an expiration time.
+
+ MicroPie may call this with:
+ - `data` empty and/or `timeout <= 0` to mean "delete this session".
+ """
+ if not session_id:
+ # If we somehow got here with an empty session_id, just ignore.
+ return
+
+ # Treat empty data or non-positive timeout as a delete / logout
+ if not data or timeout <= 0:
+ await self.collection.delete_one({"_id": session_id})
+ return
+
+ # Use the timeout provided by MicroPie if set, otherwise our default.
+ ttl = timeout or self.default_timeout
+ expires_at = datetime.utcnow() + timedelta(seconds=ttl)
+
await self.collection.update_one(
{"_id": session_id},
{"$set": {"data": data, "expires_at": expires_at}},
- upsert=True
+ upsert=True,
)
+
class MyApp(App):
async def index(self):
# Access the session via self.request.session.
@@ -42,10 +88,12 @@ class MyApp(App):
self.request.session["visits"] = 1
else:
self.request.session["visits"] += 1
+
return f"You have visited {self.request.session['visits']} times."
+
# MongoDB configuration; adjust the URI and database name as needed.
-MONGO_URI = "YOUR URI HERE"
+MONGO_URI = "your uri here"
DB_NAME = "example"
# Create an instance of the Motor session backend.
@@ -53,3 +101,4 @@ backend = MotorSessionBackend(MONGO_URI, DB_NAME)
# Pass the Motor session backend to our application.
app = MyApp(session_backend=backend)
+