~loggerhead-team/loggerhead/trunk-rich : contents of wsgitest.py at revision 159.2.22

~loggerhead-team/loggerhead/trunk-rich : (revision 159.2.22)

import cgi, os, tempfile
from bzrlib import branch, errors
from loggerhead.history import History
from loggerhead.wsgiapp import BranchWSGIApp, static_app
from paste.request import path_info_pop
from paste.wsgiwrappers import WSGIRequest, WSGIResponse
from paste import httpexceptions
from paste import httpserver
from paste.httpexceptions import make_middleware
from paste.translogger import make_filter
from loggerhead.changecache import FileChangeCache


sql_dir = tempfile.mkdtemp()

class BranchesFromFileSystemServer(object):
    def __init__(self, folder, root):
        self.folder = folder
        self.root = root

    def directory_listing(self, path, environ, start_response):
        request = WSGIRequest(environ)
        response = WSGIResponse()
        listing = [d for d in os.listdir(path) if not d.startswith('.')]
        response.headers['Content-Type'] = 'text/html'
        print >> response, '<html><body>'
        for d in sorted(listing):
            if os.path.isdir(os.path.join(path, d)):
                d = cgi.escape(d)
                print >> response, '<li><a href="%s/">%s</a></li>' % (d, d)
        print >> response, '</body></html>'
        return response(environ, start_response)

    def app_for_branch(self, b, path):
        b.lock_read()
        try:
            _history = History.from_branch(b)
            _history.use_file_cache(FileChangeCache(_history, sql_dir))
            if not self.folder:
                name = os.path.basename(os.path.abspath(path))
            else:
                name = self.folder
            h = BranchWSGIApp(_history, name).app
            self.root.cache[path] = h
            return h
        finally:
            b.unlock()

    def __call__(self, environ, start_response):
        path = os.path.join(self.root.folder, self.folder)
        if not os.path.isdir(path):
            raise httpexceptions.HTTPNotFound()
        if path in self.root.cache:
            return self.root.cache[path](environ, start_response)
        try:
            b = branch.Branch.open(path)
        except errors.NotBranchError:
            segment = path_info_pop(environ)
            if segment is None:
                raise httpexceptions.HTTPMovedPermanently(environ['SCRIPT_NAME'] + '/')
            elif segment == '':
                return self.directory_listing(path, environ, start_response)
            else:
                relpath = os.path.join(self.folder, segment)
                return BranchesFromFileSystemServer(relpath, self.root)(
                    environ, start_response)
        else:
            return self.app_for_branch(b, path)(environ, start_response)


class BranchesFromFileSystemRoot(object):
    def __init__(self, folder):
        self.cache = {}
        self.folder = folder
    def __call__(self, environ, start_response):
        environ['loggerhead.static.url'] = environ['SCRIPT_NAME']
        if environ['PATH_INFO'].startswith('/static/'):
            segment = path_info_pop(environ)
            assert segment == 'static'
            return static_app(environ, start_response)
        else:
            return BranchesFromFileSystemServer(
                '', self)(environ, start_response)

app = BranchesFromFileSystemRoot('.')

app = app
app = make_middleware(app)
app = make_filter(app, None)


httpserver.serve(app, host='127.0.0.1', port='9876')


159.2.13 by Michael Hudson something like a directory listing	1	import cgi, os, tempfile
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	2	from bzrlib import branch, errors
159.2.1 by Michael Hudson things start to work a little already!	3	from loggerhead.history import History
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	4	from loggerhead.wsgiapp import BranchWSGIApp, static_app
	5	from paste.request import path_info_pop
159.2.13 by Michael Hudson something like a directory listing	6	from paste.wsgiwrappers import WSGIRequest, WSGIResponse
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	7	from paste import httpexceptions
159.2.1 by Michael Hudson things start to work a little already!	8	from paste import httpserver
159.2.2 by Michael Hudson more workingness	9	from paste.httpexceptions import make_middleware
159.2.4 by Michael Hudson more progress	10	from paste.translogger import make_filter
159.2.12 by Michael Hudson improvements borne out of the fires of contact with reality &c	11	from loggerhead.changecache import FileChangeCache
159.2.4 by Michael Hudson more progress	12
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	13
159.2.13 by Michael Hudson something like a directory listing	14	sql_dir = tempfile.mkdtemp()
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	15
	16	class BranchesFromFileSystemServer(object):
	17	def __init__(self, folder, root):
	18	self.folder = folder
	19	self.root = root
	20
159.2.22 by Michael Hudson extract some methods	21	def directory_listing(self, path, environ, start_response):
	22	request = WSGIRequest(environ)
	23	response = WSGIResponse()
	24	listing = [d for d in os.listdir(path) if not d.startswith('.')]
	25	response.headers['Content-Type'] = 'text/html'
	26	print >> response, '<html><body>'
	27	for d in sorted(listing):
	28	if os.path.isdir(os.path.join(path, d)):
	29	d = cgi.escape(d)
	30	print >> response, '<li><a href="%s/">%s</a></li>' % (d, d)
	31	print >> response, '</body></html>'
	32	return response(environ, start_response)
	33
	34	def app_for_branch(self, b, path):
	35	b.lock_read()
	36	try:
	37	_history = History.from_branch(b)
	38	_history.use_file_cache(FileChangeCache(_history, sql_dir))
	39	if not self.folder:
	40	name = os.path.basename(os.path.abspath(path))
	41	else:
	42	name = self.folder
	43	h = BranchWSGIApp(_history, name).app
	44	self.root.cache[path] = h
	45	return h
	46	finally:
	47	b.unlock()
	48
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	49	def __call__(self, environ, start_response):
159.2.20 by Michael Hudson be more flexible about where wsgitest.py can be run from	50	path = os.path.join(self.root.folder, self.folder)
	51	if not os.path.isdir(path):
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	52	raise httpexceptions.HTTPNotFound()
159.2.20 by Michael Hudson be more flexible about where wsgitest.py can be run from	53	if path in self.root.cache:
	54	return self.root.cache[path](environ, start_response)
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	55	try:
159.2.20 by Michael Hudson be more flexible about where wsgitest.py can be run from	56	b = branch.Branch.open(path)
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	57	except errors.NotBranchError:
159.2.20 by Michael Hudson be more flexible about where wsgitest.py can be run from	58	segment = path_info_pop(environ)
	59	if segment is None:
159.2.21 by Michael Hudson redirect from /directory to /directory/	60	raise httpexceptions.HTTPMovedPermanently(environ['SCRIPT_NAME'] + '/')
	61	elif segment == '':
159.2.22 by Michael Hudson extract some methods	62	return self.directory_listing(path, environ, start_response)
159.2.20 by Michael Hudson be more flexible about where wsgitest.py can be run from	63	else:
	64	relpath = os.path.join(self.folder, segment)
	65	return BranchesFromFileSystemServer(relpath, self.root)(
	66	environ, start_response)
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	67	else:
159.2.22 by Michael Hudson extract some methods	68	return self.app_for_branch(b, path)(environ, start_response)
159.2.22 by Michael Hudson extract some methods	69
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	70
	71	class BranchesFromFileSystemRoot(object):
	72	def __init__(self, folder):
	73	self.cache = {}
	74	self.folder = folder
	75	def __call__(self, environ, start_response):
	76	environ['loggerhead.static.url'] = environ['SCRIPT_NAME']
159.2.20 by Michael Hudson be more flexible about where wsgitest.py can be run from	77	if environ['PATH_INFO'].startswith('/static/'):
	78	segment = path_info_pop(environ)
	79	assert segment == 'static'
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	80	return static_app(environ, start_response)
	81	else:
	82	return BranchesFromFileSystemServer(
159.2.20 by Michael Hudson be more flexible about where wsgitest.py can be run from	83	'', self)(environ, start_response)
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	84
159.2.13 by Michael Hudson something like a directory listing	85	app = BranchesFromFileSystemRoot('.')
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	86
	87	app = app
	88	app = make_middleware(app)
159.2.4 by Michael Hudson more progress	89	app = make_filter(app, None)
159.2.4 by Michael Hudson more progress	90
159.2.11 by Michael Hudson hacking towards server from urls based on file paths.	91
159.2.14 by Michael Hudson oopsie	92	httpserver.serve(app, host='127.0.0.1', port='9876')
159.2.1 by Michael Hudson things start to work a little already!	93