我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用cherrypy.session()。
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')
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 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
def __init__(self, id=None, **kwargs): self.id_observers = [] self._data = {} for k, v in kwargs.items(): setattr(self, k, v) self.originalid = id self.missing = False if id is None: if self.debug: cherrypy.log('No id given; making a new one', 'TOOLS.SESSIONS') self._regenerate() else: self.id = id if self._exists(): if self.debug: cherrypy.log('Set id to %s.' % id, 'TOOLS.SESSIONS') else: if self.debug: cherrypy.log('Expired or malicious session %r; ' 'making a new one' % id, 'TOOLS.SESSIONS') # Expired or malicious session. Make a new one. # See https://github.com/cherrypy/cherrypy/issues/709. self.id = None self.missing = True self._regenerate()
def now(self): """Generate the session specific concept of 'now'. Other session providers can override this to use alternative, possibly timezone aware, versions of 'now'. """ return datetime.datetime.now()
def regenerate(self): """Replace the current session (with a new id).""" self.regenerated = True self._regenerate()
def _regenerate(self): if self.id is not None: if self.debug: cherrypy.log( 'Deleting the existing session %r before ' 'regeneration.' % self.id, 'TOOLS.SESSIONS') self.delete() old_session_was_locked = self.locked if old_session_was_locked: self.release_lock() if self.debug: cherrypy.log('Old lock released.', 'TOOLS.SESSIONS') self.id = None while self.id is None: self.id = self.generate_id() # Assert that the generated id is not already stored. if self._exists(): self.id = None if self.debug: cherrypy.log('Set id to generated %s.' % self.id, 'TOOLS.SESSIONS') if old_session_was_locked: self.acquire_lock() if self.debug: cherrypy.log('Regenerated lock acquired.', 'TOOLS.SESSIONS')
def generate_id(self): """Return a new session id.""" return random20()
def load(self): """Copy stored session data into this session instance.""" data = self._load() # data is either None or a tuple (session_data, expiration_time) if data is None or data[1] < self.now(): if self.debug: cherrypy.log('Expired session %r, flushing data.' % self.id, 'TOOLS.SESSIONS') self._data = {} else: if self.debug: cherrypy.log('Data loaded for session %r.' % self.id, 'TOOLS.SESSIONS') self._data = data[0] self.loaded = True # Stick the clean_thread in the class, not the instance. # The instances are created and destroyed per-request. cls = self.__class__ if self.clean_freq and not cls.clean_thread: # clean_up is an instancemethod and not a classmethod, # so that tool config can be accessed inside the method. t = cherrypy.process.plugins.Monitor( cherrypy.engine, self.clean_up, self.clean_freq * 60, name='Session cleanup') t.subscribe() cls.clean_thread = t t.start() if self.debug: cherrypy.log('Started cleanup thread.', 'TOOLS.SESSIONS')
def delete(self): """Delete stored session data.""" self._delete() if self.debug: cherrypy.log('Deleted session %s.' % self.id, 'TOOLS.SESSIONS') # -------------------- Application accessor methods -------------------- #
def acquire_lock(self): """Acquire an exclusive lock on the currently-loaded session data.""" self.locked = True self.locks.setdefault(self.id, threading.RLock()).acquire()
def release_lock(self): """Release the lock on the currently-loaded session data.""" self.locks[self.id].release() self.locked = False
def _get_file_path(self): f = os.path.join(self.storage_path, self.SESSION_PREFIX + self.id) if not os.path.abspath(f).startswith(self.storage_path): raise cherrypy.HTTPError(400, 'Invalid session id in cookie.') return f
def _save(self, expiration_time): assert self.locked, ('The session was saved without being locked. ' "Check your tools' priority levels.") f = open(self._get_file_path(), 'wb') try: pickle.dump((self._data, expiration_time), f, self.pickle_protocol) finally: f.close()
def _delete(self): assert self.locked, ('The session deletion without being locked. ' "Check your tools' priority levels.") try: os.unlink(self._get_file_path()) except OSError: pass
def release_lock(self, path=None): """Release the lock on the currently-loaded session data.""" self.lock.release() self.lock.remove() self.locked = False
def clean_up(self): """Clean up expired sessions.""" now = self.now() # Iterate over all session files in self.storage_path for fname in os.listdir(self.storage_path): if (fname.startswith(self.SESSION_PREFIX) and not fname.endswith(self.LOCK_SUFFIX)): # We have a session file: lock and load it and check # if it's expired. If it fails, nevermind. path = os.path.join(self.storage_path, fname) self.acquire_lock(path) if self.debug: # This is a bit of a hack, since we're calling clean_up # on the first instance rather than the entire class, # so depending on whether you have "debug" set on the # path of the first session called, this may not run. cherrypy.log('Cleanup lock acquired.', 'TOOLS.SESSIONS') try: contents = self._load(path) # _load returns None on IOError if contents is not None: data, expiration_time = contents if expiration_time < now: # Session expired: deleting it os.unlink(path) finally: self.release_lock(path)
def close(): """Close the session object for this request.""" sess = getattr(cherrypy.serving, 'session', None) if getattr(sess, 'locked', False): # If the session is still locked we release the lock sess.release_lock() if sess.debug: cherrypy.log('Lock released on close.', 'TOOLS.SESSIONS')
def expire(): """Expire the current session cookie.""" name = cherrypy.serving.request.config.get( 'tools.sessions.name', 'session_id') one_year = 60 * 60 * 24 * 365 e = time.time() - one_year cherrypy.serving.response.cookie[name]['expires'] = httputil.HTTPDate(e)
def do_logout(self, from_page='..', **kwargs): """Logout. May raise redirect, or return True if request handled.""" sess = cherrypy.session username = sess.get(self.session_key) sess[self.session_key] = None if username: cherrypy.serving.request.login = None self.on_logout(username) raise cherrypy.HTTPRedirect(from_page)
def validate_password(self, login, password): if login in users : if encrypt(password) == users[login] : cherrypy.session['username'] = login cherrypy.session['database'] = userdatabase(login) return True return False
def haddfav(self, id=""): if id == "" or id == None : html += "Error missing id" return html if id == "0" : html += "0 is reserved, sorry" return html (name, genre, url) = getradio(id) if 'Fav' in genre: genre = genre.replace(', Fav', '') star = False else: genre += ', Fav' star = True if addgen(id, genre) == False: return '' (name, genre, url) = getradio(id) cherrypy.session['playing'] = id html = '<h3>Now Playing: ' html += '''<a href="%s">%s</a>''' % (url, name) html += '''<a href="#" onClick="fplayradio('%s')">''' % url 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 if star: html += '''Starred''' html += '''</h3>''' return html