def acquire_lock(self, path=None):
"""Acquire an exclusive lock on the currently-loaded session data."""
if path is None:
path = self._get_file_path()
path += self.LOCK_SUFFIX
checker = locking.LockChecker(self.id, self.lock_timeout)
while not checker.expired():
try:
self.lock = lockfile.LockFile(path)
except lockfile.LockError:
time.sleep(0.1)
else:
break
self.locked = True
if self.debug:
cherrypy.log('Lock acquired.', 'TOOLS.SESSIONS')
python类session()的实例源码
def save():
"""Save any changed session data."""
if not hasattr(cherrypy.serving, 'session'):
return
request = cherrypy.serving.request
response = cherrypy.serving.response
# Guard against running twice
if hasattr(request, '_sessionsaved'):
return
request._sessionsaved = True
if response.stream:
# If the body is being streamed, we have to save the data
# *after* the response has been written out
request.hooks.attach('on_end_request', cherrypy.session.save)
else:
# If the body is not being streamed, we save the data now
# (so we can release the lock).
if is_iterator(response.body):
response.collapse_body()
cherrypy.session.save()
def do_check(self):
"""Assert username. Raise redirect, or return True if request handled.
"""
sess = cherrypy.session
request = cherrypy.serving.request
response = cherrypy.serving.response
username = sess.get(self.session_key)
if not username:
sess[self.session_key] = username = self.anonymous()
self._debug_message('No session[username], trying anonymous')
if not username:
url = cherrypy.url(qs=request.query_string)
self._debug_message(
'No username, routing to login_screen with from_page %(url)r',
locals(),
)
response.body = self.login_screen(url)
if 'Content-Length' in response.headers:
# Delete Content-Length header so finalize() recalcs it.
del response.headers['Content-Length']
return True
self._debug_message('Setting request.login to %(username)r', locals())
request.login = username
self.on_check(username)
def p(self, id):
html = ""
if id == "" or id == None :
html += "Error no radio id"
return html
if id == "0" :
html += "0 is reserved, sorry"
return html
(radio, genre, url) = playradio(id)
if url == '':
html += "Error in parameter %s" % url
return html
cherrypy.session['playing'] = id
html += '''<h3>Now Playing: '''
html += '''<a href="%s">%s</a>''' % (url, radio)
html += '''<a href="#" onClick="fplayradio('%s')">''' % id
html += '''<span class="glyphicon glyphicon-play"></span></a>'''
html += ''' <a href="#" onClick="fmodradio('%s')"><span class="glyphicon glyphicon-pencil"></span></a></small> ''' % id
html += '''<a href="#" onClick="fdelradio('%s')"><span class="glyphicon glyphicon-trash"></span></a> ''' % id
html += '''<a href="#" onClick="faddfav('%s')"><span class="glyphicon glyphicon-star"></span></a>''' % id
html += '''</h3>'''
print html
return html
def getfooter() :
global footer, version
db = cherrypy.session['database']
try:
con = lite.connect( db )
cur = con.cursor()
sql = "select radio, genre, url from Radio where id=0"
cur.execute(sql)
(radio, genre, url) = cur.fetchone()
except:
(radio, genre, url) = ('ERROR', sql, '')
con.close()
hostname = socket.gethostname()
f = '''<footer class="footer"> <div class="container">'''
f += '''<p class="text-muted">'''
f += '''Session id: %s - Session Database %s<br>''' % (cherrypy.session.id, cherrypy.session['database'])
f += '''Host: %s - Version: %s - Updated: %s // Last: %s''' % (hostname, version, genre, url)
f += '''</p>'''
f += '''</div></footer>'''
return f + footer
def nonexist(id) :
db = cherrypy.session['database']
sql = "UPDATE Radio set exist = 0 WHERE id = '%s'" % (id)
try:
con = lite.connect( db )
cur = con.cursor()
cur.execute(sql)
ret = True
except:
print "Error in sql %s" % sql
ret = False
updateversiondb(cur)
con.commit()
con.close()
return ret
def insert(radio, genre, url) :
db = cherrypy.session['database']
sql = "INSERT INTO Radio (radio, genre, url, exist) VALUES('%s', '%s', '%s', 1)" % (radio, genre, url)
try:
con = lite.connect( db )
cur = con.cursor()
cur.execute(sql)
ret = True
except:
print "Error inserting sql %s" % sql
ret = False
updateversiondb(cur)
con.commit()
con.close()
return ret
def modify(id, radio, url, genre) :
db = cherrypy.session['database']
sql = "UPDATE Radio SET radio='%s', url='%s', genre='%s', exist=1 WHERE id = %s" % (radio, url, genre, id)
try:
con = lite.connect( db )
cur = con.cursor()
cur.execute(sql)
ret = True
except:
print "Error modify sql %s" % sql
ret = False
updateversiondb(cur)
con.commit()
con.close()
return ret
def addgen(id, genre) :
db = cherrypy.session['database']
sql = "UPDATE Radio SET genre='%s' WHERE id = %s" % (genre, id)
try:
con = lite.connect( db )
cur = con.cursor()
cur.execute(sql)
ret = True
except:
print "Error modify sql %s" % sql
ret = False
updateversiondb(cur)
con.commit()
con.close()
return ret
def getradio(id) :
db = cherrypy.session['database']
if id.isdigit() :
sql = "select radio, genre, url from Radio where id=%s" % id
else:
sql = "select radio, genre, url from Radio where url=%s" % id
try:
con = lite.connect( db )
cur = con.cursor()
cur.execute(sql)
except:
rows = [('Not Found', '', '')]
rows = cur.fetchone()
if rows == None:
rows = ('Not Found', '', '')
con.close()
return rows
def login(self, **params):
print(params)
message = ''
submit = params.get('submit', False)
user = params.get('user', '')
password = params.get('password', '')
returl = params.get('returl')
if not cmd_args.no_anonymous_access:
raise cherrypy.HTTPRedirect('/index')
if submit:
if user and password:
if user == cmd_args.admin_user and password == cmd_args.admin_password:
# default, in-memory sessions
cherrypy.session['logged_in'] = True
cherrypy.session['login_time'] = time.time()
raise cherrypy.HTTPRedirect(returl if returl else '/index')
else:
message = 'Wrong username and/or password!'
else:
message = 'Username and password needed!'
tmpl = env.get_template('login.html')
return tmpl.render(message=message, user=user, returl=returl)
def check_auth(*args, **kwargs):
"""A tool that looks in config for 'auth.require'. If found and it
is not None, a login is required and the entry is evaluated as a list of
conditions that the user must fulfill"""
conditions = cherrypy.request.config.get('auth.require', None)
if conditions is not None:
username = cherrypy.session.get(SESSION_KEY)
if username:
cherrypy.request.login = username
for condition in conditions:
# A condition is just a callable that returns true or false
if not condition():
raise cherrypy.InternalRedirect(LOGIN_URL)
else:
raise cherrypy.InternalRedirect(LOGIN_URL)
def save(self):
"""Save session data."""
try:
# If session data has never been loaded then it's never been
# accessed: no need to save it
if self.loaded:
t = datetime.timedelta(seconds=self.timeout * 60)
expiration_time = self.now() + t
if self.debug:
cherrypy.log('Saving session %r with expiry %s' %
(self.id, expiration_time),
'TOOLS.SESSIONS')
self._save(expiration_time)
else:
if self.debug:
cherrypy.log(
'Skipping save of session %r (no session loaded).' %
self.id, 'TOOLS.SESSIONS')
finally:
if self.locked:
# Always release the lock if the user didn't release it
self.release_lock()
if self.debug:
cherrypy.log('Lock released after save.', 'TOOLS.SESSIONS')
def _load(self, path=None):
assert self.locked, ("The session load without being locked. "
"Check your tools' priority levels.")
if path is None:
path = self._get_file_path()
try:
f = open(path, "rb")
try:
return pickle.load(f)
finally:
f.close()
except (IOError, EOFError):
e = sys.exc_info()[1]
if self.debug:
cherrypy.log("Error loading the session pickle: %s" %
e, 'TOOLS.SESSIONS')
return None
def save():
"""Save any changed session data."""
if not hasattr(cherrypy.serving, "session"):
return
request = cherrypy.serving.request
response = cherrypy.serving.response
# Guard against running twice
if hasattr(request, "_sessionsaved"):
return
request._sessionsaved = True
if response.stream:
# If the body is being streamed, we have to save the data
# *after* the response has been written out
request.hooks.attach('on_end_request', cherrypy.session.save)
else:
# If the body is not being streamed, we save the data now
# (so we can release the lock).
if is_iterator(response.body):
response.collapse_body()
cherrypy.session.save()
def do_login(self, username, password, from_page='..', **kwargs):
"""Login. May raise redirect, or return True if request handled."""
response = cherrypy.serving.response
error_msg = self.check_username_and_password(username, password)
if error_msg:
body = self.login_screen(from_page, username, error_msg)
response.body = body
if "Content-Length" in response.headers:
# Delete Content-Length header so finalize() recalcs it.
del response.headers["Content-Length"]
return True
else:
cherrypy.serving.request.login = username
cherrypy.session[self.session_key] = username
self.on_login(username)
raise cherrypy.HTTPRedirect(from_page or "/")
def do_check(self):
"""Assert username. Raise redirect, or return True if request handled.
"""
sess = cherrypy.session
request = cherrypy.serving.request
response = cherrypy.serving.response
username = sess.get(self.session_key)
if not username:
sess[self.session_key] = username = self.anonymous()
self._debug_message('No session[username], trying anonymous')
if not username:
url = cherrypy.url(qs=request.query_string)
self._debug_message(
'No username, routing to login_screen with from_page %(url)r',
locals(),
)
response.body = self.login_screen(url)
if "Content-Length" in response.headers:
# Delete Content-Length header so finalize() recalcs it.
del response.headers["Content-Length"]
return True
self._debug_message('Setting request.login to %(username)r', locals())
request.login = username
self.on_check(username)
def index(self, username=None):
if cherrypy.session.get('auth', False):
raise cherrypy.HTTPRedirect('/update')
else:
return {'username': username}
def update(self):
if not cherrypy.session.get('auth', False):
raise cherrypy.HTTPRedirect('/')
else:
return vars(cherrypy.session['user'])
def reset(self, token=None, username=None):
if cherrypy.engine.publish('token-verify', token, username).pop():
cherrypy.session['token'] = token
cherrypy.session['username'] = username
return {'ok': True}
else:
return {'ok': False}
def login(self, username=None, password=None):
if username is None or password is None:
raise cherrypy.HTTPError(400, 'Bad Request')
try:
cherrypy.session['user'] = User(username)
cherrypy.session['auth'] = cherrypy.session['user'].authenticate(password)
return {'ok': cherrypy.session['user'].auth}
except (InvalidUser, InvalidCredentials):
return {
'ok': False,
'error': 'Invalid credentials. Try again.'
}
except UserModelException:
return {'ok': False}
def change(self, **params):
engine = cherrypy.engine
if cherrypy.session.get('auth', False):
user = cherrypy.session['user']
oldpasswd = cherrypy.request.params.get('oldpassword')
newpasswd = cherrypy.request.params.get('newpassword')
try:
user.change_password(oldpasswd, newpasswd)
return {'ok': True}
except InvalidCredentials:
return {
'ok': False,
'error': 'Current password invalid.'
}
except UserModelException:
return {
'ok': False,
'error': 'Unknown system error. Contact your Systems Administrator.'
}
elif cherrypy.session.get('token', False):
cherrypy.session['user'] = User(cherrypy.session['username'])
newpassword = cherrypy.request.params.get('newpassword')
try:
cherrypy.session['user'].set_password(newpassword)
return {'ok': True}
except UserModelException:
return {
'ok': False,
'error': 'Unable to change your password. Try again later.'
}
def check_auth(*args, **kwargs):
"""A tool that looks in config for 'auth.require'. If found and it
is not None, a login is required and the entry is evaluated as a list of
conditions that the user must fulfill"""
conditions = cherrypy.request.config.get('auth.require', None)
if conditions is not None:
username = cherrypy.session.get(SESSION_KEY)
if username:
cherrypy.request.login = username
for condition in conditions:
# A condition is just a callable that returns true or false
if not condition():
raise cherrypy.HTTPRedirect("/auth/login")
else:
raise cherrypy.HTTPRedirect("/auth/login")
def login(self, username=None, password=None, from_page="/"):
if username is None or password is None:
return self.get_loginform("", from_page=from_page)
error_msg = check_credentials(username, password)
if error_msg:
return self.get_loginform(username, error_msg, from_page)
else:
cherrypy.session[SESSION_KEY] = cherrypy.request.login = username
self.on_login(username)
raise cherrypy.HTTPRedirect(from_page or "/")
def logout(self, from_page="/"):
sess = cherrypy.session
username = sess.get(SESSION_KEY, None)
sess[SESSION_KEY] = None
if username:
cherrypy.request.login = None
self.on_logout(username)
raise cherrypy.HTTPRedirect(from_page or "/")
#from auth import AuthController, require, member_of, name_is
def check_auth(*args, **kwargs):
"""A tool that looks in config for 'auth.require'. If found and it
is not None, a login is required and the entry is evaluated as a list of
conditions that the user must fulfill"""
conditions = cherrypy.request.config.get('auth.require', None)
if conditions is not None:
username = cherrypy.session.get(SESSION_KEY)
if username:
cherrypy.request.login = username
for condition in conditions:
# A condition is just a callable that returns true or false
if not condition():
raise cherrypy.HTTPRedirect("/auth/login")
else:
raise cherrypy.HTTPRedirect("/auth/login")
def login(self, username=None, password=None, from_page="/"):
if username is None or password is None:
return self.get_loginform("", from_page=from_page)
error_msg = check_credentials(username, password)
if error_msg:
return self.get_loginform(username, error_msg, from_page)
else:
cherrypy.session[SESSION_KEY] = cherrypy.request.login = username
self.on_login(username)
raise cherrypy.HTTPRedirect(from_page or "/")
def read_comic(self, comic_path, page_num=0):
logging.debug("Reader Requested")
cherrypy.session.load()
username = cherrypy.request.login
if 'sizepref' not in cherrypy.session:
cherrypy.session['sizepref'] = 'wide'
user_size_pref = cherrypy.session['sizepref']
scanner = ComicScanner()
scanner.user_unpack_comic(comic_path, username)
image_list = scanner.reading_images(username)
num_pages = len(image_list)
if num_pages == 0:
image_list = ['static/images/imgnotfound.png']
cookie_comic = re.sub(r'\W+', '', comic_path)
cookie_check = cherrypy.request.cookie
if cookie_comic not in cookie_check:
logging.debug("Cookie Creation")
cookie_set = cherrypy.response.cookie
cookie_set[cookie_comic] = 0
cookie_set[cookie_comic]['path'] = '/'
cookie_set[cookie_comic]['max-age'] = 2419200
next_page = 1
last_page = num_pages - 1
else:
logging.debug("Cookie Read")
page_num = int(cookie_check[cookie_comic].value)
logging.debug("Cookie Set To %d" % page_num)
next_page = page_num + 1
last_page = page_num - 1
logging.info("Reader Served")
return serve_template(templatename="read.html", pages=image_list, current_page=page_num, np=next_page, lp=last_page, nop=num_pages, size=user_size_pref, cc=cookie_comic)
def change_page(self, page_str, cookie_comic):
cherrypy.session.load()
if 'sizepref' not in cherrypy.session:
cherrypy.session['sizepref'] = 'wide'
user_size_pref = cherrypy.session['sizepref']
username = cherrypy.request.login
page_num = int(page_str)
next_page = page_num + 1
last_page = page_num - 1
scanner = ComicScanner()
image_list = scanner.reading_images(username)
num_pages = len(image_list)
if page_num == -1:
page_num = num_pages - 1
next_page = 0
last_page = num_pages - 2
if page_num == num_pages:
page_num = 0
next_page = 1
last_page = num_pages - 1
cookie_set = cherrypy.response.cookie
cookie_set[cookie_comic] = page_num
cookie_set[cookie_comic]['path'] = '/'
cookie_set[cookie_comic]['max-age'] = 2419200
return serve_template(templatename="read.html", pages=image_list, current_page=page_num, np=next_page, lp=last_page, nop=num_pages, size=user_size_pref, cc=cookie_comic)
def up_size_pref(self, pref):
cherrypy.session.load()
cherrypy.session['sizepref'] = pref
cherrypy.session.save()
return
