python requests 模块

Posted on Posted in python

requests 作者网站 帅的吊炸天

准备工作

python --version  
pip --version  
  
pip install gunicorn httpbin  
gunicorn httpbin:app  
  
gunicorn -b :10000 httpbin:app  

urllib2 和 requests 举例

# urllib2 demo  
  
#!/usr/bin/env python  
# -*- coding: utf-8 -*-  
import urllib  
import urllib2  
  
URL_IP = 'http://123.207.137.130:10000/ip'  
URL_GET = 'http://123.207.137.130:10000/get'  
  
def use_simple_urllib2():  
    response = urllib2.urlopen(URL_IP)  
    print '>>> Response Headers:'  
    print response.info()  
    print '>>> Response Body:'  
    print ''.join([line for line in response.readlines()])  
  
def use_params_urllib2():  
    # 构建请求参数  
    params = urllib.urlencode({'param1':'hello','param2':'world'})  
    print 'Request pamrams:'  
    print params  
    # 发送请求  
    response = urllib2.urlopen('?'.join([URL_GET,'%s']) % params)  
    # 处理响应  
    print '>>> Response Headers:'  
    print response.info()  
    print '>>> Status Code:'  
    print response.getcode()  
    print '>>> Response Body:'  
    print ''.join([line for line in response.readlines()])  
if __name__ == '__main__':  
    print '>>> Use simple urllib2'  
    use_simple_urllib2()  
    print '>>> Use params urllib2:'  
    use_params_urllib2()  
  
# requests demo  
  
#!/usr/bin/env python  
# -*- coding: utf-8 -*-  
import requests  
  
URL_IP = 'http://123.207.137.130:10000/ip'  
URL_GET = 'http://123.207.137.130:10000/get'  
  
def use_simple_requests():  
    response = requests.get(URL_IP)  
    print '>>> Response Headers:'  
    print response.headers  
    print '>>> Response Body:'  
    print response.text  
  
def use_params_requests():  
    params = {'param1':'hello','param2':'world'}  
    response = requests.get(URL_IP, params = params)  
    print '>>> Response Headers:'  
    print response.headers  
    print '>>> Status Code:'  
    print response.status_code  
    print reaponse.reason  
    print '>>> Response Body:'  
    print response.json()  
  
if __name__ == '__main__':  
    print '>>> Use simple requests'  
    use_simple_requests()  
    print '>>> Use params requests:'  
    use_params_requests()  

requests 请求方法

#!/usr/bin/env python  
import json  
import requests  
from requests import exceptions  
  
URL = 'https://api.github.com'  
  
def build_uri(endpoint):  
    return '/'.join([URL, endpoint])  
  
def better_print(json_str):  
    return json.dumps(json.loads(json_str), indent = 4)  
  
def request_method():  
    #response = requests.get(build_uri('users/liuhonghe'))  
    response = requests.get(build_uri('user/emails'), auth=('username','password')) # 这里user就不是users了   
    print better_print(response.text)  
  
def params_request():  
    response = requests.get(build_uri('users'),params={'since':11}) # 利用params发送参数  
    print better_print(response.text)  
    print response.request.headers  
    print response.url  
  
def json_request():  
    #response = requests.patch(build_uri('user'), auth=('username','password'),json={'name':'liuhonghe01', 'email':'abc@liuhonghe.me'}) # patch修改自己github上的相关信息  
    response = requests.post(build_rui('user/emails'), auth=('username','password'),json = ['a@liuhonghe.me'])  
    print better_print(response.text)  
    print response.request.headers  
    print response.request.body  
    print response.request.status_code  
  
def timeout_request(): # requests 异常处理  
    try:  
        response = requests.get(build_uri('user/emails'), timeout=0.1) # timeout=(3,7) 这是另一种写法,3秒,7秒  
        response.raise_for_status()  
    except exceptions.Timeout as e:  
        print e.message  
    except exceptions.HTTPError as e:  
        print e.message  
    else:  
        print response.text  
        print response.status_code  
if __name__ == '__main__':  
    #request_method()  
    #params_request()  
    timeout_request()  

自定义 request

def hard_requests():    
    from requests import Request, Session    
    s = Session()    
    headers = {'User-Agent':'fake1.3.4'}    
    req = Request('GET',build_uri('user/emails'), auth=('username','password'), headers = headers)    
    prepped = req.prepare()    
    print prepped.body    
    print prepped.headers    
    
    resp = s.send(prepped, timeout=5)    
    print resp.status_code    
    print resp.request.headers    
    print resp.text    

响应基本api

#!/usr/bin/env python  
import requests  
response = requests.get('http://api.github.com')  
print response.status_code  
print response.reason  
print response.headers  
print response.url  
print response.history  
print dir(response)  
print response.elapsed  
print response.request  
print response.headers  
print response.encoding  
print response.raw.read(100)  
print response.content  
print response.text  
print response.json()  
print response.json()['team_url']  

下载图片

#!/usr/bin/env python  
import requests  
  
def download_image():  
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36"}  
    url = "https://ss0.bdstatic.com/5aV1bjqh_Q23odCf/static/superman/img/logo/logo_white.png"  
    response = requests.get(url, headers = headers, stream=True)  
    print response.status_code, response.reason  
    print response.headers  
    with open('demo.jpg','wb') as fd:  
        for chunk in response.iter_content(128):  
            fd.write(chunk)  
  
def download_image_improved():  
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36"}  
        url = "https://ss0.bdstatic.com/5aV1bjqh_Q23odCf/static/superman/img/logo/logo_white.png"  
        response = requests.get(url, headers = headers, stream=True) # 增加了流传递  
    from contextlib import closing # 这个模块管理自动关闭流  
    with closing(requests.get(url, headers = headers, stream=True)) as response: # 创建response  
        with open('demo1.jpg','wb') as fd:  
                    for chunk in response.iter_content(128):  
                            fd.write(chunk)  
download_image_improved()  

requests 事件钩子

#!/usr/bin/env python  
import requests  
  
def get_key_info(response, *args, **kwargs):  
        print response.headers['Content-Type']  
  
def main():  
        requests.get('http://www.baidu.com', hooks = dict(response=get_key_info))  
  
main()  

代理 proxy

# pip install requests[socks]  
import requests  
proxies = {'http':'socks5://127.0.0.1:1080', 'https':'socks5://127.0.0.1:1080'}  
url = "https://www.facebook.com"  
response = requests.get(url, proxies=proxies, timeout=10)  
print response.status_code  
» 转载请注明来源:若我若鱼 » python requests 模块

Leave a Reply

Your email address will not be published. Required fields are marked *

one × 1 =