WSGI(Web Server Gateway Interface,Web 服务器网关接口)是 Python 语言中定义的一种标准接口,用于规范 Web 服务器 和 Python Web 应用程序之间的通信。它存在的主要目的就是提供一个统一的、标准化的协议,使得任何支持 WSGI 的 Web 服务器都可以运行任何支持 WSGI 的 Python Web 应用程序,从而实现 python应用服务器与 python应用程序之间的解耦。
WSGI 协议第一版推出的时间是2003年12月[v1.01],最后的改进版[v1.01]定格在2010年10月。在 WSGI 出现之前,框架与服务器强耦合, 每个Python框架(如Django、Flask)需自行实现 Apache 的模块适配代码 ;如编写 apache.Request 处理逻辑,Django 是一种编码方式,而 Flask 可能是另一种编码方式。相比之下,Java web开发就没有这些烦恼, 因为所有框架都基于 Servlet API,Tomcat 是一个支持 Servlet API的 java 应用服务器,可以直接运行任何基于 Servlet API 的 Java Web 应用。
WSGI协议简介
WSGI 协议主要涉及三个方面:
- Web 服务器(servers/gateways):负责接收 HTTP 请求并将其转换为 WSGI 格式的 environ 字典,
常见的 WSGI 服务器包括 Gunicorn、uWSGI、Waitress。
- 应用程序(applications/frameworks): 一个可调用的对象(如函数或类),接收 environ 和 start_response,并返回响应体
- 中间件(middleware): 介于服务器和应用程序之间的处理层,用于扩展功能或修改请求/响应流程
application 是一个简单的可调用的对象,它接收 两个参数。术语“对象”不应被误解为 一个对象的实例:一个函数、方法、类、 或实例都可以用作应用程序的对象。下面是两个示例应用程序对象:一个是函数,一个是类。
HELLO_WORLD = b"Hello world!\n" def simple_app(environ, start_response): """Simplest possible application object""" status = '200 OK' response_headers = [('Content-type', 'text/plain')] start_response(status, response_headers) return [HELLO_WORLD] class AppClass: """Produce the same output, but using a class (Note: 'AppClass' is the "application" here, so calling it returns an instance of 'AppClass', which is then the iterable return value of the "application callable" as required by the spec. If we wanted to use *instances* of 'AppClass' as application objects instead, we would have to implement a '__call__' method, which would be invoked to execute the application, and we would need to create an instance for use by the server or gateway. """ def __init__(self, environ, start_response): self.environ = environ self.start = start_response def __iter__(self): status = '200 OK' response_headers = [('Content-type', 'text/plain')] self.start(status, response_headers) yield HELLO_WORLD
服务器/网关 每收到一个来自 HTTP 客户端的、针对该应用程序的请求,就会调用一次应用程序可调用函数。为了说明这一点,下面是一个简单的 CGI 网关,它接受应用程序对象的函数。这个简单示例的错误处理有限,因为默认情况下,未捕获的异常将被转储到 Web 服务器并记录。
import os, sys
enc, esc = sys.getfilesystemencoding(), 'surrogateescape'
def unicode_to_wsgi(u):
# Convert an environment variable to a WSGI "bytes-as-unicode" string
return u.encode(enc, esc).decode('iso-8859-1')
def wsgi_to_bytes(s):
return s.encode('iso-8859-1')
def run_with_cgi(application):
environ = {k: unicode_to_wsgi(v) for k,v in os.environ.items()}
environ['wsgi.input'] = sys.stdin.buffer
environ['wsgi.errors'] = sys.stderr
environ['wsgi.version'] = (1, 0)
environ['wsgi.multithread'] = False
environ['wsgi.multiprocess'] = True
environ['wsgi.run_once'] = True
if environ.get('HTTPS', 'off') in ('on', '1'):
environ['wsgi.url_scheme'] = 'https'
else:
environ['wsgi.url_scheme'] = 'http'
headers_set = []
headers_sent = []
def write(data):
out = sys.stdout.buffer
if not headers_set:
raise AssertionError("write() before start_response()")
elif not headers_sent:
# Before the first output, send the stored headers
status, response_headers = headers_sent[:] = headers_set
out.write(wsgi_to_bytes('Status: %s\r\n' % status))
for header in response_headers:
out.write(wsgi_to_bytes('%s: %s\r\n' % header))
out.write(wsgi_to_bytes('\r\n'))
out.write(data)
out.flush()
def start_response(status, response_headers, exc_info=None):
if exc_info:
try:
if headers_sent:
# Re-raise original exception if headers sent
raise exc_info[1].with_traceback(exc_info[2])
finally:
exc_info = None # avoid dangling circular ref
elif headers_set:
raise AssertionError("Headers already set!")
headers_set[:] = [status, response_headers]
# Note: error checking on the headers should happen here,
# *after* the headers are set. That way, if an error
# occurs, start_response can only be re-called with
# exc_info set.
return write
result = application(environ, start_response)
try:
for data in result:
if data: # don't send headers until body appears
write(data)
if not headers_sent:
write(b'') # send headers now if body was empty
finally:
if hasattr(result, 'close'):
result.close()
下面是一个中间件组件示例,“真正的”中间件组件可能会使用更健壮的方式来检查内容类型,并且还应该检查内容编码。
from piglatin import piglatin class LatinIter: """Transform iterated output to piglatin, if it's okay to do so Note that the "okayness" can change until the application yields its first non-empty bytestring, so 'transform_ok' has to be a mutable truth value. """ def __init__(self, result, transform_ok): if hasattr(result, 'close'): self.close = result.close self._next = iter(result).__next__ self.transform_ok = transform_ok def __iter__(self): return self def __next__(self): data = self._next() if self.transform_ok: return piglatin(data) # call must be byte-safe on Py3 else: return data class Latinator: # by default, don't transform output transform = False def __init__(self, application): self.application = application def __call__(self, environ, start_response): transform_ok = [] def start_latin(status, response_headers, exc_info=None): # Reset ok flag, in case this is a repeat call del transform_ok[:] for name, value in response_headers: if name.lower() == 'content-type' and value == 'text/plain': transform_ok.append(True) # Strip content-length if present, else it'll be wrong response_headers = [(name, value) for name, value in response_headers if name.lower() != 'content-length' ] break write = start_response(status, response_headers, exc_info) if transform_ok: def write_latin(data): write(piglatin(data)) # call must be byte-safe on Py3 return write_latin else: return write return LatinIter(self.application(environ, start_latin), transform_ok) # Run foo_app under a Latinator's control, using the example CGI gateway from foo_app import foo_app run_with_cgi(Latinator(foo_app))
Odoo 内置的三种 WSGI 服务器
odoo 中内置了三种不同类型的 WSGI 服务器,它们的运作模式和定位也各有不同,以下我们来逐个介绍。
ThreadedServer
ThreadedServer 的 WSGI 服务是通过
werkzeug.serving.ThreadedWSGIServer 来实现的,ThreadedWSGIServer
是 werkzeug 库中内置的多线程 WSGI 服务器实现,专门用于本地开发与调试场景。
PreforkServer
PreforkServer 的 WSGI 服务设计灵感来自 Gunicorn,是一种多处理器模式的 WSGI 服务器,引入了工作者进程 worker 的概念,主进程监听指定的端口号,如8069;然后创建并管理多个工作者进程,工作者进程共享主进程的监听端口,主进程接收的每一个 web 请求,都交给工作者进程来处理,适合在生产环境中部署。
GeventServer
GeventServer的 WSGI 服务是通过
gevent.pywsgi.WSGIServer 来实现的, gevent 是一个基于协程的 Python 网络库,它使用 libev 或 libuv 事件循环来提供高性能的并发支持。
每个请求都在一个协程中处理,避免了线程切换的开销,适合 I/O 密集型任务,比如即时通信。
由于 gevent 是单线程的,不适合处理 CPU 密集型任务。