Python实现常见网络通信的示例详解<

import requests

# HTTP GET
response = requests.get('http://httpbin.org/get')
print(response.text)

# HTTPS POST
response = requests.post(
    'https://httpbin.org/post',
    data={'key': 'value'},
    verify=True  # 验证 SSL 证书（默认）
)
print(response.json())

from flask import Flask, request

app = Flask(__name__)

@app.route('/api', methods=['GET', 'POST'])
def handle_request():
    if request.method == 'GET':
        return {'message': 'GET received'}
    elif request.method == 'POST':
        return {'data': request.json}

if __name__ == '__main__':
    app.run(ssl_context='adhoc')  # 启用 HTTPS

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server.bind(('0.0.0.0', 9999))

while True:
    data, addr = server.recvfrom(1024)
    print(f"Received from {addr}: {data.decode()}")
    server.sendto(b'UDP response', addr)

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client.sendto(b'Hello UDP', ('localhost', 9999))
response, addr = client.recvfrom(1024)
print(f"Received: {response.decode()}")

import asyncio
import websockets

async def handler(websocket):
    async for message in websocket:
        print(f"Received: {message}")
        await websocket.send(f"Echo: {message}")

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        await asyncio.Future()  # 永久运行

asyncio.run(main())

import asyncio
import websockets

async def client():
    async with websockets.connect("ws://localhost:8765") as ws:
        await ws.send("Hello WebSocket!")
        response = await ws.recv()
        print(f"Received: {response}")

asyncio.run(client())

from flask import Flask, Response

app = Flask(__name__)

@app.route('/stream')
def stream():
    def event_stream():
        for i in range(5):
            yield f"data: Message {i}\n\n"
    return Response(event_stream(), mimetype="text/event-stream")

if __name__ == '__main__':
    app.run()

import requests
from sseclient import SSEClient

url = 'http://localhost:5000/stream'
response = requests.get(url, stream=True)
client = SSEClient(response)

for event in client.events():
    print(f"Received event: {event.data}")