This library follows the FastCGI spec. It only supports the Responder role, and does not support multiplexing (which is not supported by any servers, so is unlikely to be an issue).
These enums are used throughout the library, and have the same meanings as the FCGI_ constants #defined in the spec.
for o in Record,Role,Status: print(list(o))
t = 1_000_000_101
s = struct.pack('!L', t | (1<<31))
test_eq(readlen(BytesIO(s).read), t)
b = BytesIO(d).read
recv_record(b)
typ,p = recv_record(b); typ,p
recv_record(b)
params(p)
class _Stream(BytesIO):
def write(self, b):
if isinstance(b,bytes): b=b.decode()
super().write(b)
This is used in much the same way as StreamRequestHandler, except that receiving the data is handled for you before your handle method is called. All headers are available in the params dictionary. All streams are available through indexing (i.e. using __getitem__). The stdin stream contains the data sent to your handler. Write to the stdout and stderr streams to send data to the client.
Here's an example subclass:
class TestHandler(FcgiHandler):
def handle(self):
print('query:', self.environ['QUERY_STRING'])
print('content type:', self.environ.get('HTTP_CONTENT_TYPE', 'N/A'))
print('stdin:', self['stdin'].read())
self['stdout'].write(b"Content-type: text/html\r\n\r\n<html>foobar</html>\r\n")
To test it, we'll use an http➡fcgi proxy. We can download http2fcgi and run it in the background as follows:
run('./get_http2fcgi.sh')
proc = subprocess.Popen(['./http2fcgi'])
We can now test the handler by running a server in the background...
p = Path('fcgi.sock')
if p.exists(): p.unlink()
@threaded
def _f():
with UnixStreamServer(str(p), TestHandler) as srv: srv.handle_request()
_f()
time.sleep(0.2) # wait for server to start
...and use curl to test it:
urlread('http://localhost:6065/setup.py?a=1', foo='bar', greeting='你好')
Finally, we kill the http2fcgi background process.
proc.terminate()
Instead of self.stdout.write(...) (which requires byte strings and \r\n line endings, and does not append a line ending automatically) we can use print.
For errors, you can either write to stderr, or call err, which is like print, but for stderr.
Here we repeat the previous example, but using some of these helper functions:
class TestHandler(FcgiHandler):
def handle(self):
print('stdin:', self.content())
self.write(b"Content-type: text/html\r\n\r\n<html>foobar</html>\r\n")
self.print("<html>foobar</html>")
proc = subprocess.Popen(['./http2fcgi'])
if p.exists(): p.unlink()
t = _f()
time.sleep(0.2)
print(urlread('http://localhost:6065/setup.py?a=1', foo='bar', greeting='你好'))
proc.terminate()