Sample WSGI debug application

pdebug.py

#!/usr/local/bin/python2.7
from __future__ import print_function

import os
import threading
import time

startup_lock = threading.Lock()

COUNTER = None

def _Startup():
  global COUNTER, startup_lock
  with startup_lock:
    if COUNTER is not None:
      return
    COUNTER = CounterThread()
    COUNTER.start()

class CounterThread(threading.Thread):
  def __init__(self):
    threading.Thread.__init__(self)
    self.name = 'Counting'
    self.daemon = True
    self.value = 0
    self.timer = 0
    self.update_lock = threading.Lock()

  def run(self):
    while True:
      with self.update_lock:
        self.timer += 1
      time.sleep(1)

  def Get(self):
    with self.update_lock:
      self.value += 1
      return '%d / %d\r\n\r\n' % (self.value, self.timer)


def dump_environ(environ):
  for k in sorted(environ):
    yield '%s -> "%s"\r\n' % (k, environ[k])


def dump_threads():
  yield 'Threads: {{\r\n'
  for t in sorted(threading.enumerate()):
    aliveness = not t.is_alive() and ' DEAD' or ''
    ident = t.ident and ' [%d]' % t.ident or ''
    daemonic = not t.daemon and ' <non-daemon>' or ''
    yield '%s%s%s%s\r\n' % (t.name, aliveness, ident, daemonic)
  yield '}}\r\n'

def application(environ, start_response):
  if COUNTER is None:
    _Startup()

  start_response('200 OK', [
    ('Content-Type', 'text/plain; charset=UTF-8'),
    ])
  results = []
  results.append(''.join(dump_environ(environ)))
  results.append('\r\n\r\n')
  results.append(COUNTER.Get())
  results.append('PID: %d\r\n\r\n' % os.getpid())
  results.append(''.join(dump_threads()))
  return ''.join(results)

if __name__ == '__main__':
  from wsgiref.simple_server import make_server
  s = make_server('localhost', 8765, application)
  print('Listening at:  http://localhost:8765/')
  s.serve_forever()

# vim: set sw=2 ts=2 et :

This was written to debug a problem described at http://bridge.grumpy-troll.org/2012/03/porting-python-wsgi-app-to-nginxuwsgi.html