Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

update for use with TTL collection

Still needs two new tests to cover the missing lines
  • Loading branch information...
commit af6430cda338c9c4ed169da526a3d271ba2ec66e 1 parent ed63cea
@reedobrien reedobrien authored
Showing with 32 additions and 7 deletions.
  1. +31 −6 lumin/session.py
  2. +1 −1  lumin/tests/test_session.py
View
37 lumin/session.py
@@ -1,6 +1,7 @@
from __future__ import unicode_literals
import binascii
+from datetime import datetime
import os
import time
@@ -18,6 +19,8 @@
from pyramid.session import signed_deserialize
from pyramid.session import signed_serialize
+datetime_now = datetime.now
+
## ripped and modified from
## pyramid.session.UnencryptedCookieSessionFactoryConfig
@@ -92,7 +95,17 @@ def __init__(self, request):
"lumin.session.collection",
"lumin.sessions")
self.db = self.request.db
-
+ ## These should be run manually in an entry point or by a
+ ## sysadm so they don't get called on a fresh db wiht 10000
+ ## sessions
+ # self.db[self.collection].ensure_index(
+ # "atime",
+ # expireAfterSeconds=self._timeout
+ # )
+ # self.db[self.collection].ensure_index(
+ # "ctime",
+ # expireAfterSeconds=self._cookie_max_age
+ # )
now = time.time()
created = accessed = now
new = True
@@ -111,11 +124,15 @@ def __init__(self, request):
accessed, created, oid = value
new = False
state = self.db[self.collection].find_one({'_id': oid})
- if now - accessed > self._timeout:
- if state: # pragma: no branch
- ## if someone deleted the collection
- ## there would be no state.
+ if state:
+ if now - accessed > self._timeout:
+ ## Belt and braces check for expired cookie,
+ ## but this should occur in a TTL collection in
+ ## which case there **should** be
+ ## no state
self.invalidate({'_id': state['_id']})
+ state = self._new_session()
+ else:
state = self._new_session()
self.created = created
@@ -124,7 +141,15 @@ def __init__(self, request):
dict.__init__(self, state)
def _new_session(self):
- return {'_id': ObjectId()}
+ if self._cookie_max_age:
+ ctime = atime = datetime_now()
+ else:
+ atime, ctime = datetime_now(), None
+ return {
+ '_id': ObjectId(),
+ "ctime": ctime,
+ "atime": atime
+ }
# ISession methods
def changed(self):
View
2  lumin/tests/test_session.py
@@ -16,7 +16,7 @@ def test_ctor_no_cookie(self):
request.db = DummyDB()
session = self._makeOne(request)
self.assertEqual(type(dict(session)['_id']), ObjectId)
- self.assertEqual(len(dict(session)), 1)
+ self.assertEqual(len(dict(session)), 3)
def test_instance_conforms(self):
from zope.interface.verify import verifyObject
Please sign in to comment.
Something went wrong with that request. Please try again.