1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 客户端发起握手（ Upgrade 头部表示要升级协议）
"""
GET /chat HTTP/1.1
Host: example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
"""

# 服务器响应
"""
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import socket

# 创建一个 TCP socket
# socket.AF_INET 表示 IPv4，socket.SOCK_STREAM 表示 TCP
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 绑定到地址和端口
# '' 表示监听所有可用网络接口，8080 是端口号
server_socket.bind(('', 8080))

# 开始监听，参数表示最大排队连接数
server_socket.listen(5)

print("🖥️  TCP 服务器启动，监听 8080 端口...")

while True:
    # 接受客户端连接（阻塞等待）
    # 返回一个新的 socket（用于和该客户端通信）和客户端地址
    client_socket, client_addr = server_socket.accept()
    print(f"📞 有客户端连接：{client_addr}")
    
    # 接收客户端数据（最多 1024 字节）
    data = client_socket.recv(1024)
    print(f"📩 收到数据：{data.decode('utf-8')}")
    
    # 发送响应
    response = "你好，客户端！我是服务器！👋"
    client_socket.send(response.encode('utf-8'))
    
    # 关闭与客户端的连接
    client_socket.close()
    print(f"🔌 已关闭与 {client_addr} 的连接")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import socket

# 创建 socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 连接到服务器（注意：这里客户端也需要指定地址和端口吗？
# 不需要！客户端的端口是系统自动分配的"临时工位"）
server_addr = ('127.0.0.1', 8080)  # 127.0.0.1 是本机地址（localhost）
client_socket.connect(server_addr)

print("🔗 已连接到服务器")

# 发送数据
message = "服务器你好，我是客户端！"
client_socket.send(message.encode('utf-8'))
print(f"📤 已发送：{message}")

# 接收响应
response = client_socket.recv(1024)
print(f"📥 收到响应：{response.decode('utf-8')}")

# 关闭连接
client_socket.close()
print("🔌 连接已关闭")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import socket

# 创建 UDP socket
# SOCK_DGRAM 表示数据报模式（UDP）
udp_server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 绑定地址
udp_server.bind(('', 8080))

print("📡 UDP 服务器启动，监听 8080 端口...")

while True:
    # recvfrom 返回数据和数据来源的地址
    data, client_addr = udp_server.recvfrom(1024)
    print(f"📩 收到来自 {client_addr} 的数据：{data.decode('utf-8')}")
    
    # 发送响应（不需要先 accept）
    response = f"收到！你的消息是：{data.decode('utf-8')}"
    udp_server.sendto(response.encode('utf-8'), client_addr)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import socket

# 创建 UDP socket
udp_client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

server_addr = ('127.0.0.1', 8080)

# 直接发送数据，不需要 connect！
message = "UDP 测试消息"
udp_client.sendto(message.encode('utf-8'), server_addr)

# 接收响应
response, server_addr = udp_client.recvfrom(1024)
print(f"📥 收到响应：{response.decode('utf-8')}")

udp_client.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

"""
UDP 多人聊天室
原理：所有用户加入同一个广播组，发的消息大家都能看到
"""

import socket
import threading

# 配置
BROADCAST_IP = '192.168.255.255'  # 局域网广播地址（需要根据实际网络修改）
PORT = 5000
BUFFER_SIZE = 1024

class ChatRoom:
    def __init__(self, username):
        self.username = username
        # 创建 UDP socket
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # 设置允许广播
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(('', PORT))
        self.running = True
    
    def send_message(self, message):
        """发送消息到聊天室"""
        full_message = f"[{self.username}]: {message}"
        self.sock.sendto(full_message.encode('utf-8'), (BROADCAST_IP, PORT))
    
    def receive_messages(self):
        """接收聊天室消息（在独立线程中运行）"""
        while self.running:
            try:
                data, addr = self.sock.recvfrom(BUFFER_SIZE)
                message = data.decode('utf-8')
                # 打印消息（排除自己发的，避免刷屏）
                if not message.startswith(f"[{self.username}]"):
                    print(f"\r📢 {message}\n[{self.username}]: ", end='')
            except Exception:
                pass
    
    def start(self):
        """启动聊天"""
        # 启动接收线程
        receive_thread = threading.Thread(target=self.receive_messages)
        receive_thread.daemon = True
        receive_thread.start()
        
        print(f"🚀 进入聊天室（广播地址: {BROADCAST_IP}:{PORT}）")
        print("开始聊天吧！输入消息后按回车发送，输入 'quit' 退出\n")
        
        try:
            while True:
                message = input(f"[{self.username}]: ")
                if message.lower() == 'quit':
                    self.send_message("👋 离开了聊天室")
                    break
                self.send_message(message)
        except KeyboardInterrupt:
            print("\n⚠️ 被中断，退出聊天室")
        finally:
            self.running = False
            self.sock.close()
            print("🔌 已断开连接")


# 使用示例
if __name__ == '__main__':
    username = input("请输入你的昵称：")
    chat_room = ChatRoom(username)
    chat_room.start()

1
2

# 安装 requests
pip install requests

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import requests

# 最简单的 GET 请求
response = requests.get('https://api.github.com/users/octocat')
print(response.status_code)  # 状态码：200 表示成功
print(response.json())  # 解析 JSON 响应

# 带参数的 GET 请求
params = {
    'page': 1,
    'per_page': 10,
    'q': 'python'
}
response = requests.get('https://api.github.com/search/repositories', params=params)
data = response.json()
print(f"找到 {data['total_count']} 个仓库")

# 设置请求头
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Accept': 'application/json'
}
response = requests.get('https://api.github.com/users/octocat', headers=headers)
print(response.headers['Content-Type'])  # application/json; charset=utf-8

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import requests

# POST 请求（提交表单数据）
data = {
    'username': 'test_user',
    'password': '123456'
}
response = requests.post('https://httpbin.org/post', data=data)
print(response.json())

# POST JSON 数据
json_data = {
    'title': '我的博客文章',
    'content': '这是一篇关于 Python 网络编程的文章',
    'author': '小明'
}
response = requests.post('https://httpbin.org/post', json=json_data)
print(response.json())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import requests

response = requests.get('https://httpbin.org/json')

# 常用属性
print(response.status_code)   # 状态码
print(response.headers)       # 响应头
print(response.text)          # 响应体（文本）
print(response.content)       # 响应体（字节）
print(response.json())        # JSON 解析

# 判断是否成功
if response.status_code == 200:
    data = response.json()
    print("请求成功！")
elif response.status_code == 404:
    print("资源不存在")
else:
    print(f"出错了，状态码：{response.status_code}")

1

pip install httpx

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import httpx

# 同步 GET 请求（和 requests 几乎一样）
response = httpx.get('https://httpbin.org/json')
print(response.status_code)
print(response.json())

# 带参数和头部
headers = {'Authorization': 'Bearer token123'}
params = {'page': 1, 'limit': 10}
response = httpx.get('https://httpbin.org/get', headers=headers, params=params)
print(response.json())

# POST 请求
response = httpx.post('https://httpbin.org/post', json={'key': 'value'})
print(response.json())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import httpx
import asyncio

async def fetch_data():
    """异步获取多个网页（并发执行，比串行快很多）"""
    
    async with httpx.AsyncClient() as client:
        # 并发请求多个 URL
        urls = [
            'https://httpbin.org/delay/1',  # 故意延迟 1 秒
            'https://httpbin.org/delay/2',  # 故意延迟 2 秒
            'https://httpbin.org/get',
        ]
        
        # 使用 asyncio.gather 并发执行
        # 如果串行执行需要 1+2+0.1=3.1 秒，并发只需要 max(1,2,0.1)=2 秒
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        
        for resp in responses:
            print(f"{resp.url}: {resp.status_code}")

# 运行异步函数
asyncio.run(fetch_data())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import requests

# 最简单的文件下载
url = 'https://www.python.org/static/img/python-logo.png'
response = requests.get(url, stream=True)

with open('python-logo.png', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        f.write(chunk)

print("✅ 下载完成！")

# 带进度显示的下载
def download_with_progress(url, filename):
    """带进度显示的文件下载"""
    response = requests.get(url, stream=True)
    total_size = int(response.headers.get('Content-Length', 0))
    
    downloaded = 0
    with open(filename, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
            downloaded += len(chunk)
            if total_size:
                percent = (downloaded / total_size) * 100
                print(f"\r📥 下载进度：{percent:.1f}%", end='')
    
    print(f"\n✅ {filename} 下载完成！")

# 使用示例
download_with_progress(
    'https://www.python.org/ftp/python/3.11.0/python-3.11.0.exe',
    'python-installer.exe'
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import requests

# 上传文件
with open('my-document.pdf', 'rb') as f:
    files = {'file': ('document.pdf', f, 'application/pdf')}
    response = requests.post('https://httpbin.org/post', files=files)
    print(response.json())

# 上传多个文件
with open('image1.png', 'rb') as f1, open('image2.png', 'rb') as f2:
    files = [
        ('images', ('image1.png', f1, 'image/png')),
        ('images', ('image2.png', f2, 'image/png')),
    ]
    response = requests.post('https://httpbin.org/post', files=files)
    print(response.json())

# 上传文件 + 附带表单数据
with open('avatar.jpg', 'rb') as f:
    files = {'avatar': ('my-avatar.jpg', f, 'image/jpeg')}
    data = {'username': '小明', 'bio': 'Python爱好者'}
    response = requests.post('https://httpbin.org/post', files=files, data=data)
    print(response.json())

1

pip install flask

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

from flask import Flask, jsonify, request

app = Flask(__name__)

# 模拟数据库
users = [
    {'id': 1, 'name': '张三', 'email': 'zhangsan@example.com'},
    {'id': 2, 'name': '李四', 'email': 'lisi@example.com'},
]
next_id = 3

@app.route('/')
def index():
    """首页"""
    return jsonify({
        'message': '欢迎使用用户管理 API',
        'endpoints': ['/users', '/users/<id>']
    })

@app.route('/users', methods=['GET'])
def get_users():
    """获取所有用户"""
    return jsonify({'users': users})

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    """获取单个用户"""
    user = next((u for u in users if u['id'] == user_id), None)
    if user:
        return jsonify(user)
    return jsonify({'error': '用户不存在'}), 404

@app.route('/users', methods=['POST'])
def create_user():
    """创建用户"""
    global next_id
    data = request.json
    
    new_user = {
        'id': next_id,
        'name': data.get('name'),
        'email': data.get('email')
    }
    users.append(new_user)
    next_id += 1
    
    return jsonify(new_user), 201

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    """更新用户"""
    user = next((u for u in users if u['id'] == user_id), None)
    if not user:
        return jsonify({'error': '用户不存在'}), 404
    
    data = request.json
    user['name'] = data.get('name', user['name'])
    user['email'] = data.get('email', user['email'])
    
    return jsonify(user)

@app.route('/users/<int:user_id>', methods=['PATCH'])
def patch_user(user_id):
    """部分更新用户（仅更新提供的字段）"""
    user = next((u for u in users if u['id'] == user_id), None)
    if not user:
        return jsonify({'error': '用户不存在'}), 404
    
    data = request.json
    if 'name' in data:
        user['name'] = data['name']
    if 'email' in data:
        user['email'] = data['email']
    
    return jsonify(user)

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    """删除用户"""
    global users
    user = next((u for u in users if u['id'] == user_id), None)
    if not user:
        return jsonify({'error': '用户不存在'}), 404
    
    users = [u for u in users if u['id'] != user_id]
    return '', 204

if __name__ == '__main__':
    print("🚀 Flask API 服务器启动：http://127.0.0.1:5000")
    app.run(debug=True, port=5000)

1

pip install fastapi uvicorn

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uvicorn

app = FastAPI(title="用户管理 API", version="1.0.0")

# Pydantic 模型（自动验证数据）
class User(BaseModel):
    name: str
    email: str

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None

# 模拟数据库
users_db = {
    1: {'id': 1, 'name': '张三', 'email': 'zhangsan@example.com'},
    2: {'id': 2, 'name': '李四', 'email': 'lisi@example.com'},
}
next_id = 3

@app.get('/')
def index():
    """首页"""
    return {'message': '欢迎使用 FastAPI 用户管理 API'}

@app.get('/users')
def get_users():
    """获取所有用户"""
    return {'users': list(users_db.values())}

@app.get('/users/{user_id}')
def get_user(user_id: int):
    """获取单个用户"""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail='用户不存在')
    return users_db[user_id]

@app.post('/users', status_code=201)
def create_user(user: User):
    """创建用户"""
    global next_id
    new_user = {'id': next_id, **user.model_dump()}
    users_db[next_id] = new_user
    next_id += 1
    return new_user

@app.put('/users/{user_id}')
def update_user(user_id: int, user: UserUpdate):
    """更新用户"""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail='用户不存在')
    
    db_user = users_db[user_id]
    update_data = user.model_dump(exclude_unset=True)
    db_user.update(update_data)
    return db_user

@app.delete('/users/{user_id}')
def delete_user(user_id: int):
    """删除用户"""
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail='用户不存在')
    del users_db[user_id]
    return {'message': '用户已删除'}

if __name__ == '__main__':
    print("🚀 FastAPI 服务器启动：http://127.0.0.1:8000")
    print("📖 API 文档：http://127.0.0.1:8000/docs")
    uvicorn.run(app, host='127.0.0.1', port=8000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import jwt
import datetime

# JWT 配置
SECRET_KEY = '你的超级密钥，别让人知道！'  # 生产环境要放环境变量
ALGORITHM = 'HS256'  # HMAC SHA-256

def create_token(user_id: int, username: str) -> str:
    """生成 JWT token"""
    payload = {
        'user_id': user_id,
        'username': username,
        'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=24),  # 24 小时过期
        'iat': datetime.datetime.utcnow()  # 签发时间
    }
    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return token

def verify_token(token: str) -> dict:
    """验证 JWT token"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise Exception('Token 已过期')
    except jwt.InvalidTokenError:
        raise Exception('无效的 Token')

# 使用示例
token = create_token(user_id=1, username='小明')
print(f"生成的 Token：{token}")

try:
    payload = verify_token(token)
    print(f"Token 验证成功！用户信息：{payload}")
except Exception as e:
    print(f"Token 验证失败：{e}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

from flask import Flask, request, jsonify
import jwt
import datetime

app = Flask(__name__)
SECRET_KEY = 'secret_key_123'
ALGORITHM = 'HS256'

# 模拟用户数据库（实际项目中要连接真实数据库）
users = {
    'admin': {'password': 'password123', 'role': 'admin'},
    'user': {'password': 'user456', 'role': 'user'}
}

@app.route('/login', methods=['POST'])
def login():
    """登录接口"""
    data = request.json
    username = data.get('username')
    password = data.get('password')
    
    # 验证用户
    if username not in users or users[username]['password'] != password:
        return jsonify({'error': '用户名或密码错误'}), 401
    
    # 生成 token
    token = jwt.encode({
        'username': username,
        'role': users[username]['role'],
        'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=24)
    }, SECRET_KEY, algorithm=ALGORITHM)
    
    return jsonify({'token': token})

@app.route('/protected', methods=['GET'])
def protected():
    """受保护的接口，需要携带有效 token"""
    auth_header = request.headers.get('Authorization')
    
    if not auth_header or not auth_header.startswith('Bearer '):
        return jsonify({'error': '未提供认证信息'}), 401
    
    token = auth_header.split(' ')[1]
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return jsonify({
            'message': f'欢迎，{payload["username"]}！',
            'role': payload['role']
        })
    except jwt.ExpiredSignatureError:
        return jsonify({'error': 'Token 已过期'}), 401
    except jwt.InvalidTokenError:
        return jsonify({'error': '无效的 Token'}), 401

if __name__ == '__main__':
    app.run(debug=True, port=5000)