Pythonで自分用の簡単なWebアプリを作ろうと思ったが、ぱっと探した感じでは標準っぽいシンプルなセッション管理モジュールが見つからなかった。そんなわけで、ちょっと自分で書いてみた。
セッション管理の最低限の仕組みがあればよいので、次のような要件のものを作ろうと思う。
- セッションの新規作成、更新、削除ができる（当然）
- セッション情報はテキストファイルで管理する
- セッションのタイムアウトが設定できる
- 簡単なパラメータを保持ができる
- セッションIDはアクセスごとに変更する
この要件から、次のように動作するものとする。
インスタンス生成時
- セッションIDない状態でインスタンス生成 --> 新規セッションの作成（セッションIDの発行+セッションID名のテキストファイルをセッションファイルとしてセッション保存用フォルダに作成)
- 有効期限内のセッションIDを引数にインスタンス生成 --> 既存のセッションを更新(新規セッションIDを発行+既存セッションファイルを新規のセッションIDに変更)
- 有効期限切れのセッションIDを引数にインスタンス生成 --> 既存セッションを削除し、新規セッションの作成
セッションパラメータの読み書き
セッションパラメータは「key=value」の形式でテキスト内に保存し、読み書きのためのメンバー関数を用意する。
- 読み込みはkeyを指定せず、辞書型で返す
- 書き込みはkeyと valueを指定する
- 削除はkeyを指定する
セッション削除
ログオフ時を想定し、任意のタイミングでセッションを削除できるようにする。セッション削除後はセッションパラメータへのアクセスを行おうとすると、「False」を受け取る。
セッションIDの生成
セッションIDとして、アルファベット(大文字、小文字)+数字からランダムな文字列を生成する。
コード
#/usr/local/bin/python
import os
import time
import random
import string
from stat import ST_ATIME, ST_MTIME
DEFAULT_EXPIRE_SECONDS = 36000 # Default Session Expire Seconds : 10hr
DEFAULT_SESSION_PATH = './' # Default Session Storage Path : Current Dir
NULL_SESSION_ID = '' # NULL Session ID
aZ09 = string.digits + string.letters # Charactor set for Session ID
class SimpleSession:
def __init__(self, sessionPath = DEFAULT_SESSION_PATH,
sessionID = NULL_SESSION_ID,
expire = DEFAULT_EXPIRE_SECONDS):
self.sessPath = sessionPath
self.sessID = sessionID
self.expire = expire
# If the session id is null, a new session id will be published.
# If the session id exists, the session id will be renamed.
if not self.sessID:
self.sessID = self.publishSessID()
else:
self.sessID = self.checkSession()
# If the session is expired, the session is closed and a new session
# is created. If the session is valid, the session id is renamed.
def checkSession( self ):
if os.path.exists(self.sessPath + self.sessID):
sessFile = self.sessPath + self.sessID
sessAtime = os.stat(sessFile)[ST_ATIME]
now = time.time()
if now - sessAtime > self.expire:
self.close()
return self.publishSessID()
else:
return self.publishSessID()
# A new session ID is created. If the session exists, it copy to a new
# one.
def publishSessID( self ):
newSessID = SimpleSession.mkSessID()
while os.path.exists(self.sessPath + newSessID):
newSessID = self.mkSessID()
if self.sessID:
if os.path.exists(self.sessPath + self.sessID):
try:
os.rename(self.sessPath + self.sessID,
self.sessPath + newSessID)
except OSError:
self.close()
try:
fd = open(self.sessPath + newSessID, 'w')
except:
return NULL_SESSION_ID
fd.write('')
fd.close()
return newSessID
# The session ID is closed.
def close( self ):
if self.sessID:
if os.path.exists(self.sessPath + self.sessID):
try:
os.remove(self.sessPath + self.sessID)
except OSError:
self.sessID = NULL_SESSION_ID
return False
self.sessID = NULL_SESSION_ID
return True
# A Key-Value Parameter is set in the session data.
def setParam( self, key, value = ''):
if self.sessID:
if os.path.exists(self.sessPath + self.sessID):
try:
fd = open(self.sessPath + self.sessID, 'r')
except:
return False
sessParam = {}
for line in fd:
lstStr = []
line = line.rstrip('\n')
lstStr = line.split('=')
sessParam[lstStr[0]] = lstStr[1]
try:
fd = open(self.sessPath + self.sessID, 'w')
except:
return False
sessParam[key] = value
for k, v in sessParam.items():
fd.write(k + '=' + v + '\n')
fd.close()
return True
return False
# The session parameters is got as a dictionary value.
def getParam( self ):
sessParam = {}
if self.sessID:
if os.path.exists( self.sessPath + self.sessID):
try:
fd = open(self.sessPath + self.sessID, 'r')
except:
pass
for line in fd:
lstStr = []
line = line.rstrip('\n')
lstStr = line.split('=')
if not lstStr[1]:
lstStr[1] = ''
sessParam[lstStr[0]] = lstStr[1]
return sessParam
# The session parameter of key will be deleted.
def deleteParam( self, key ):
if self.sessID:
if os.path.exists(self.sessPath + self.sessID):
try:
fd = open(self.sessPath + self.sessID, 'r')
except:
return False
sessParam = {}
for line in fd:
lstStr = []
line = line.rstrip('\n')
lstStr = line.split('=')
sessParam[lstStr[0]] = lstStr[1]
fd.close()
try:
fd = open(self.sessPath + self.sessID, 'w')
except:
return False
for k, v in sessParam.items():
if k != key:
fd.write(k + '=' + v + '\n')
fd.close()
return True
return False
# A Random string for a session id is returned.
# This is a static function.
@staticmethod
def mkSessID( sets = aZ09, length = 36):
random.seed()
newSessID = ''.join(random.choice(sets) for i in xrange(length))
newSessID = 'Sess_' + newSessID
return newSessID
if __name__ == '__main__':
print "mkSessID : " + SimpleSession.mkSessID()
sess1 = SimpleSession("../sess/", "")
print "SessionPath = " + sess1.sessPath
print "SessionID = " + sess1.sessID
print "Expire = " + str(sess1.expire)
print "Session Param : "
print sess1.getParam()
print "SetParam('KEY', 'VALUE')"
sess1.setParam('KEY', 'VALUE')
print "Session Param : "
print sess1.getParam()
print "Session Dir : "
print os.listdir(sess1.sessPath)
print "Refresh Session Using " + sess1.sessID
sess1 = SimpleSession("../sess/", sess1.sessID, 1000)
print "SessionPath = " + sess1.sessPath
print "SessionID = " + sess1.sessID
print "Expire = " + str(sess1.expire)
print "Session Param : "
print sess1.getParam()
print "SetParam('KEY', 'VALUE2')"
sess1.setParam('KEY', 'VALUE2')
print "SetParam('KEY2', 'VALUE2')"
sess1.setParam('KEY2', 'VALUE2')
print "SetParam('KEY3')"
sess1.setParam('KEY3')
print "Session Param : "
print sess1.getParam()
print "DeleteParam('KEY')"
sess1.deleteParam('KEY')
print "Session Param : "
print sess1.getParam()
print "Session Dir : "
print os.listdir(sess1.sessPath)
print "Session Close"
sess1.close()
print "Session Dir : "
print os.listdir(sess1.sessPath)
