Python网络编程实战第4篇
Python网络编程实战第4篇。本文讲解Python socket网络编程、HTTP请求、API调用、异步网络编程等核心技能。Introduction:网络编程是Python实现分布式系统、爬虫、API服务的基础技能。Python的标准库提供了socket、urllib、http.client等模块,可以直接实现TCP/UDP通信、HTTP请求等常见网络功能。结合asyncio更可以实现高并发网络编程。本文涵盖socket基础、TCP/UDP编程、HTTP请求、RESTful API调用、爬虫基础、异步HTTP等核心知识点,配完整代码示例。
socket基础:Python的socket模块提供底层网络通信能力。import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect(('example.com', 80)) s.send(b'GET / HTTP/1.0\r\n\r\n') response = s.recv(4096) s.close() print(response.decode()) 这是最基本的TCP客户端示例,连接远程服务器并发送HTTP请求。
TCP服务器编程:import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(('0.0.0.0', 8080)) s.listen(5) print('Server running on port 8080') while True: client, addr = s.accept() print(f'Connection from {addr}') data = client.recv(1024) client.sendall(b'HTTP/1.0 200 OK\r\n\r\nHello') client.close()
UDP编程:UDP是无连接的协议,不需要建立连接直接发送数据。import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.sendto(b'Hello', ('127.0.0.1', 9999)) data, addr = s.recvfrom(1024) print(f'Received from {addr}: {data}') s.close() UDP适合对实时性要求高但不要求可靠传输的场景,如视频流、DNS查询等。
HTTP请求urllib:Python标准库的urllib提供高层HTTP客户端功能。import urllib.request, urllib.parse url = 'https://api.example.com/data' data = {'name': 'test', 'value': 123} params = urllib.parse.urlencode(data) with urllib.request.urlopen(f'{url}?{params}') as response: print(response.read().decode()) POST请求:req = urllib.request.Request(url, data=params.encode(), headers={'Content-Type': 'application/x-www-form-urlencoded'}) with urllib.request.urlopen(req) as response: print(response.read().decode())
requests库高级HTTP:第三方库requests比urllib更易用。import requests r = requests.get('https://api.github.com/repos/python/cpython') print(r.status_code) print(r.json()['description']) POST请求:payload = {'username': 'admin', 'password': '123456'} r = requests.post('https://httpbin.org/post', data=payload) print(r.json()) 设置超时:r = requests.get('https://example.com', timeout=5)
RESTful API调用实战:实际项目中经常需要调用第三方REST API。import requests, json class APIClient: def __init__(self, base_url, token): self.base_url = base_url self.headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'} def get(self, path): r = requests.get(f'{self.base_url}{path}', headers=self.headers) return r.json() def post(self, path, data): r = requests.post(f'{self.base_url}{path}', headers=self.headers, json=data) return r.json() client = APIClient('https://api.example.com', 'your_token') result = client.get('/users/me') print(result)
异步HTTP请求:同步HTTP在大量请求时效率很低,aiohttp支持异步并发请求。import asyncio, aiohttp async def fetch(session, url): async with session.get(url) as response: return await response.text() async def main(): async with aiohttp.ClientSession() as session: tasks = [fetch(session, f'https://httpbin.org/delay/1') for _ in range(10)] results = await asyncio.gather(*tasks) print(f'Got {len(results)} responses') asyncio.run(main())
网络错误处理:网络操作容易失败,务必做好异常处理。import requests, socket for attempt in range(3): try: r = requests.get('https://example.com', timeout=5) if r.status_code == 200: print('Success') break except requests.exceptions.Timeout: print(f'Attempt {attempt+1} timeout, retrying...') except requests.exceptions.ConnectionError: print(f'Connection error: {e}') break except Exception as e: print(f'Unexpected error: {e}') break
以上就是Python网络编程的核心知识点,建议配合实际项目练习掌握。