1. 简单的 Web 服务器
  2. 接受请求并返回结果的服务器
    1. 服务端
    2. 客户端

使用 Python 的 Web Server Gateway Interface (WSGI) 搭建 Web 服务器,参考官网:

简单的 Web 服务器

保存以下代码为 server.py,运行 python server.py,访问 http://127.0.0.1:8000 就能看到效果了。

其中,

  • environ:一个包含所有 HTTP 请求信息的 dict 对象
  • start_response:一个发送 HTTP 响应的函数
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from wsgiref.simple_server import make_server


# Every WSGI application must have an application object - a callable
# object that accepts two arguments. For that purpose, we're going to
# use a function (note that you're not limited to a function, you can
# use a class for example). The first argument passed to the function
# is a dictionary containing CGI-style environment variables and the
# second variable is the callable object (see PEP 333).
def hello_world_app(environ, start_response):
status = '200 OK' # HTTP Status
headers = [('Content-type', 'text/plain')] # HTTP Headers
start_response(status, headers)
# The returned object is going to be printed
return ["Hello World"]


httpd = make_server('', 8000, hello_world_app)
print "Serving on port 8000..."

# Serve until process is killed
httpd.serve_forever()

接受请求并返回结果的服务器

服务端

运行 python server.py 即启动服务器。

  • 使用 environ['REQUEST_METHOD'] 即可判断请求的类型。
  • 使用 environ['wsgi.input'].read(environ[CONTENT_LENGTH]) 即可读请求里的内容。

server.py:

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
"""Server.
"""
from __future__ import print_function

import os
import json
import time
import base64
import argparse
from wsgiref.simple_server import make_server


def get_args():
"""Get arguments.

Returns:
Namespace, arguments.
"""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--server_host',
type=str,
default='0.0.0.0',
help='Server host.')
parser.add_argument('--server_port',
type=int,
default=8000,
help='Server port.')
parser.add_argument('--images_dir',
type=str,
default='upload_images/',
help='Images directory.')
args = parser.parse_args()
return args


ARGS = get_args()


def application(environ, start_response):
"""Application

Args:
environ: dict, environment values.
start_response: func, start_response function.

Returns:
str, response.
"""
method = environ['REQUEST_METHOD']
path = environ['PATH_INFO']
status = '200 OK'
if method == 'POST' and path == '/upload':
headers = [('Content-type', 'application/json')]
content_length = int(environ.get('CONTENT_LENGTH', 0))
request_body = environ['wsgi.input'].read(content_length)
request_body = json.loads(request_body)
image_data = request_body['image']
image_path = os.path.join(ARGS.images_dir,
'{}.jpg'.format(time.time()))
image_data = base64.b64decode(image_data)
open(image_path, 'wb').write(image_data)
response = {'info': 'Get text: {}'.format(request_body['text'])}
start_response(status, headers)
return [json.dumps(response)]
headers = [('Content-type', 'text/plain')]
start_response(status, headers)
return 'Hello World'


def main():
"""main.
"""
print('----------- Configuration Arguments -----------')
for arg, value in sorted(vars(ARGS).items()):
print('{}: {}'.format(arg, value))
print('------------------------------------------------')
if not os.path.exists(ARGS.images_dir):
os.makedirs(ARGS.images_dir)
httpd = make_server(ARGS.server_host, ARGS.server_port, application)
print('[INFO] Serving on port {}...'.format(ARGS.server_port))
# Serve until process is killed
httpd.serve_forever()


if __name__ == '__main__':
main()

客户端

使用 urllib2 对服务进行 GET/POST 请求:运行 python client.py 即可进行简单的请求。

client.py:

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
"""Request demo.
"""
from __future__ import print_function

import json
import time
import base64
import urllib2
import argparse


def get_args():
"""Get arguments.

Returns:
Namespace, arguments.
"""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--server_url',
type=str,
default='0.0.0.0:8000',
help='Server url.')
parser.add_argument('--image_path',
type=str,
default='1.jpg',
help='Image path.')
args = parser.parse_args()
return args


def do_request(args):
"""Request.

Args:
args: args.
"""
# get
service_url = 'http://{}'.format(args.server_url)
print('{} GET'.format(time.ctime()))
request = urllib2.Request(service_url)
request.add_header('Content-Type', 'text/plain')
response = urllib2.urlopen(request, timeout=3000).read()
print(response)
# post
service_url = 'http://{}/upload'.format(args.server_url)
print('{} POST {}'.format(time.ctime(), args.image_path))
image_data = open(args.image_path, 'rb').read()
image_data = base64.b64encode(image_data)
request_json = {
'text': 'Upload {}'.format(args.image_path),
'image': image_data
}
request_json = json.dumps(request_json)
request = urllib2.Request(service_url)
request.add_header('Content-Type', 'application/json')
response = urllib2.urlopen(request, request_json, 3000).read()
response = json.loads(response)
print(response)


def main():
"""main.
"""
args = get_args()
print('----------- Configuration Arguments -----------')
for arg, value in sorted(vars(args).items()):
print('{}: {}'.format(arg, value))
print('------------------------------------------------')
do_request(args)


if __name__ == '__main__':
main()