Python请求和持久会话

Question

我正在使用请求模块（Python 2.5的版本0.10.0）。我已经弄清楚了如何将数据提交到网站上的登录表单并检索会话密钥，但是我看不到在后续请求中使用此会话密钥的明显方法。有人可以在下面的代码中填写省略号还是建议其他方法？

>>> import requests

>>> login_data =  {'formPosted':'1', 'login_email':'[e-mail]', 'password':'pw'}

>>> r = requests.post('https://localhost/login.py', login_data)

>>> 

>>> r.text

u'You are being redirected <a href="profilePage?_ck=1349394964">here</a>'

>>> r.cookies

{'session_id_myapp': '127-0-0-1-825ff22a-6ed1-453b-aebc-5d3cf2987065'}

>>> 

>>> r2 = requests.get('https://localhost/profile_data.json', ...)

Answer 1

其他答案有助于了解如何维护此类会话。另外，我想提供一个类，该类可以使会话在脚本的不同运行（带有缓存文件）上得以维护。这意味着仅在需要时才执行正确的“登录”（超时或缓存中不存在会话）。它还支持在随后的“ get”或“ post”调用中的代理设置。已通过Python3测试。使用它作为您自己的代码的基础。GPL v3发行了以下片段import pickleimport datetimeimport osfrom urllib.parse import urlparseimport requests&nbsp; &nbsp;&nbsp;class MyLoginSession:&nbsp; &nbsp; """&nbsp; &nbsp; a class which handles and saves login sessions. It also keeps track of proxy settings.&nbsp; &nbsp; It does also maintine a cache-file for restoring session data from earlier&nbsp; &nbsp; script executions.&nbsp; &nbsp; """&nbsp; &nbsp; def __init__(self,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;loginUrl,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;loginData,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;loginTestUrl,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;loginTestString,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;sessionFileAppendix = '_session.dat',&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;maxSessionTimeSeconds = 30 * 60,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;proxies = None,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;userAgent = 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:40.0) Gecko/20100101 Firefox/40.1',&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;debug = True,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;forceLogin = False,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;**kwargs):&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; save some information needed to login the session&nbsp; &nbsp; &nbsp; &nbsp; you'll have to provide 'loginTestString' which will be looked for in the&nbsp; &nbsp; &nbsp; &nbsp; responses html to make sure, you've properly been logged in&nbsp; &nbsp; &nbsp; &nbsp; 'proxies' is of format { 'https' : 'https://user:pass@server:port', 'http' : ...&nbsp; &nbsp; &nbsp; &nbsp; 'loginData' will be sent as post data (dictionary of id : value).&nbsp; &nbsp; &nbsp; &nbsp; 'maxSessionTimeSeconds' will be used to determine when to re-login.&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; urlData = urlparse(loginUrl)&nbsp; &nbsp; &nbsp; &nbsp; self.proxies = proxies&nbsp; &nbsp; &nbsp; &nbsp; self.loginData = loginData&nbsp; &nbsp; &nbsp; &nbsp; self.loginUrl = loginUrl&nbsp; &nbsp; &nbsp; &nbsp; self.loginTestUrl = loginTestUrl&nbsp; &nbsp; &nbsp; &nbsp; self.maxSessionTime = maxSessionTimeSeconds&nbsp; &nbsp; &nbsp; &nbsp; self.sessionFile = urlData.netloc + sessionFileAppendix&nbsp; &nbsp; &nbsp; &nbsp; self.userAgent = userAgent&nbsp; &nbsp; &nbsp; &nbsp; self.loginTestString = loginTestString&nbsp; &nbsp; &nbsp; &nbsp; self.debug = debug&nbsp; &nbsp; &nbsp; &nbsp; self.login(forceLogin, **kwargs)&nbsp; &nbsp; def modification_date(self, filename):&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; return last file modification date as datetime object&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; t = os.path.getmtime(filename)&nbsp; &nbsp; &nbsp; &nbsp; return datetime.datetime.fromtimestamp(t)&nbsp; &nbsp; def login(self, forceLogin = False, **kwargs):&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; login to a session. Try to read last saved session from cache file. If this fails&nbsp; &nbsp; &nbsp; &nbsp; do proper login. If the last cache access was too old, also perform a proper login.&nbsp; &nbsp; &nbsp; &nbsp; Always updates session cache file.&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; wasReadFromCache = False&nbsp; &nbsp; &nbsp; &nbsp; if self.debug:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print('loading or generating session...')&nbsp; &nbsp; &nbsp; &nbsp; if os.path.exists(self.sessionFile) and not forceLogin:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; time = self.modification_date(self.sessionFile)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; # only load if file less than 30 minutes old&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; lastModification = (datetime.datetime.now() - time).seconds&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if lastModification < self.maxSessionTime:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; with open(self.sessionFile, "rb") as f:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self.session = pickle.load(f)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; wasReadFromCache = True&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if self.debug:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print("loaded session from cache (last access %ds ago) "&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; % lastModification)&nbsp; &nbsp; &nbsp; &nbsp; if not wasReadFromCache:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self.session = requests.Session()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self.session.headers.update({'user-agent' : self.userAgent})&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res = self.session.post(self.loginUrl, data = self.loginData,&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; proxies = self.proxies, **kwargs)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if self.debug:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print('created new session with login' )&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; self.saveSessionToCache()&nbsp; &nbsp; &nbsp; &nbsp; # test login&nbsp; &nbsp; &nbsp; &nbsp; res = self.session.get(self.loginTestUrl)&nbsp; &nbsp; &nbsp; &nbsp; if res.text.lower().find(self.loginTestString.lower()) < 0:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; raise Exception("could not log into provided site '%s'"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; " (did not find successful login string)"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; % self.loginUrl)&nbsp; &nbsp; def saveSessionToCache(self):&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; save session to a cache file&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; # always save (to update timeout)&nbsp; &nbsp; &nbsp; &nbsp; with open(self.sessionFile, "wb") as f:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; pickle.dump(self.session, f)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if self.debug:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; print('updated session cache-file %s' % self.sessionFile)&nbsp; &nbsp; def retrieveContent(self, url, method = "get", postData = None, **kwargs):&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; return the content of the url with respect to the session.&nbsp; &nbsp; &nbsp; &nbsp; If 'method' is not 'get', the url will be called with 'postData'&nbsp; &nbsp; &nbsp; &nbsp; as a post request.&nbsp; &nbsp; &nbsp; &nbsp; """&nbsp; &nbsp; &nbsp; &nbsp; if method == 'get':&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res = self.session.get(url , proxies = self.proxies, **kwargs)&nbsp; &nbsp; &nbsp; &nbsp; else:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; res = self.session.post(url , data = postData, proxies = self.proxies, **kwargs)&nbsp; &nbsp; &nbsp; &nbsp; # the session has been updated on the server, so also update in cache&nbsp; &nbsp; &nbsp; &nbsp; self.saveSessionToCache()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; return res使用上述类的代码片段可能如下所示：if __name__ == "__main__":&nbsp; &nbsp; # proxies = {'https' : 'https://user:pass@server:port',&nbsp; &nbsp; #&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;'http' : 'http://user:pass@server:port'}&nbsp; &nbsp; loginData = {'user' : 'usr',&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;'password' :&nbsp; 'pwd'}&nbsp; &nbsp; loginUrl = 'https://...'&nbsp; &nbsp; loginTestUrl = 'https://...'&nbsp; &nbsp; successStr = 'Hello Tom'&nbsp; &nbsp; s = MyLoginSession(loginUrl, loginData, loginTestUrl, successStr,&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;#proxies = proxies&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;)&nbsp; &nbsp; res = s.retrieveContent('https://....')&nbsp; &nbsp; print(res.text)&nbsp; &nbsp; # if, for instance, login via JSON values required try this:&nbsp; &nbsp; s = MyLoginSession(loginUrl, None, loginTestUrl, successStr,&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;#proxies = proxies,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;json = loginData)