当前位置：网站首页 > 技术资源 > 正文

Python教程(三十):网络编程基础

off999 2025-08-06 22:30 43 浏览 0 评论

今日目标

o 理解网络编程的基本概念和协议

o 掌握Socket编程的基础知识

o 学会创建HTTP客户端和服务器

o 了解网络编程的安全考虑

网络编程概述

网络编程是计算机之间进行通信的基础，Python提供了丰富的网络编程库：

o socket：底层网络通信

o http.client：HTTP客户端

o urllib：URL处理

o requests：高级HTTP库

o asyncio：异步网络编程

网络协议层次

应用层    HTTP, FTP, SMTP
传输层    TCP, UDP
网络层    IP
链路层    以太网, WiFi
物理层    电缆, 光纤

Socket编程基础

1. 创建Socket

import socket

# 创建TCP Socket
tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

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

# Socket类型说明
# AF_INET: IPv4地址族
# AF_INET6: IPv6地址族
# SOCK_STREAM: TCP协议
# SOCK_DGRAM: UDP协议

2. TCP客户端示例

import socket
import time

def tcp_client():
    """TCP客户端示例"""
    # 创建Socket
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        # 连接服务器
        server_address = ('localhost', 8888)
        print(f"连接到服务器: {server_address}")
        client.connect(server_address)
        
        # 发送数据
        message = "Hello, Server!"
        client.send(message.encode('utf-8'))
        print(f"发送: {message}")
        
        # 接收响应
        data = client.recv(1024)
        response = data.decode('utf-8')
        print(f"收到: {response}")
        
    except ConnectionRefusedError:
        print("连接被拒绝，请确保服务器正在运行")
    except Exception as e:
        print(f"连接错误: {e}")
    finally:
        client.close()
        print("连接已关闭")

# 运行客户端
if __name__ == "__main__":
    tcp_client()

3. TCP服务器示例

import socket
import threading
import time

class TCPServer:
    def __init__(self, host='localhost', port=8888):
        self.host = host
        self.port = port
        self.server_socket = None
        self.clients = []
    
    def start(self):
        """启动服务器"""
        try:
            # 创建服务器Socket
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            
            # 设置端口重用
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            
            # 绑定地址
            self.server_socket.bind((self.host, self.port))
            
            # 开始监听
            self.server_socket.listen(5)
            print(f"服务器启动成功，监听地址: {self.host}:{self.port}")
            
            # 接受客户端连接
            while True:
                client_socket, client_address = self.server_socket.accept()
                print(f"新客户端连接: {client_address}")
                
                # 为每个客户端创建新线程
                client_thread = threading.Thread(
                    target=self.handle_client,
                    args=(client_socket, client_address)
                )
                client_thread.start()
                self.clients.append(client_socket)
                
        except Exception as e:
            print(f"服务器错误: {e}")
        finally:
            self.stop()
    
    def handle_client(self, client_socket, client_address):
        """处理客户端连接"""
        try:
            while True:
                # 接收数据
                data = client_socket.recv(1024)
                if not data:
                    break
                
                # 解码数据
                message = data.decode('utf-8')
                print(f"来自 {client_address}: {message}")
                
                # 发送响应
                response = f"服务器收到: {message}"
                client_socket.send(response.encode('utf-8'))
                
        except Exception as e:
            print(f"处理客户端 {client_address} 时出错: {e}")
        finally:
            client_socket.close()
            if client_socket in self.clients:
                self.clients.remove(client_socket)
            print(f"客户端 {client_address} 断开连接")
    
    def stop(self):
        """停止服务器"""
        if self.server_socket:
            self.server_socket.close()
        for client in self.clients:
            client.close()
        print("服务器已停止")

# 启动服务器
if __name__ == "__main__":
    server = TCPServer()
    try:
        server.start()
    except KeyboardInterrupt:
        print("\n正在关闭服务器...")
        server.stop()

HTTP编程

1. 使用urllib

import urllib.request
import urllib.parse
import urllib.error
import json

def http_get_example():
    """HTTP GET请求示例"""
    try:
        # 发送GET请求
        url = "https://jsonplaceholder.typicode.com/posts/1"
        with urllib.request.urlopen(url) as response:
            data = response.read()
            result = json.loads(data.decode('utf-8'))
            print("GET请求结果:")
            print(f"标题: {result['title']}")
            print(f"内容: {result['body']}")
            
    except urllib.error.URLError as e:
        print(f"URL错误: {e}")
    except Exception as e:
        print(f"请求错误: {e}")

def http_post_example():
    """HTTP POST请求示例"""
    try:
        # 准备数据
        post_data = {
            "title": "Python网络编程",
            "body": "学习Python网络编程很有趣",
            "userId": 1
        }
        
        # 编码数据
        data = json.dumps(post_data).encode('utf-8')
        
        # 创建请求
        url = "https://jsonplaceholder.typicode.com/posts"
        req = urllib.request.Request(
            url,
            data=data,
            headers={'Content-Type': 'application/json'}
        )
        
        # 发送请求
        with urllib.request.urlopen(req) as response:
            result = json.loads(response.read().decode('utf-8'))
            print("POST请求结果:")
            print(f"新帖子ID: {result['id']}")
            print(f"标题: {result['title']}")
            
    except urllib.error.URLError as e:
        print(f"URL错误: {e}")
    except Exception as e:
        print(f"请求错误: {e}")

# 运行示例
http_get_example()
print("-" * 50)
http_post_example()

2. 使用requests库

import requests
import json

class HTTPClient:
    def __init__(self, base_url="https://jsonplaceholder.typicode.com"):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Python-HTTPClient/1.0',
            'Content-Type': 'application/json'
        })
    
    def get(self, endpoint, params=None):
        """发送GET请求"""
        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.get(url, params=params, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"GET请求错误: {e}")
            return None
    
    def post(self, endpoint, data=None):
        """发送POST请求"""
        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.post(url, json=data, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"POST请求错误: {e}")
            return None
    
    def put(self, endpoint, data=None):
        """发送PUT请求"""
        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.put(url, json=data, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"PUT请求错误: {e}")
            return None
    
    def delete(self, endpoint):
        """发送DELETE请求"""
        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.delete(url, timeout=10)
            response.raise_for_status()
            return response.status_code == 204
        except requests.RequestException as e:
            print(f"DELETE请求错误: {e}")
            return False

# 使用示例
client = HTTPClient()

# 获取所有帖子
posts = client.get("/posts")
if posts:
    print(f"获取到 {len(posts)} 个帖子")
    for post in posts[:3]:
        print(f"- {post['title']}")

# 创建新帖子
new_post = {
    "title": "Python网络编程教程",
    "body": "这是一个关于Python网络编程的详细教程",
    "userId": 1
}

created_post = client.post("/posts", new_post)
if created_post:
    print(f"新帖子创建成功，ID: {created_post['id']}")

# 更新帖子
update_data = {"title": "更新的标题"}
updated_post = client.put(f"/posts/{created_post['id']}", update_data)
if updated_post:
    print(f"帖子更新成功: {updated_post['title']}")

# 删除帖子
if client.delete(f"/posts/{created_post['id']}"):
    print("帖子删除成功")

网络编程安全

1. 输入验证

import re
import socket

def validate_ip_address(ip):
    """验证IP地址格式"""
    pattern = r'^(\d{1,3}\.){3}\d{1,3}#39;
    if not re.match(pattern, ip):
        return False
    
    parts = ip.split('.')
    for part in parts:
        if not 0 <= int(part) <= 255:
            return False
    return True

def validate_port(port):
    """验证端口号"""
    try:
        port_num = int(port)
        return 1 <= port_num <= 65535
    except ValueError:
        return False

def safe_socket_connect(host, port):
    """安全的Socket连接"""
    if not validate_ip_address(host):
        raise ValueError(f"无效的IP地址: {host}")
    
    if not validate_port(port):
        raise ValueError(f"无效的端口号: {port}")
    
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(10)  # 设置超时
        sock.connect((host, port))
        return sock
    except socket.timeout:
        raise ConnectionError("连接超时")
    except socket.error as e:
        raise ConnectionError(f"连接失败: {e}")

2. 数据加密

import ssl
import socket

def create_ssl_client(host, port):
    """创建SSL客户端"""
    try:
        # 创建SSL上下文
        context = ssl.create_default_context()
        
        # 创建Socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
        # 包装为SSL Socket
        ssl_sock = context.wrap_socket(sock, server_hostname=host)
        
        # 连接
        ssl_sock.connect((host, port))
        return ssl_sock
        
    except ssl.SSLError as e:
        print(f"SSL错误: {e}")
        return None
    except Exception as e:
        print(f"连接错误: {e}")
        return None

def create_ssl_server(host, port, cert_file, key_file):
    """创建SSL服务器"""
    try:
        # 创建SSL上下文
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(cert_file, key_file)
        
        # 创建Socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind((host, port))
        sock.listen(5)
        
        # 包装为SSL Socket
        ssl_sock = context.wrap_socket(sock, server_side=True)
        return ssl_sock
        
    except Exception as e:
        print(f"SSL服务器创建失败: {e}")
        return None