本文整理汇总了Python中requests.sessions.Session类的典型用法代码示例。如果您正苦于以下问题:Python Session类的具体用法?Python Session怎么用?Python Session使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Session类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self, test, test_result, dns_resolve=True):
_Session.__init__(self)
self.verify = not os.getenv('BYPASS_SSL_CHECK')
self.test = test
self.test_result = test_result
self.loads_status = None, None, None, None
self.dns_resolve = dns_resolve
开发者ID:Awingu,项目名称:loads,代码行数:7,代码来源:measure.py
示例2: test_session_persistent_params
def test_session_persistent_params(self):
params = {'a': 'a_test'}
s = Session()
s.params = params
# Make 2 requests from Session object, should send header both times
r1 = s.get(httpbin('get'))
assert params['a'] in r1.content
params2 = {'b': 'b_test'}
r2 = s.get(httpbin('get'), params=params2)
assert params['a'] in r2.content
assert params2['b'] in r2.content
params3 = {'b': 'b_test', 'a': None, 'c': 'c_test'}
r3 = s.get(httpbin('get'), params=params3)
assert not params['a'] in r3.content
assert params3['b'] in r3.content
assert params3['c'] in r3.content
开发者ID:fredrik,项目名称:requests,代码行数:26,代码来源:test_requests.py
示例3: curl
def curl(self, method, endpoint, params=None):
url = '{scheme}://{host}{endpoint}'.format(scheme=self.SCHEME, host=self.HOST, endpoint=endpoint)
params = params or {}
session = Session()
request = Request(method, url, params=params)
request = request.prepare()
request.headers.update({
'X-Application-Key': self.KEY,
})
response = session.send(request)
return GforceResponse(response)
开发者ID:quaspas,项目名称:Gforce,代码行数:11,代码来源:client.py
示例4: test_session_persistent_headers
def test_session_persistent_headers(self):
heads = {'User-agent': 'Mozilla/5.0'}
s = Session()
s.headers = heads
# Make 2 requests from Session object, should send header both times
r1 = s.get(httpbin('user-agent'))
assert heads['User-agent'] in r1.content
r2 = s.get(httpbin('user-agent'))
assert heads['User-agent'] in r2.content
self.assertEqual(r2.status_code, 200)
开发者ID:cakebread,项目名称:requests,代码行数:14,代码来源:test_requests.py
示例5: __init__
class MeiPai:
def __init__(self):
self.session = Session()
self.session.headers = headers
def search(self, query, search_type='mv', page=1):
query = query.strip()
# topic
if query.startswith('#'):
topic = query.strip('#')
return self.get_videos_by_topic(topic)
cache_exists, result = check_cache_and_return_result(query=query, search_type=search_type, page=page)
if cache_exists:
return result
url = 'http://www.meipai.com/search/{search_type}?'.format(search_type=search_type) + \
urlencode({'q': query, 'page': page})
resp = self.session.request('GET', url)
html = BeautifulSoup(resp.content, 'html.parser')
if search_type == 'mv':
video_links = [div.attrs['data-video'].strip() for div in html.find_all(class_='content-l-video')]
# associated_words = self.word_association(query)
# print("你是否还想搜索:" + ",".join(associated_words))
result = video_links
elif search_type == 'topic':
result = [div.text.strip().strip('#') for div in html.find_all(class_='tcard-name')]
else:
result = []
cache_search_result(query, search_type, page, result)
return result
def get_videos_by_topic(self, topic_name):
"""
get top videos by topic
:param topic_name:
:return:
"""
topic = Topic(topic_name)
topic_id = topic.topic_id
url = "http://www.meipai.com/topics/hot_timeline?page=1&count=24&tid={topic_id}".format(topic_id=topic_id)
resp = self.session.request('GET', url)
result = json.loads(resp.text)
return [media['video'] for media in result['medias']]
# get associated words
def word_association(self, word):
url = 'http://www.meipai.com/search/word_assoc?' + urlencode({'q': word})
resp = self.session.request('GET', url)
return json.loads(resp.text)
开发者ID:PyBeaner,项目名称:meipai,代码行数:50,代码来源:api.py
示例6: EMDRUploader
class EMDRUploader(Thread):
def __init__(self, statsCollector):
Thread.__init__(self)
self._queue = Queue()
self.setDaemon(True)
self._session = Session()
self._session.headers.update({
"User-Agent": "CRESTMarketTrawler/{0} ([email protected])".format(VERSION)
})
self._pool = Pool(size=10)
self.statsCollector = statsCollector
def notify(self, regionID, typeID, orders):
self._queue.put((timestampString(), regionID, typeID, orders))
self.statsCollector.tally("emdr_send_queued")
queueSize = self._queue.qsize()
self.statsCollector.datapoint("emdr_queue_size", queueSize)
if queueSize > 100:
logger.error("EMDR submit queue is about {0} items long!".format(queueSize))
elif queueSize > 10:
logger.warn("EMDR submit queue is about {0} items long!".format(queueSize))
def run(self):
def submit(generationTime, regionID, typeID, orders):
uudif = json.dumps(EMDROrdersAdapter(generationTime, regionID, typeID, orders))
res = self._session.post("http://upload.eve-emdr.com/upload/", data=uudif)
self.statsCollector.tally("emdr_sent")
if res.status_code != 200:
logger.error("Error {0} submitting to EMDR: {1}".format(res.status_code, res.content))
self.statsCollector.tally("emdr_errored")
while True:
(generationTime, regionID, typeID, orders) = self._queue.get()
self._pool.spawn(submit, generationTime, regionID, typeID, orders)
开发者ID:jrausch12,项目名称:CrestTrawler,代码行数:34,代码来源:emdr.py
示例7: fetch_data
def fetch_data(dl_type=None, **kwargs):
"""
Fetch Receipts, Expenditures, and Committees.
dl_type is one of those three choices.
kwargs depend on the choice.
Receipts and Expenditures need start_date and end_date for search.
Committees need a name_start kwarg to pass into the search.
Seems like the maximum that you can get is about 250,000 records at a time.
"""
s = Session()
post_data = {
'__EVENTTARGET': 'ctl00$ContentPlaceHolder1$btnText',
'ctl00$pnlMenu_CollapsiblePanelExtender_ClientState': 'true',
'ctl00$AccordionStateBoardMenu_AccordionExtender_ClientState': '0',
'ctl00$mtbSearch': '',
'ctl00$AccordionPaneStateBoardMenu_content$AccordionMainContent_AccordionExtender_ClientState': '-1',
'hiddenInputToUpdateATBuffer_CommonToolkitScripts': '1',
'__EVENTARGUMENT': '',
'__VIEWSTATEGENERATOR': 'E8D1F59A'
}
if dl_type == 'Receipts':
CONT_GET_PARAMS['RcvDate'] = kwargs['start_date']
CONT_GET_PARAMS['RcvDateThru'] = kwargs['end_date']
url = '%s/DownloadList.aspx?%s' % (BASE_URL, urlencode(CONT_GET_PARAMS))
elif dl_type == 'Committees':
COMM_GET_PARAMS['Name'] = kwargs['name_start']
url = '%s/DownloadList.aspx?%s' % (BASE_URL, urlencode(COMM_GET_PARAMS))
elif dl_type == 'Expenditures':
EXP_GET_PARAMS['ExpendedDate'] = kwargs['start_date']
EXP_GET_PARAMS['ExpendedDateThru'] = kwargs['end_date']
url = '%s/DownloadList.aspx?%s' % (BASE_URL, urlencode(EXP_GET_PARAMS))
elif dl_type == 'Candidates':
url = 'http://www.elections.state.il.us/ElectionInformation/CandDataFile.aspx?id=%s' % kwargs['election_id']
g = s.get(url)
if 'Unexpected errors occurred trying to populate page' in g.content:
return None
soup = BeautifulSoup(g.content)
view_state = soup.find('input', attrs={'id': '__VIEWSTATE'}).get('value')
event_val = soup.find('input', attrs={'id': '__EVENTVALIDATION'}).get('value')
post_data['__VIEWSTATE'] = view_state
post_data['__EVENTVALIDATION'] = event_val
dl_page = s.post(url, data=post_data)
if dl_page.status_code == 200:
return dl_page.content
else:
return None
开发者ID:influence-usa,项目名称:campaign-finance_state_IL,代码行数:47,代码来源:scraper.py
示例8: __init__
def __init__(self, device_key):
self.base_url = '{0}/{1}/'.format(self.API_URL, self.API_VERSION)
self.session = Session()
self.session.auth = KeyAuth(device_key)
self.session.headers.update({
'Content-Type': 'application/json'
})
开发者ID:exilium,项目名称:pifarm-device-python,代码行数:8,代码来源:api.py
示例9: delete
def delete(self, url, **kwargs):
return Session.delete(
self,
self._get_resource_uri(url),
**self._set_default_timeout(
**kwargs
)
)
开发者ID:WesleyPeng,项目名称:uiXautomation,代码行数:8,代码来源:restclient.py
示例10: configure_http_session
def configure_http_session(size=20, max_retries=1, _session=None):
"""
Return a :class:`requests.Session` object configured with
a :class:`requests.adapters.HTTPAdapter` (connection pool)
for http and https connections.
:param size: The connection pool and maximum size.
:type size: int
:param max_retries: The maximum number of retries for each connection.
:type max_retries: int
:param _session: Test-only hook to provide a pre-configured session.
"""
if _session is not None:
return _session
adapter = HTTPAdapter(
pool_connections=size,
pool_maxsize=size,
max_retries=max_retries,
)
session = Session()
session.mount("http://", adapter)
session.mount("https://", adapter)
session.max_redirects = 1
session.verify = certifi.where()
return session
开发者ID:amjadm61,项目名称:ichnaea,代码行数:28,代码来源:http.py
示例11: patch
def patch(self, url, data=None, **kwargs):
return Session.patch(
self,
self._get_resource_uri(url),
data,
**self._set_default_timeout(
**kwargs
)
)
开发者ID:WesleyPeng,项目名称:uiXautomation,代码行数:9,代码来源:restclient.py
示例12: send
def send(self, request, **kwargs):
request.url = resolve(request.url)
# started
start = datetime.datetime.utcnow()
res = _Session.send(self, request, **kwargs)
res.started = start
res.method = request.method
_measure(res)
return res
开发者ID:matrixise,项目名称:loads,代码行数:9,代码来源:measure.py
示例13: APIClient
class APIClient(BaseAPIClient):
verify = True
base_url = None
def __init__(self, *args, **kwargs):
self.session = Session()
spec = self.call(SpecEndpoint())
super(APIClient, self).__init__(*args, spec=spec, **kwargs)
def make_request(self, endpoint, request):
request.url = self.base_url + request.url
prepared = self.session.prepare_request(request)
return self.session.send(prepared,
stream=False,
timeout=None,
verify=self.verify,
cert=None,
proxies={},
allow_redirects=True)
开发者ID:cosmic-api,项目名称:cosmic.py,代码行数:19,代码来源:client.py
示例14: __init__
def __init__(self, statsCollector):
Thread.__init__(self)
self._queue = Queue()
self.setDaemon(True)
self._session = Session()
self._session.headers.update({
"User-Agent": "CRESTMarketTrawler/{0} ([email protected])".format(VERSION)
})
self._pool = Pool(size=10)
self.statsCollector = statsCollector
开发者ID:jrausch12,项目名称:CrestTrawler,代码行数:10,代码来源:emdr.py
示例15: __init__
def __init__(self, statsCollector):
Thread.__init__(self)
self._queue = Queue(EMDR_QUEUE_SIZE)
self.setDaemon(True)
self._session = Session()
self._session.headers.update({
"User-Agent": USER_AGENT_STRING
})
self._pool = Pool(size=10)
self.statsCollector = statsCollector
开发者ID:jamesremuscat,项目名称:CRESTMarketTrawler,代码行数:10,代码来源:emdr.py
示例16: __init__
def __init__(self, service, region_name, host, auth, proxies=None):
self.service = service
self.session = self.service.session
self.region_name = region_name
self.host = host
self.verify = True
self.auth = auth
if proxies is None:
proxies = {}
self.proxies = proxies
self.http_session = Session()
开发者ID:jonparrott,项目名称:botocore,代码行数:11,代码来源:endpoint.py
示例17: send
def send(self, request, **kwargs):
"""Do the actual request from within the session, doing some
measures at the same time about the request (duration, status, etc).
"""
# attach some information to the request object for later use.
start = datetime.datetime.utcnow()
res = _Session.send(self, request, **kwargs)
res.started = start
res.method = request.method
self._analyse_request(res)
return res
开发者ID:loads,项目名称:loads-tester,代码行数:11,代码来源:measure.py
示例18: __init__
def __init__(self, authentication_key, change_source):
super(PortalConnection, self).__init__()
self._authentication_handler = \
_QueryStringAuthenticationHandler(authentication_key)
self._change_source = change_source
self._session = Session()
self._session.headers['User-Agent'] = _USER_AGENT
http_adapter = HTTPAdapter(max_retries=_HTTP_CONNECTION_MAX_RETRIES)
self._session.mount('', http_adapter)
开发者ID:iantaylor,项目名称:hubspot-connection,代码行数:12,代码来源:__init__.py
示例19: __init__
def __init__(
self,
base_url,
port=None,
username=None,
password=None,
**kwargs
):
Session.__init__(self)
Client.__init__(
self, base_url, port,
username, password, **kwargs
)
self.verify = False
self._set_auth(
username, password
)
urllib3.disable_warnings()
开发者ID:WesleyPeng,项目名称:uiXautomation,代码行数:22,代码来源:restclient.py
示例20: Api
class Api(object):
API_URL = 'http://192.168.33.10' # todo: add to config
API_VERSION = 'v1'
def __init__(self, device_key):
self.base_url = '{0}/{1}/'.format(self.API_URL, self.API_VERSION)
self.session = Session()
self.session.auth = KeyAuth(device_key)
self.session.headers.update({
'Content-Type': 'application/json'
})
def request(self, method, url, **kwargs):
"""Constructs and sends a Request to the Pinaple API."""
full_url = urljoin(self.base_url, url)
if 'data' in kwargs:
kwargs['data'] = self._encode_data(kwargs['data'])
return super(Api, self).request(method, full_url, **kwargs)
def _encode_data(self, data, **kwargs):
"""Returns data encoded as JSON using a custom encoder."""
encoder = JSONEncoder(**kwargs) if kwargs else self._json_encoder
return encoder.encode(data)
def test(self):
url = urljoin(self.base_url, 'functions/test')
response = self.session.get(url)
return response
def login(self):
url = urljoin( self.base_url, 'login' )
response = self.session.post( url )
if response.status_code is not 200:
print('[error] device is not authorized')
exit()
data = response.json()
self.session.auth = SessionAuth(data['session_token'])
开发者ID:exilium,项目名称:pifarm-device-python,代码行数:39,代码来源:api.py
注:本文中的requests.sessions.Session类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论