第十一章 网络与web编程
1 作为客户端与http服务交互
示例代码如下:
服务端使用django,部分代码如下:
# 处理get请求 if request.method == 'GET': # TODO 测试get请求参数 print('receive the get request') print('==' * 20) print("param1", request.GET.get('name1')) print("param2", request.GET.get('name2')) print('==' * 20)# 处理post请求 if request.method == 'POST': # TODO 测试post请求参数 print('receive the post request...') param1 = request.POST.get('name1', None) param2 = request.POST.get('name2', None) print('==' * 20) print('param1', param1) print('param2', param2)
服务端输出结果如下:
receive the get request
========================================
param1 value1
param2 value2
========================================
客记端代码如下:
from urllib import request, parse # 使用本地工程进行测试 url = 'http://localhost:8000/jkx/index/' parms = {'name1': 'value1', 'name2': 'value2' } queryString = parse.urlencode(parms) # 发送get请求 #u = request.urlopen(url + '?' + queryString) # 发送post请求 u2 = request.urlopen(url, queryString.encode('ascii')) #resp = u.read() #print(resp) resp2 = u2.read() print(resp2) #服务端返回内容 # b'<!DOCTYPE html>\n<html>\n<head>\n <meta charset="utf-8">\n ...
2 创建TCP服务器
服务器端
from socketserver import BaseRequestHandler, \ TCPServer, \ StreamRequestHandler, \ ThreadingTCPServer, \ ForkingTCPServer import socket from threading import Thread # 方式一 class EchoHandler(BaseRequestHandler): def handle(self): print('Got connection from', self.client_address) while True: msg = self.request.recv(8192) if not msg: break print('msg:', msg) self.request.send(b'hello') # 方式二 class EchoHandler2(StreamRequestHandler): timeout = 5 rbufsize = -1 wbufsize = 0 disable_nagle_algorithm = False def handle(self): print('Got connection from', self.client_address) try: for line in self.rfile: self.wfile.write(line) except socket.timeout: print('Timed out!') def thread_pool(): NWROKERS = 16 serv = TCPServer(('', 20000), EchoHandler) for n in range(NWROKERS): t = Thread(target=serv.serve_forever()) t.daemon = True t.start() if __name__ == '__main__': # 单线程的tcp服务器 # 可以在TCPServer初始化前设置(允许服务器重新绑定一个之前使用过的端口号) # TCPServer.allow_reuse_address = True # bind_and_activate可以在初始化完成后,调整socket serv = TCPServer(('', 20000), EchoHandler, bind_and_activate=False) # 允许服务器重新绑定一个之前使用过的端口号 serv.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) serv.server_bind() serv.server_activate() # 处理多个客户端请求, # 1 可以合用 # serv = ThreadingTCPServer(('', 20000), EchoHandler) # 2 为每个客端连接创建一个新的进程或线程 # serv = ForkingTCPServer(('', 20000), EchoHandler) # 3 也可以使用线程池的方式 # thread_pool() serv.serve_forever()
客户端
from socket import socket, AF_INET, SOCK_STREAM # 创建套接字 s = socket(AF_INET, SOCK_STREAM) # 连接服务器 s.connect(('localhost', # 服务器ip 20000)) # 端口号 # 发送消息 print(s.send(b'hi retacn')) # 返回信息 print(s.recv(8192))
使用socket实现的简单服务器
from socket import socket, AF_INET, SOCK_STREAM def echo_handler(address, client_sock): print('Go connection from {}'.format(address)) while True: msg = client_sock.recv(8192) if not msg: break client_sock.sendall(msg) client_sock.close() def echo_server(address, backlog=5): sock = socket(AF_INET, SOCK_STREAM) sock.bind(address) sock.listen(backlog) while True: client_sock, client_addr = sock.accept() echo_handler(client_addr, client_sock) if __name__ == '__main__': echo_server('', 20000)
3 创建UDP服务器
服务器端
from socketserver import BaseRequestHandler, UDPServer, ThreadingUDPServer import time class TimeHandler(BaseRequestHandler): def handle(self): print('Got connection from', self.client_address) msg, sock = self.request resp = time.ctime() # sock.sendto(resp.encode('ascii'), self.client_address) print('msg', msg) sock.sendto(b'hello', self.client_address) # udp无连接,消息可能丢失,应用场景流媒体/游戏 if __name__ == '__main__': serv = UDPServer(('', 20000), TimeHandler) # 并发操作可以使用 # serv = ThreadingUDPServer(('', 20000), TimeHandler) serv.serve_forever()
客户端
from socket import socket, AF_INET, SOCK_DGRAM s = socket(AF_INET, SOCK_DGRAM) print(s.sendto(b'hi retacn', ('localhost', 20000))) print(s.recvfrom(8192))
使用socket实现的udp服务器
from socket import socket, AF_INET, SOCK_DGRAM import time def time_server(address): sock = socket(AF_INET, SOCK_DGRAM) sock.bind(address) while True: msg, addr = sock.recvfrom(8192) print('Got message from ', addr) resp = time.ctime() sock.sendto(resp.encode('asscii'), addr) if __name__ == '__main__': time_server(('', 20000))
4 通过CIDR地址生成对应的ip地址
import ipaddress def ipv4(): net = ipaddress.ip_network('123.45.67.64/27') print('net', net) # 可以按索引取值 print('the first ip: ', net[0], type(net)) # 成员检查 a = ipaddress.ip_network('123.45.67.69') print(a in net) for a in net: print(a) def ipv6(): net6 = ipaddress.ip_network('12:3456:78:90ab:cd:ef01:23:30/125') print('net6', net6) for a in net6: print(a) if __name__ == '__main__': ipv4() # 运行结果为: # net 123.45.67.64/27 # 123.45.67.64 # ...` # 123.45.67.95 # ipv6() # 运行结查为: # net6 12:3456:78:90ab:cd:ef01:23:30 / 125 # 12:3456:78:90ab:cd:ef01:23:30 # 12:3456:78:90ab:cd:ef01:23:31 # 12:3456:78:90ab:cd:ef01:23:32 # 12:3456:78:90ab:cd:ef01:23:33 # 12:3456:78:90ab:cd:ef01:23:34 # 12:3456:78:90ab:cd:ef01:23:35 # 12:3456:78:90ab:cd:ef01:23:36 # 12:3456:78:90ab:cd:ef01:23:37
5 创建一个简单的REST接口
调度器
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2016/4/17 12:41 # @Author : Retacn # @Site : 简单的rest接口(一个简单的调度器), # 不提供一些高级特性(认证,cookies,重定向等), # 实现更多支持,可以使用webob和paste等第三方库 # @File : rest_demo.py # @Software: PyCharm __author__ = "retacn" __copyright__ = "property of mankind." __license__ = "CN" __version__ = "0.0.1" __maintainer__ = "retacn" __email__ = "zhenhuayue@sina.com" __status__ = "Development" import cgi def notfound_404(environ, start_response): start_response('404 Not Found', [('Content-type', 'text/plain')]) return [b'Not found'] # 实现路径调度,映射到处理器的函数上 class PathDispatcher: def __init__(self): self.pathmap = {} def __call__(self, environ, # 包含web服务器提供的CGI接口中取得的值 start_response): # 初始化一个请求对象而必须调用的函数 # 请求资源的路径 path = environ['PATH_INFO'] # 查询参数 params = cgi.FieldStorage(environ['wsgi.input'], environ=environ) # 请求方法 method = environ['REQUEST_METHOD'].lower() environ['params'] = {key: params.getvalue(key) for key in params} handler = self.pathmap.get((method, path), notfound_404) return handler(environ, start_response) def register(self, method, path, function): self.pathmap[method.lower(), path] = function return function
处理器
import time _hello_resp = '''\ <html> <head> <title>Hello {name}</title> </head> <body> <h1>Hello {name}!</h1> </body> </html>''' # 一个WSGI程序必须要返回一个字符串序列 def hello_world(environ, # 包含web服务器提供的CGI接口中取得的值 start_response): # 初始化一个请求对象而必须调用的函数 start_response('200 OK', # http状态值 [('Content-type', 'text/html')]) # http请求头 params = environ['params'] resp = _hello_resp.format(name=params.get('name')) yield resp.encode('utf-8') # 编码后返回 _localtime_resp = '''\ <?xml version="1.0"?> <time> <year>{t.tm_year}</year> <month>{t.tm_mon}</month> <day>{t.tm_mday}</day> <hour>{t.tm_hour}</hour> <minute>{t.tm_min}</minute> <second>{t.tm_sec}</second> </time>''' def localtime(environ, start_response): start_response('200 OK', [('Content-type', 'application/xml')]) resp = _localtime_resp.format(t=time.localtime()) yield resp.encode('utf-8') if __name__ == '__main__': from Eleven.rest.resty import PathDispatcher from wsgiref.simple_server import make_server dispatcher = PathDispatcher() dispatcher.register('GET', '/hello', hello_world) dispatcher.register('GET', '/localtime', localtime) httpd = make_server('', 8080, dispatcher) print('Serving on port 8080...') httpd.serve_forever() # 可以使用浏览器测试,也可以编写代码进行测试
测试程序
from urllib import request u = request.urlopen('http://localhost:8080/hello?name=Retacn') print(u.read().decode('utf-8')) # 运行结果如下: # <html> # <head> # <title>Hello Retacn</title> # </head> # <body> # <h1>Hello Retacn!</h1> # </body> # </html> u = request.urlopen('http://localhost:8080/localtime') print(u.read().decode('utf-8')) # 运行结果如下: # <?xml version="1.0"?> # <time> # <year>2016</year> # <month>4</month> # <day>17</day> # <hour>13</hour> # <minute>28</minute> # <second>39</second> # </time>
6 通过xml-rpc实现简单的远程序用
服务端程序
from xmlrpc.server import SimpleXMLRPCServer class KeyValueServer: _rpc_methods_ = ['get', 'set', 'delete', 'exists', 'keys'] def __init__(self, address): self._data = {} # 实例化rpc服务,单线程的 self._serv = SimpleXMLRPCServer(address, allow_none=True) # 注册rpc方法 for name in self._rpc_methods_: self._serv.register_function(getattr(self, name)) def get(self, name): return self._data[name] def set(self, name, value): self._data[name] = value def delete(self, name): del self._data[name] def exists(self, name): return name in self._data def keys(self): return list(self._data) def serve_forever(self): self._serv.serve_forever() if __name__ == '__main__': kvserv = KeyValueServer(('', 15000)) kvserv.serve_forever()
客户端调用
from xmlrpc.client import ServerProxy s = ServerProxy('http://localhost:15000', allow_none=True) s.set('foo', 'bar') s.set('spam', [1, 2, 3]) print(s.keys()) print(s.get('foo')) print(s.get('spam')) print(s.delete('spam')) print(s.exists('spam')) #运行结果如下: # ['foo', 'spam'] # bar # [1, 2, 3] # None # False
7 在不同的python解释器之间交互
from multiprocessing.connection import Listener import traceback def echo_client(conn): try: while True: msg = conn.recv() conn.send(msg) except EOFError: print('Connection closed') def echo_server(address, authkey): serv = Listener(address, authkey=authkey) while True: try: client = serv.accept() echo_client(client) except Exception: traceback.print_exc() if __name__ == "__main__": echo_server(('', 25000), authkey=b'retacn')
from multiprocessing.connection import Client c = Client(('localhost', 25000), authkey=b'retacn') c.send('hello') print(c.recv()) c.send(42) print(c.recv()) c.send([1, 2, 3, 4, 5, 6]) print(c.recv()) #运行结果如下: # hello # 42 # [1, 2, 3, 4, 5, 6]
8 实现远程方法调用
RPC处理器
import pickle import json class RPCHandler: def __init__(self): self._functions = {} # 初始化远程方法列表 # 注册远程调用方法 def register_function(self, func): self._functions[func.__name__] = func def handle_connection(self, connection): try: while True: # 使用json格式的序列化消息 # func_name,args,kwargs=json.loads(connection.recv()) func_name, args, kwargs = pickle.loads(connection.recv()) try: r = self._functions[func_name](*args, **kwargs) # connection.send(json.dumps(r)) connection.send(pickle.dumps(r)) except Exception as e: # connection.send(json.dumps(e)) connection.send(pickle.dumps(e)) except EOFError: pass
消息服务器
from multiprocessing.connection import Listener from threading import Thread from Eleven.msg_rpc.msg_rpc_server import RPCHandler def rpc_server(handler, address, authkey): sock = Listener(address, authkey=authkey) while True: client = sock.accept() t = Thread(target=handler.handle_connection, args=(client,)) t.daemon = True t.start() # 定义远程调用方法 def add(x, y): return x + y def sub(x, y): return x - y # 注册方法 handler = RPCHandler() handler.register_function(add) handler.register_function(sub) # 启动消息服务器 rpc_server(handler, ('localhost', 17000), authkey=b'retacn')
客户端用来传送请求的RPC代理类
import pickle import json class RPCProxy: def __init__(self, connection): self._connection = connection def __getattr__(self, item): def do_rpc(*args, **kwargs): # 使用json格式的序列化消息 # self._connection.send(json.dumps((item, args, kwargs))) self._connection.send(pickle.dumps((item, args, kwargs))) # result = json.loads(self._connection.recv()) result = pickle.loads(self._connection.recv()) # TODO 语句含义 if isinstance(result, Exception): raise result return result return do_rpc
消息层远程调用客户端
from multiprocessing.connection import Client from Eleven.msg_rpc.msg_rpc_client_proxy import RPCProxy c = Client(('localhost', 17000), authkey=b'retacn') proxy = RPCProxy(c) print(proxy.add(2, 3)) print(proxy.sub(2, 3)) print(proxy.sub([1, 2], 4)) #运行结果如下 # 5 # -1 # Traceback (most recent call last): # File "D:/workspace_pycharm/python3_cookbook/Eleven/msg_rpc/msg_rpc_client.py", line 25, in <module> # print(proxy.sub([1, 2], 4)) # File "D:\workspace_pycharm\python3_cookbook\Eleven\msg_rpc\msg_rpc_client_proxy.py", line 29, in do_rpc # raise result # TypeError: unsupported operand type(s) for -: 'list' and 'int'
9 简单的客户端认证
认证过程
import hmac import os def client_authenticate(connection, secret_key): message = connection.recv(32) hash = hmac.new(secret_key, message) digest = hash.digest() connection.send(digest) def server_authenticate(connection, secret_key): message = os.urandom(32) connection.send(message) hash = hmac.new(secret_key, message) digest = hash.digest() response = connection.recv(len(digest)) return hmac.compare_digest(digest, response)
服务器端
from socket import socket, AF_INET, SOCK_STREAM from Eleven.auth.auth_hmac import server_authenticate secret_key = b'retacn' def echo_handler(client_sock): if not server_authenticate(client_sock, secret_key): client_sock.close() return while True: msg = client_sock.recv(8192) if not msg: break client_sock.sendall(msg) def echo_server(address): s = socket(AF_INET, SOCK_STREAM) s.bind(address) s.listen(5) while True: c, a = s.accept() echo_handler(c) if __name__ == '__main__': echo_server(('', 18000))
客户端
from socket import socket, AF_INET, SOCK_STREAM from Eleven.auth.auth_hmac import client_authenticate secret_key = b'retacn' s = socket(AF_INET, SOCK_STREAM) s.connect(('localhost', 18000)) client_authenticate(s, secret_key) s.send(b'Hello World') resp = s.recv(1024) print(resp) # 运行结果: # b'Hello World'
10 在网络服务中加入ssl
下载openssl安装略
生成自签名的证书,便于测试
D:\workspace_pycharm\python3_cookbook\Eleven\ssl>opensslreq -new -x509 -days 365 -nodes -out server_cert.pem -keyout server_key.pem
Loading 'screen' into random state - done
Generating a 1024 bit RSA private key
.......................................................................++++++
.............++++++
writing new private key to 'server_key.pem'
-----
You are about to be asked to enterinformation that will be incorporated
into your certificate request.
What you are about to enter is what iscalled a Distinguished Name or a DN.
There are quite a few fields but you canleave some blank
For some fields there will be a defaultvalue,
If you enter '.', the field will be leftblank.
-----
Country Name (2 letter code) [AU]:CN
State or Province Name (full name)[Some-State]:SD
Locality Name (eg, city) []:ZB
Organization Name (eg, company) [InternetWidgits Pty Ltd]:retacn
Organizational Unit Name (eg, section)[]:retacn
#如果使用本机测试,可以填写localhost,否则使用服务器的域名
Common Name (e.g. server FQDN or YOUR name)[]:localhost
Email Address []:zhenhuayue@sina.com
D:\workspace_pycharm\python3_cookbook\Eleven\ssl>ls
server_cert.pem server_key.pem ssl_client.py ssl_server.py
服务器端
from socket import socket, AF_INET, SOCK_STREAM import ssl # 私钥 KEYFILE = 'server_key.pem' # 签名证书 CERTFILE = 'server_cert.pem' def echo_client(s): while True: data = s.recv(8192) if data == b'': break s.send(data) s.close() print('Connection closed') def echo_server(address): s = socket(AF_INET, SOCK_STREAM) s.bind(address) s.listen(1) s_ssl = ssl.wrap_socket(s, keyfile=KEYFILE, certfile=CERTFILE, server_side=True) while True: try: c, a = s_ssl.accept() print('Got connection', c, a) echo_client(c) except Exception as e: print('{}:{}'.format(e.__class__.__name__, e)) if __name__ == '__main__': echo_server(('', 20000)) #运行结果: # Got connection <ssl.SSLSocket fd=444, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 20000), raddr=('127.0.0.1', 6273)> ('127.0.0.1', 6273) # Connection closed
客户端
from socket import socket, AF_INET, SOCK_STREAM import ssl s = socket(AF_INET, SOCK_STREAM) s_ssl = ssl.wrap_socket(s, cert_reqs=ssl.CERT_REQUIRED, ca_certs='server_cert.pem') s_ssl.connect(('localhost', 20000)) print(s_ssl.send(b'Hello World')) print(s_ssl.recv(8192)) #运行结果如下: # 11 # b'Hello World'
以上程序存在的问题是不能与标准库中存在的网络服务兼容,可以通过以下方式实现
未作测试
11 进程间传递socket文件描述符
应用场景:服务器响应连接请求,但实际的响应逻辑在另一个解释器中执行
import multiprocessing from multiprocessing.reduction import recv_handle, send_handle import socket def worker(in_p, out_p): out_p.close() while True: fd = recv_handle(in_p) print('CHILD: GOT FD', fd) with socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=fd) as s: while True: msg = s.recv(1024) if not msg: break print('CHILD:RECV {!r}'.format(msg)) s.send(msg) def server(address, in_p, out_p, worker_pid): in_p.close() s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) s.bind(address) s.listen(1) while True: client, addr = s.accept() print('SERVER:GOT connection from ', addr) send_handle(out_p, client.fileno(), worker_pid) client.close() if __name__ == '__main__': c1, c2 = multiprocessing.Pipe() worker_p = multiprocessing.Process(target=worker, args=(c1, c2)) worker_p.start() server_p = multiprocessing.Process(target=server, args=(('', 15000), c1, c2, worker_p.pid)) server_p.start() c1.close() c2.close() # 可以使用secureCRT与服务器进行连接 # 运行结果如下: # SERVER:GOT connection from ('127.0.0.1', 25309) # CHILD: GOT FD 220 # CHILD:RECV b'hello retacn\r\n'
服务器和工作者以单独的程序来运行
from multiprocessing.connection import Listener from multiprocessing.reduction import send_handle import socket def server(work_address, port): # 等待工作者来连接 work_serv = Listener(work_address, authkey=b'retacn') worker = work_serv.accept() worker_pid = worker.recv() s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) s.bind(('', port)) s.listen(1) while True: client, addr = s.accept() print('SERVER: GOT connection from ', addr) send_handle(worker, client.fileno(), worker_pid) client.close() if __name__ == "__main__": import sys if len(sys.argv) != 3: print('Usage: server.py server_address port', file=sys.stderr) raise SystemExit(1) server(sys.argv[1], int(sys.argv[2])) print(sys.argv[1], sys.argv[2])from multiprocessing.connection import Client from multiprocessing.reduction import recv_handle import os from socket import socket, AF_INET, SOCK_STREAM def worker(server_address): serv = Client(server_address, authkey=b'retacn') serv.send(os.getpid()) while True: fd = recv_handle(serv) print('WORKER: GOT FD', fd) with socket(AF_INET, SOCK_STREAM, fileno=fd) as client: while True: msg = client.recv(1024) if not msg: break print('WORKER: RECV {!r}'.format(msg)) client.send(msg) if __name__ == '__main__': import sys if len(sys.argv) != 2: print('Usage: worker.py server_address', file=sys.stderr) raise SystemExit(1) worker(sys.argv[1])
使用套接字来完成进程间传递socket文件描述符
import socket import struct def send_fd(sock, fd): ''' Send a single file descriptor. ''' sock.sendmsg([b'x'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, struct.pack('i', fd))]) ack = sock.recv(2) assert ack == b'OK' def server(work_address, port): # Wait for the worker to connect work_serv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) work_serv.bind(work_address) work_serv.listen(1) worker, addr = work_serv.accept() # Now run a TCP/IP server and send clients to worker s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) s.bind(('', port)) s.listen(1) while True: client, addr = s.accept() print('SERVER: Got connection from', addr) send_fd(worker, client.fileno()) client.close() if __name__ == '__main__': import sys if len(sys.argv) != 3: print('Usage: server.py server_address port', file=sys.stderr) raise SystemExit(1) server(sys.argv[1], int(sys.argv[2]))
import socket import struct def recv_fd(sock): ''' Receive a single file descriptor ''' msg, ancdata, flags, addr = sock.recvmsg(1, socket.CMSG_LEN(struct.calcsize('i'))) cmsg_level, cmsg_type, cmsg_data = ancdata[0] assert cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS sock.sendall(b'OK') return struct.unpack('i', cmsg_data)[0] def worker(server_address): serv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) serv.connect(server_address) while True: fd = recv_fd(serv) print('WORKER: GOT FD', fd) with socket.socket(socket.AF_INET, socket.SOCK_STREAM, fileno=fd) as client: while True: msg = client.recv(1024) if not msg: break print('WORKER: RECV {!r}'.format(msg)) client.send(msg) if __name__ == '__main__': import sys if len(sys.argv) != 2: print('Usage: worker.py server_address', file=sys.stderr) raise SystemExit(1) worker(sys.argv[1])
