Python实战之Python网络编程方面的一些笔记

写在前面

• 今天和小伙伴们分享一些Python网络编程的笔记
• 博文为《Python Cookbook》读书后笔记整理
• 涉及内容包括：

  • python实现作为客户端与HTTP服务交互
  • 创建TCP服务器
  • 创建UDP服务器
  • Python通过IP掩码生成可分配的IP地址集
  • 不使用框架创建一个REST风格的web服务
• 理解不足小伙伴帮忙指正

「 傍晚时分，你坐在屋檐下，看着天慢慢地黑下去，心里寂寞而凄凉，感到自己的生命被剥夺了。当时我是个年轻人，但我害怕这样生活下去，衰老下去。在我看来，这是比死亡更可怕的事。--------王小波」

作为客户端与HTTP服务交互

「你需要通过 HTTP 协议以客户端的方式访问多种服务。例如，下载数据或者与基于 REST 的 API 进行交互。」

对于简单的事情来说，通常使用 urllib.request 模块就够了.一个Get请求的Demo

┌──[root@liruilongs.github.io]-[~]
└─$python3
Python 3.6.8 (default, Nov 16 2020, 16:55:22)
[GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from urllib import request, parse
>>> url = 'http://httpbin.org/get'
>>> parms = {
...     'name1': 'value1',
...     'name2': 'value2'
... }
>>> querystring = parse.urlencode(parms)
>>> querystring
'name1=value1&name2=value2'
>>> request.urlopen(url+'?' + querystring)
<http.client.HTTPResponse object at 0x7ffa0ef0f710>
>>> u = request.urlopen(url+'?' + querystring)
>>> u.read()
b'{
    "args": {
        "name1": "value1",
        "name2": "value2"
    },
    "headers": {
        "Accept-Encoding": "identity",
        "Host": "httpbin.org",
        "User-Agent": "Python-urllib/3.6",
        "X-Amzn-Trace-Id": "Root=1-62707b15-41a1169c0897c9001a07f948"
    },
    "origin": "39.154.13.139",
    "url": "http://httpbin.org/get?name1=value1&name2=value2"
}'

如果你需要使用 POST 方法在请求主体中发送查询参数，可以将参数编码后作为可选参数提供给 urlopen() 函数，就像这样：

>>> from urllib import request, parse
>>> url = 'http://httpbin.org/post'
>>> parms = {
... 'name1' : 'value1',
... 'name2' : 'value2'
... }
>>> querystring = parse.urlencode(parms)
>>> querystring.encode('ascii')
b'name1=value1&name2=value2'
>>> u = request.urlopen(url, querystring.encode('ascii'))
>>> resp = u.read()
>>> resp
b'{
    "args": {},
    "data": "",
    "files": {},
    "form": {
        "name1": "value1",
        "name2": "value2"
    },
    "headers": {
        "Accept-Encoding": "identity",
        "Content-Length": "25",
        "Content-Type": "application/x-www-form-urlencoded",
        "Host": "httpbin.org",
        "User-Agent": "Python-urllib/3.6",
        "X-Amzn-Trace-Id": "Root=1-62707d24-15e9944760d3bbaa36c3714a"
    },
    "json": null,
    "origin": "39.154.13.139",
    "url": "http://httpbin.org/post"
}'
>>>

在发出的请求中提供一些自定义的 HTTP 请求首部，创建一个 Request 实例然后将其传给urlopen()

>>> from urllib import request, parse
>>> headers = {
... 'User-agent' : 'none/ofyourbusiness',
... 'Spam' : 'Eggs'
... }
>>> req = request.Request(url, querystring.encode('ascii'), headers=headers)
>>> u = request.urlopen(req)
>>> u.read()
b'{
    "args": {},
    "data": "",
    "files": {},
    "form": {
        "name1": "value1",
        "name2": "value2"
    },
    "headers": {
        "Accept-Encoding": "identity",
        "Content-Length": "25",
        "Content-Type": "application/x-www-form-urlencoded",
        "Host": "httpbin.org",
        "Spam": "Eggs",
        "User-Agent": "none/ofyourbusiness",
        "X-Amzn-Trace-Id": "Root=1-62707f0e-308a8137555e15797d950018"
    },
    "json": null,
    "origin": "39.154.13.139",
    "url": "http://httpbin.org/post"
}'
>>>

如果需要交互的服务,可以使用 requests 模块， 这个不是自带模块，需要安装python3 -m pip install requests

>>> import requests
>>> url = 'http://httpbin.org/post'
>>> parms = {
... 'name1' : 'value1',
... 'name2' : 'value2'
... }
>>> headers = {
... 'User-agent' : 'none/ofyourbusiness',
... 'Spam' : 'Eggs'
... }
>>> resp = requests.post(url, data=parms, headers=headers)
>>> resp.text
'{\n  "args": {}, \n  "data": "", \n  "files": {}, \n  "form": {\n    "name1": "value1", \n    "name2": "value2"\n  }, \n  "headers": {\n    "Accept": "*/*", \n    "Accept-Encoding": "gzip, deflate", \n    "Content-Length": "25", \n    "Content-Type": "application/x-www-form-urlencoded", \n    "Host": "httpbin.org", \n    "Spam": "Eggs", \n    "User-Agent": "none/ofyourbusiness", \n    "X-Amzn-Trace-Id": "Root=1-62708080-7a14319e699baa2e35a352fb"\n  }, \n  "json": null, \n  "origin": "39.154.13.139", \n  "url": "http://httpbin.org/post"\n}\n'
>>> resp.json()
{'args': {}, 'data': '', 'files': {}, 'form': {'name1': 'value1', 'name2': 'value2'}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '25', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'httpbin.org', 'Spam': 'Eggs', 'User-Agent': 'none/ofyourbusiness', 'X-Amzn-Trace-Id': 'Root=1-62708080-7a14319e699baa2e35a352fb'}, 'json': None, 'origin': '39.154.13.139', 'url': 'http://httpbin.org/post'}
>>>

requests 模块支持很多种数据的返会方式，可以直接返回以 Unicode 解码的响应文本,也可以返回JSON数据

利用 requests 库发起一个 HEAD 请求

>>> import requests
>>> resp = requests.head( 'http://httpbin.org/post')
>>> resp
<Response [405]>
>>> resp = requests.head( 'http://httpbin.org/')
>>> resp
<Response [200]>
>>> resp.status_code
200
>>> resp.headers['content-length']
'9593'
>>> resp.headers['content-type']
'text/html; charset=utf-8'
>>> resp.text
''
>>>

如果你决定坚持使用标准的程序库而不考虑像requests这样的第三方库，可以使用底层的 http.client 模块来实现自己的代码。

from http.client import HTTPConnection
from urllib import parse


c = HTTPConnection('www.python.org', 80)
c.request('HEAD', '/index.html')
resp = c.getresponse()
print('Status', resp.status)
for name, value in resp.getheaders():
    print(name, value)

测试 HTTP 客户端,考虑使用httpbin服务(http://httpbin.org)。这个站点会接收发出的请求，然后以JSON 的形式将相应信息回传回来。

>>> import requests
>>> r = requests.get('http://httpbin.org/get?name=Dave&n=37',
... headers = { 'User-agent': 'goaway/1.0' })
>>> resp = r.json()
>>> resp['headers']
{'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Host': 'httpbin.org', 'User-Agent': 'goaway/1.0', 'X-Amzn-Trace-Id': 'Root=1-62708c06-7c7d8cc4441479c65faea5b4'}
>>>

创建TCP服务器

「你想实现一个服务器，通过 TCP 协议和客户端通信。」

创建一个 TCP 服务器的一个简单方法是使用 socketserver 库。下面是一个简单的TCP服务器

from socketserver import BaseRequestHandler, TCPServer


class EchoHandler(BaseRequestHandler):
    def handle(self):
        print('Got connection from', self.client_address)
        while True:
            #接收客户端发送的数据, 这次接收数据的最大字节数是8192
            msg = self.request.recv(8192)
            # 接收的到数据在发送回去
            if not msg:
                break
            self.request.send(msg)

if __name__ == '__main__':
    # 20000端口，默认IP为本地IP，监听到消息交个EchoHandler处理器
    serv = TCPServer(('', 20000), EchoHandler)
    serv.serve_forever()

Got connection from ('127.0.0.1', 1675)

建立好服务端之后我们看下客户端，AF_INET：表示ipv4,SOCK_STREAM: tcp传输协议

>>> from socket import socket, AF_INET, SOCK_STREAM
>>> s = socket(AF_INET, SOCK_STREAM)
>>> s.connect(('localhost', 20000))
>>> s.send(b'Hello')
5
>>> s.recv(8192)
b'Hello'
>>>

使用StreamRequestHandler 基类将一个类文件接口放置在底层 socket 上的例子,StreamRequestHandler类支持像操作文件对象那样操作输入套字节;嗯，这个Demo还有问题，时间关系，之前在研究下

from socketserver import StreamRequestHandler, TCPServer


class EchoHandler(StreamRequestHandler):
    def handle(self):
        print('Got connection from', self.client_address)
        #   rfile: 这个文件对应着从 socket 进来的数据
        for line in self.rfile:
            #  wfile: 这个文件对应着从 socket 发送的数据
            self.wfile.write(line)
   

if __name__ == '__main__':
    serv = TCPServer(('', 20000), EchoHandler)
    serv.serve_forever()

同时 StreamRequestHandler 支持通过类变量来设置连接参数

import socket 
from socketserver import StreamRequestHandler

class EchoHandler(StreamRequestHandler):
    # Optional settings (defaults shown)
    timeout = 5  # Timeout on all socket operations
    rbufsize = -1  # Read buffer size
    wbufsize = 0  # Write buffer size
    disable_nagle_algorithm = False  # Sets TCP_NODELAY socket option


def handle(self):
    print('Got connection from', self.client_address)
    try:
        for line in self.rfile:
            # self.wfile is a file-like object for writing
            self.wfile.write(line)
    except socket.timeout:
        print('Timed out!')

socketserver 默认情况下这种服务器是单线程的，一次只能为一个客户端连接服务。如果你想处理多个客户端，可以初始化一个ForkingTCPServer 或者是ThreadingTCPServer对象。

from socketserver import ThreadingTCPServer

if __name__ == '__main__':
    serv = ThreadingTCPServer(('', 20000), EchoHandler)
    serv.serve_forever()

使用 fork 或线程服务器有个潜在问题就是它们会为每个客户端连接创建一个新的进程或线程。由于客户端连接数是没有限制的，因此一个恶意的黑客可以同时发送大量的连接让你的服务器奔溃。

可以创建一个预先分配大小的 工作线程池或进程池 先创建一个普通的非线程服务器，然后在一个线程池中使用serve forever()方法来启动它们。

if __name__ == '__main__':
    from threading import Thread
    NWORKERS = 16
    serv = TCPServer(('', 20000), EchoHandler)
    for n in range(NWORKERS):
        t = Thread(target=serv.serve_forever)
        t.daemon = True
        t.start()
    serv.serve_forever()

一般来讲，一个TCPServer在实例化的时候会绑定并激活相应的socket，有时候你想通过设置某些选项去调整底下的socket，可以设置参数bind_and_activate=False。

if __name__ == '__main__':
serv = TCPServer(('', 20000), EchoHandler, bind_and_activate=False)
# Set up various socket options
serv.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
# Bind and activate
serv.server_bind()
serv.server_activate()
serv.serve_forever()

socket.SO_REUSEADDR允许服务器重新绑定一个之前使用过的端口号。由于要被经常使用到，它被放置到类变量中，可以直接在 TCPServer上面设置

也可以不使用一个工具类，直接使用原生的Socket的编写TCP服务端

from socket import socket, AF_INET, SOCK_STREAM


def echo_handler(address, client_sock):
    print('Got 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))

创建UDP服务器

「你想实现一个基于 UDP 协议的服务器来与客户端通信。」

跟 TCP 一样，UDP 服务器也可以通过使用socketserver库很容易的被创建。例如，下面是一个简单的时间服务器：

from socketserver import BaseRequestHandler, UDPServer
import time
class TimeHandler(BaseRequestHandler):
    def handle(self):
        print('Got connection from', self.client_address)
        # Get message and client socket request 属性是一个包含了数据报和底层 socket 对象的元组
        msg, sock = self.request
        resp = time.ctime()
        sock.sendto(resp.encode('ascii'), self.client_address)

if __name__ == '__main__':
    serv = UDPServer(('', 20000), TimeHandler)
    serv.serve_forever()

测试一下

>>> from socket import socket, AF_INET, SOCK_DGRAM
>>> s = socket(AF_INET, SOCK_DGRAM)
>>> s.sendto(b'', ('localhost', 20000))
0
>>> s.recvfrom(8192)
(b'Tue May  3 11:48:53 2022', ('127.0.0.1', 20000))
>>>

对于UPD协议而言，对于数据报的传送，你应该使用 socket 的sendto() 和 recvfrom() 方法

通过CIDR地址生成对应的IP地址集

「你有一个 CIDR 网络地址比如“123.45.67.89/27”，你想将其转换成它所代表的所有 IP (比如，“123.45.67.64”, “123.45.67.65”, …, “123.45.67.95”))」

意思获取当前掩码指定网段内的所有可用IP，可以使用 ipaddress 模块很容易的实现这样的计算

>>> net = ipaddress.ip_network('192.168.26.0/24')
>>> for a in net:
...     print(a)
...
192.168.26.0
192.168.26.1
192.168.26.2
192.168.26.3
........
192.168.26.254
192.168.26.255
>>>

顺便回顾一下掩码的知识，掩码为27位

>>> net = ipaddress.ip_network('192.168.26.0/27')
>>> for a in net:
...     print(a)
...
192.168.26.0
192.168.26.1
192.168.26.2
....
192.168.26.29
192.168.26.30
192.168.26.31

ipv6 同样也可以

>>> net6 = ipaddress.ip_network('12:3456:78:90ab:cd:ef01:23:30/125')
>>> for a in net6:
...     print(a)
...
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
>>>

Network 也允许像数组一样的索引取值,还可以执行网络成员检查之类的操作,一个 IP 地址和网络地址能通过一个 IP 接口来指定

>>> net6.num_addresses
8
>>> net6[3]
IPv6Address('12:3456:78:90ab:cd:ef01:23:33')
>>> a in net6
True
>>> a
IPv6Address('12:3456:78:90ab:cd:ef01:23:37')
>>> inet = ipaddress.ip_interface('123.45.67.73/27')
>>> inet.network
IPv4Network('123.45.67.64/27')
>>> inet.ip
IPv4Address('123.45.67.73')
>>>

创建一个简单的REST接口

使用一个简单的 REST 接口通过网络远程控制或访问你的应用程序，但是你又不想自己去安装一个完整的 web 框架。

构建一个 REST 风格的接口最简单的方法是创建一个基于 WSGI 标准(PEP 3333)的很小的库

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@File    :   app.py
@Time    :   2022/05/03 14:43:56
@Author  :   Li Ruilong
@Version :   1.0
@Contact :   1224965096@qq.com
@Desc    :   None
"""

# here put the import lib

import time
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, 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


_hello_resp = "wo jiao {name}"


def hello_world(environ, start_response):
    start_response('200 OK', [('Content-type', 'text/html')])
    params = environ['params']
    resp = _hello_resp.format(name=params.get('name'))
    yield resp.encode('utf-8')


_localtime_resp = "dang qian shjian {t}"

# 路由的回调
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 wsgiref.simple_server import make_server
    # Create the dispatcher and register functions
    dispatcher = PathDispatcher()
    # 注册路由，对应的回调
    dispatcher.register('GET', '/hello', hello_world)
    dispatcher.register('GET', '/localtime', localtime)
    # Launch a basic server 监听8080端口，注入核心控制器
    httpd = make_server('', 8080, dispatcher)
    print('Serving on port 8080...')
    httpd.serve_forever()

测试一下

┌──[root@liruilongs.github.io]-[~]
└─$coproc (./app.py)
[2] 130447
┌──[root@liruilongs.github.io]-[~]
└─$cutl localhost:8080/
-bash: cutl: 未找到命令
┌──[root@liruilongs.github.io]-[~]
└─$curl  localhost:8080/
127.0.0.1 - - [03/May/2022 16:09:00] "GET / HTTP/1.1" 404 9
Not Found┌──[root@liruilongs.github.io]-[~]
└─$curl  localhost:8080/hello
127.0.0.1 - - [03/May/2022 16:09:12] "GET /hello HTTP/1.1" 200 12
wo jiao None┌──[root@liruilongs.github.io]-[~]
└─$curl  localhost:8080/hello?name=liruilong
127.0.0.1 - - [03/May/2022 16:09:47] "GET /hello?name=liruilong HTTP/1.1" 200 17
wo jiao liruilong┌──[root@liruilongs.github.io]-[~]
└─$jobs
[1]+  已停止               curl -X 'HEAD' -v 'https://www.python.org/index.html'
[2]-  运行中               coproc COPROC ( ./app.py ) &

实现一个简单的 REST 接口，你只需让你的程序代码满足 Python 的 WSGI标准即可。WSGI 被标准库支持，同时也被绝大部分第三方 web 框架支持。

长期运行的程序可能会使用一个 REST API 来实现监控或诊断。大数据应用程序可以使用 REST 来构建一个数据查询或提取系统。REST 还能用来控制硬件设备比如机器人、传感器、工厂或灯泡。更重要的是，REST API 已经被大量客户端编程环境所支持

看到这里，感觉python 的WSGI标准和Java Web 体系的Servlet规范特别接近，但是Servlet是侵入式的，而WSGI好像对代码的影响很少...感兴趣小伙伴可以研究下.

以一个可调用对象形式来实现路由匹配要操作的方法

import cgi

def wsgi_app(environ, start_response):
    pass

environ 属性是一个字典，包含了从 web 服务器如 Apache[参考 Internet RFC 3875]提供的 CGI 接口中获取的值。要将这些不同的值提取出来，你可以像这么这样写：

def wsgi_app(environ, start_response):
    method = environ['REQUEST_METHOD']
    path = environ['PATH_INFO']
    # Parse the query parameters
    params = cgi.FieldStorage(environ['wsgi.input'], environ=environ)

start_response 参数是一个为了初始化一个请求对象而必须被调用的函数。第一个参数是返回的 HTTP 状态值，第二个参数是一个 (名, 值) 元组列表，用来构建返回的 HTTP 头。

def wsgi_app(environ, start_response):
    pass
    start_response('200 OK', [('Content-type', 'text/plain')])

为了返回数据，一个 WSGI 程序必须返回一个字节字符串序列。可以像下面这样使用一个列表来完成

def wsgi_app(environ, start_response):
    pass
    start_response('200 OK', [('Content-type', 'text/plain')])
    resp = []
    resp.append(b'Hello World\n')
    resp.append(b'Goodbye!\n')
    return resp

或者，你还可以使用 yield

def wsgi_app(environ, start_response):
    pass
    start_response('200 OK', [('Content-type', 'text/plain')])
    yield b'Hello World\n'
    yield b'Goodbye!\n'

最后返回的必须是字节字符串。如果返回结果包含文本字符串，必须先将其编码成字节。图片也是OK的

class WSGIApplication:
    def __init__(self):
        ...
    def __call__(self, environ, start_response)
        ...

PathDispatcher 类。这个分发器仅仅只是管理一个字典，将 (方法, 路径) 对映射到处理器函数上面。当一个请求到来时，它的方法和路径被提取出来，然后被分发到对应的处理器上面去。

    dispatcher = PathDispatcher()
    # 注册路由，对应的回调
    dispatcher.register('GET', '/hello', hello_world)
    dispatcher.register('GET', '/localtime', localtime)

任何查询变量会被解析后放到一个字典中，以 environ['params'] 形式存储。后面这个步骤太常见，所以建议你在分发器里面完成，这样可以省掉很多重复代码。使用分发器的时候，你只需简单的创建一个实例，然后通过它注册各种 WSGI 形式的函数。编写这些函数应该超级简单了，只要你遵循 start_response() 函数的编写规则，并且最后返回字节字符串 即可。

WSGI 本身是一个很小的标准。因此它并没有提供一些高级的特性比如认证、cookies、重定向等。这些你自己实现起来也不难。不过如果你想要更多的支持，可以考虑第三方库