[Web-SIG] [WSGI] mod_python wrapper: minimal first attempt

[Robert Brewer]

In order to test my application's WSGI interface, I wrote a quick
mod_python server interface for WSGI. It's not bulletproof, but the
parts I use work. Sorry, Phillip, I didn't subclass
wsgiref.handlers.BaseHandler yet. ;(


class ModPythonInputWrapper(object):
    
    def __init__(self, req):
        self.req = req
    
    def read(self, size=-1):
        return self.req.read(size)
    
    def readline(self):
        return self.req.readline()
    
    def readlines(self, hint=-1):
        return self.req.readlines(hint)
    
    def __iter__(self):
        return iter(self.req.readlines())


class ModPythonErrorWrapper(object):
    
    def __init__(self, req):
        self.req = req
    
    def flush(self):
        pass
    
    def write(self, content):
        self.req.log_error(content)
    
    def writelines(self, seq):
        for content in seq:
            self.req.log_error(content)


def wrap_mod_python(application, req):
    """WSGI wrapper for mod_python 3.1 (Apache 2).
    
    Write your own short handler function, obtain your application,
    and pass it and the apache Request object to this function.
    """
    
    from mod_python import apache
    
    req.add_common_vars()
    environ = dict(req.subprocess_env.items())
    environ['wsgi.input']        = ModPythonInputWrapper(req)
    environ['wsgi.errors']       = ModPythonErrorWrapper(req)
    environ['wsgi.version']      = (1, 0)
    environ['wsgi.multithread']  = True
    environ['wsgi.multiprocess'] = False
    if req.protocol.count(u'HTTPS') > 0:
        environ['wsgi.url_scheme'] = 'https'
    else:
        environ['wsgi.url_scheme'] = 'http'
    
    nested_status = [apache.OK]
    
    def start_response(status, headers):
        if status:
            if status == "200 OK":
                nested_status[0] = apache.OK
            else:
                nested_status[0] = int(status[:3])
        for key, val in headers:
            req.headers_out[key] = val
        return req.write
    
    result = application(environ, start_response)
    try:
        for data in result:
            req.write(data)
    finally:
        if hasattr(result,'close'):
            result.close()
    return nested_status[0]

-----------

Example handler (for Junct, my wiki, built on Cation, my app framework):

from cation.html import uiwsgi
import junct

def handler(req):
    ui = uiwsgi.UserInterfaceWSGI(junct.junctapp)
    ui.sandbox = junct.arena.new_sandbox()
    app = ui.request
    result = uiwsgi.wrap_mod_python(app, req)
    ui.sandbox.flush_all()
    return result



Robert Brewer
MIS
Amor Ministries
fumanchu at amor.org