pip install httpx  # 安裝庫
pip install httpx[http2]  # 獲取http2的支持
pip install httpx[brotli]  # 包括可選的 brotli 解碼器支持

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"
import httpx
from fake_useragent import UserAgent

headers = {
    "user-agent": UserAgent().random,
}
params = {
    "wd": "python"  # 輸入百度搜索的內(nèi)容
}
resp = httpx.get("https://www.baidu.com/s", params=params, headers=headers, cookies=None, proxies=None)  # 和原來requests的使用方法類似
resp.encoding = resp.charset_encoding  # 根據(jù)文檔的編碼還對文檔進行編碼
print(resp.text)  # 獲取數(shù)據(jù)信息

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"
import httpx

data = {'key1': 'value1', 'key2': 'value2'}
r = httpx.post("https://httpbin.org/post", data=data)
print(r.text)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"
import httpx

files = {'upload-file': open('a.jpg', 'rb')}
# 也可以通過元組來指定數(shù)據(jù)類型
# files = {'upload-file': ('report.xls', open('report.xls', 'rb'), 'application/vnd.ms-excel')}
r = httpx.post("https://httpbin.org/post", files=files)
print(r.text)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"
import httpx

data = {'integer': 123, 'boolean': True, 'list': ['a', 'b', 'c']}
r = httpx.post("https://httpbin.org/post", json=data)
print(r.text)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"
import httpx

content = b'Hello, world'
r = httpx.post("https://httpbin.org/post", content=content, headers={
    "Content-Type": "application/octet-stream",
})
print(r.text)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx

resp = httpx.request("GET", "https://www.baidu.com")
if resp.status_code == httpx.codes.OK:
    print(resp.text)  # 如果請求成功

print(resp.raise_for_status())  # 判斷響應是否成功，成功返回None，失敗則報錯

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx

with httpx.stream("GET", "https://www.example.com") as r:
    for data in r.iter_bytes():  # 流式傳輸響應的二進制內(nèi)容
        # for text in r.iter_text():  # 獲取全部的文本內(nèi)容
        # for line in r.iter_lines():  # 逐行獲取傳輸響應的文本內(nèi)容
        # for chunk in r.iter_raw():  # 獲取編碼前的原始數(shù)據(jù)
        # if r.headers['Content-Length'] < TOO_LONG:  # 有條件的加載內(nèi)容
        print(data)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx
# 獲取cookie
r = httpx.get('https://httpbin.org/cookies/set?chocolate=chip')
print(r.cookies['chocolate'])  # 獲取請求中的cookie

# 設置cookie
cookies_1 = {"peanut": "butter"}

cookies_2 = httpx.Cookies()
cookies_2.set('cookie_on_domain', 'hello, there!', domain='httpbin.org')
cookies_2.set('cookie_off_domain', 'nope.', domain='example.org')
r = httpx.get('http://httpbin.org/cookies', cookies=cookies_2)
print(r.json())

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx
r = httpx.get('http://github.com/')
print(r.status_code)
print(r.history)  # 查看重定向的記錄
print(r.next_request)  # 獲取到重定向以后的請求對象
resp = httpx.Client().send(r.next_request) # 對請求對象發(fā)送請求
print(resp.text)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx
r = httpx.get('http://github.com/', follow_redirects=True)
print(r.history)  # 查看重定向記錄
print(r.url)  # 獲取請求的url
print(r.text)  # 獲取請求數(shù)據(jù)

httpx.get('https://github.com/', timeout=0.001)  # 同時也可以禁止超時行為
httpx.get('https://github.com/', timeout=None)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx
httpx.get("https://example.com", auth=("my_user", "password123"))  # 驗證方法一
auth = httpx.DigestAuth("my_user", "password123")  # 驗證方法二
httpx.get("https://example.com", auth=auth)

# 使用方法1
with httpx.Client() as client:
    ...
    
# 使用方法2
client = httpx.Client()
try:
    ...
finally:
    client.close()

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx

# 共用請求頭
url = 'http://httpbin.org/headers'
headers = {'user-agent': 'my-app/0.0.1'}
with httpx.Client(headers=headers) as client:
    # 這里面的所有請求的請求頭都包含{'user-agent': 'my-app/0.0.1'}
    r = client.get(url)

print(r.json()['headers']['User-Agent'])

# 共用 + 私有
headers = {'X-Auth': 'from-client'}
params = {'client_id': 'client1'}
with httpx.Client(headers=headers, params=params) as client:
    headers_ = {'X-Custom': 'from-request'}
    params_ = {'request_id': 'request1'}
    r = client.get('https://example.com', headers=headers_,
                   params=params_)  # 這個參數(shù)結(jié)合了headers+headers_ , params+params_，但是只限于params和headers，對于所有其他參數(shù)，內(nèi)部請求級別的值優(yōu)先

print(r.request.url)
print(r.request.headers['X-Auth'])
print(r.request.headers['X-Custom'])

# 優(yōu)先級
with httpx.Client(auth=('tom', 'mot123')) as client:
    r = client.get('https://example.com', auth=('alice', 'ecila123'))

_, _, auth = r.request.headers['Authorization'].partition(' ')
import base64

print(base64.b64decode(auth))

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import httpx

with httpx.Client(base_url='http://httpbin.org') as client:
     r = client.get('/headers')

print(r.request.url)

import httpx
import chardet  # pip install chardet

def autodetect(content):
    return chardet.detect(content).get("encoding")  # 對html的編碼進行自動的檢測

# Using a client with character-set autodetection enabled.
client = httpx.Client(default_encoding=autodetect)
response = client.get(...)
print(response.encoding)  # This will either print the charset given in
                          # the Content-Type charset, or else the auto-detected
                          # character set.
print(response.text)

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

from flask import Flask
import httpx

app = Flask(__name__)

@app.route("/")
def hello():
    return "Hello World!"

with httpx.Client(app=app, base_url="http://localhost") as client:
    # base_url：指定app的根路由
    r = client.get("/")  # 獲取根路由下的響應數(shù)據(jù)
    print(r.text)
    assert r.status_code == 200  # 斷言
    assert r.text == "Hello World!"

# Instantiate a client that makes WSGI requests with a client IP of "1.2.3.4".
transport = httpx.WSGITransport(app=app, remote_addr="1.2.3.4")
with httpx.Client(transport=transport, base_url="http://testserver") as client:
    ...

request = httpx.Request("GET", "https://example.com")

with httpx.Client() as client:
    response = client.send(request)
    ...

headers = {"X-Api-Key": "...", "X-Client-ID": "ABC123"}

with httpx.Client(headers=headers) as client:
    request = client.build_request("GET", "https://api.example.com")

    print(request.headers["X-Client-ID"])  # "ABC123"

    # Don't send the API key for this particular request.
    del request.headers["X-Api-Key"]

    response = client.send(request)
    ...

def log_request(request):
    print(f"Request event hook: {request.method} {request.url} - Waiting for response")

def log_response(response):
    request = response.request
    print(f"Response event hook: {request.method} {request.url} - Status {response.status_code}")

client = httpx.Client(event_hooks={'request': [log_request], 'response': [log_response]})  # 綁定鉤子函數(shù)

def raise_on_4xx_5xx(response):
    response.raise_for_status()

client = httpx.Client(event_hooks={'response': [raise_on_4xx_5xx]})

def add_timestamp(request):
    request.headers['x-request-timestamp'] = datetime.now(tz=datetime.utc).isoformat()

client = httpx.Client(event_hooks={'request': [add_timestamp]})

client = httpx.Client()
client.event_hooks['request'] = [log_request]
client.event_hooks['response'] = [log_response, raise_on_4xx_5xx]

import tempfile

import httpx
from tqdm import tqdm

with tempfile.NamedTemporaryFile() as download_file:  # 創(chuàng)建一個臨時文件。程序結(jié)束就刪除
    url = "https://speed.hetzner.de/100MB.bin"
    with httpx.stream("GET", url) as response:  # 使用流發(fā)送請求
        total = int(response.headers["Content-Length"])

        with tqdm(total=total, unit_scale=True, unit_divisor=1024, unit="B") as progress:
            num_bytes_downloaded = response.num_bytes_downloaded
            for chunk in response.iter_bytes():
                download_file.write(chunk)
                progress.update(response.num_bytes_downloaded - num_bytes_downloaded)
                num_bytes_downloaded = response.num_bytes_downloaded

httpx.get('https://example.org/', trust_env=False)

import os
os.environ["NETRC"] = "my_default_folder/.my_netrc"

machine netrcexample.org
login example-username
password example-password

...

client = httpx.Client(trust_env=False)

with httpx.Client(proxies="http://localhost:8030") as client:
    ...

proxies = {
    "http://": "http://localhost:8030",
    "https://": "http://localhost:8031",
}

with httpx.Client(proxies=proxies) as client:
    ...

proxies = {
    "http://": "http://username:password@localhost:8030",
    # ...
}

proxies = {
    "all://": "http://localhost:8030",
}

proxies = {
    "http://": "http://localhost:8030",
    "https://": "http://localhost:8031",
}

# 代理域“example.com”上的所有請求，讓其他請求通過... 
proxies = {
    "all://example.com": "http://localhost:8030",
}
# 代理域“example.com”上的 HTTP 請求，讓 HTTPS 和其他請求通過...
proxies = {
    "http://example.com": "http://localhost:8030",
}
# 將所有請求代理到“example.com”及其子域，讓其他請求通過...
proxies = {
    "all://*example.com": "http://localhost:8030",
}
# 代理所有請求到“example.com”的嚴格子域，讓“example.com”等請求通過...
proxies = {
    "all://*.example.com": "http://localhost:8030",
}

proxies = {
    "https://example.com:1234": "http://localhost:8030",
}

proxies = {
    "all://*:1234": "http://localhost:8030",
}

proxies = {
    # Route requests through a proxy by default...
    "all://": "http://localhost:8031",
    # Except those for "example.com".
    "all://example.com": None,
}

# 普通的代理
import requests

proxies = {
  'http': 'http://10.10.1.10:3128',
  'https': 'http://10.10.1.10:1080',
}  

requests.get('http://example.org', proxies=proxies)

# 權(quán)限認證
proxies = {'http': 'http://user:pass@10.10.1.10:3128/'}

# 給特定的方案和主機提供代理，這將匹配對給定方案和確切主機名的任何請求。
proxies = {'http://example.org': 'http://10.10.1.10:5323'}  # 其為一個簡單的路由功能，進行簡單的代理分發(fā)

async with aiohttp.ClientSession() as session:
    proxy_auth = aiohttp.BasicAuth('user', 'pass')
    async with session.get("http://python.org",
                           proxy="http://proxy.com",
                           proxy_auth=proxy_auth) as resp:
        print(resp.status)

def start_requests(self):
    for url in self.start_urls:
        return Request(url=url, callback=self.parse,
                       headers={"User-Agent": "scrape web"},
                       meta={"proxy": "http:/154.112.82.262:8050"})  

# 權(quán)限認證：
# request.headers["Proxy-Authorization"] = basic_auth_header("<proxy_user>", "<proxy_pass>")

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
__author__ = "A.L.Kun"
__file__ = "demo01.py"
__time__ = "2022/9/9 7:55"

import asyncio
import httpx

async def test():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://www.baidu.com")

    print(r)

tasks = [test() for i in range(100)]
asyncio.run(asyncio.wait(tasks))

async with httpx.AsyncClient() as client:
    ...

client = httpx.AsyncClient()
...
await client.aclose()

client = httpx.AsyncClient()
async with client.stream('GET', 'https://www.example.com/') as response:
    async for chunk in response.aiter_bytes():
        ...

import httpx
from starlette.background import BackgroundTask
from starlette.responses import StreamingResponse

client = httpx.AsyncClient()

async def home(request):
    req = client.build_request("GET", "https://www.example.com/")
    r = await client.send(req, stream=True)
    return StreamingResponse(r.aiter_text(), background=BackgroundTask(r.aclose))

async def upload_bytes():
    ...  # yield byte content

await client.post(url, content=upload_bytes())

import asyncio
import httpx

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://www.example.com/')
        print(response)

asyncio.run(main())

import httpx
import trio

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://www.example.com/')
        print(response)

trio.run(main)

import httpx
import anyio

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://www.example.com/')
        print(response)

anyio.run(main, backend='trio')

from starlette.applications import Starlette
from starlette.responses import HTMLResponse
from starlette.routing import Route

async def hello(request):
    return HTMLResponse("Hello World!")

app = Starlette(routes=[Route("/", hello)])

import httpx
async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
    r = await client.get("/")
    assert r.status_code == 200
    assert r.text == "Hello World!"

# Instantiate a client that makes ASGI requests with a client IP of "1.2.3.4",
# on port 123.
transport = httpx.ASGITransport(app=app, client=("1.2.3.4", 123))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
    ...