''' Property that maps to a key in a local dict-like attribute. '''

def __init__(self, attr, key=None, read_only=False):

self.attr, self.key, self.read_only = attr, key, read_only

def __call__(self, func):

functools.update_wrapper(self, func, updated=[])

self.getter, self.key = func, self.key or func.__name__

return self

def __get__(self, obj, cls):

if obj is None: return self

key, storage = self.key, getattr(obj, self.attr)

if key not in storage: storage[key] = self.getter(obj)

return storage[key]

def __set__(self, obj, value):

if self.read_only: raise AttributeError("Read-Only property.")

getattr(obj, self.attr)[self.key] = value

def __delete__(self, obj):

if self.read_only: raise AttributeError("Read-Only property.")

''' A property that is only computed once per instance and then replaces

def __init__(self, func):

self.func = func

def __get__(self, obj, cls):

if obj is None: return self

value = obj.__dict__[self.func.__name__] = self.func(obj)

''' A property that caches itself to the class object. '''

def __init__(self, func):

functools.update_wrapper(self, func, updated=[])

self.getter = func

def __get__(self, obj, cls):

value = self.getter(cls)

setattr(cls, self.__name__, value)

""" Used to break execution and immediately finish the response """

def __init__(self, output='', status=200, header=None):

super(BottleException, self).__init__("HTTP Response %d" % status)

self.status = int(status)

self.output = output

self.headers = HeaderDict(header) if header else None

def apply(self, response):

if self.headers:

for key, value in self.headers.allitems():

response.headers[key] = value

""" Used to generate an error page """

def __init__(self, code=500, output='Unknown Error', exception=None,

traceback=None, header=None):

super(HTTPError, self).__init__(output, code, header)

self.exception = exception

self.traceback = traceback

def __repr__(self):

''' A Router is an ordered collection of route->target pairs. It is used to

default_pattern = '[^/]+'

default_filter = 're'

#: Sorry for the mess. It works. Trust me.

rule_syntax = re.compile('(\\\\*)'\

'(?:(?::([a-zA-Z_][a-zA-Z_0-9]*)?()(?:#(.*?)#)?)'\

'|(?:<([a-zA-Z_][a-zA-Z_0-9]*)?(?::([a-zA-Z_]*)'\

'(?::((?:\\\\.|[^\\\\>]+)+)?)?)?>))')

def __init__(self, strict=False):

self.rules = {} # A {rule: Rule} mapping

self.builder = {} # A rule/name->build_info mapping

self.static = {} # Cache for static routes: {path: {method: target}}

self.dynamic = [] # Cache for dynamic routes. See _compile()

#: If true, static routes are no longer checked first.

self.strict_order = strict

self.filters = {'re': self.re_filter, 'int': self.int_filter,

'float': self.float_filter, 'path': self.path_filter}

def re_filter(self, conf):

return conf or self.default_pattern, None, None

def int_filter(self, conf):

return r'-?\d+', int, lambda x: str(int(x))

def float_filter(self, conf):

return r'-?[\d.]+', float, lambda x: str(float(x))

def path_filter(self, conf):

return r'.+?', None, None

def add_filter(self, name, func):

''' Add a filter. The provided function is called with the configuration

self.filters[name] = func

def parse_rule(self, rule):

''' Parses a rule into a (name, filter, conf) token stream. If mode is

offset, prefix = 0, ''

for match in self.rule_syntax.finditer(rule):

prefix += rule[offset:match.start()]

g = match.groups()

if len(g[0])%2: # Escaped wildcard

prefix += match.group(0)[len(g[0]):]

offset = match.end()

continue

if prefix: yield prefix, None, None

name, filtr, conf = g[1:4] if not g[2] is None else g[4:7]

if not filtr: filtr = self.default_filter

yield name, filtr, conf or None

offset, prefix = match.end(), ''

if offset <= len(rule) or prefix:

yield prefix+rule[offset:], None, None

def add(self, rule, method, target, name=None):

''' Add a new route or replace the target for an existing route. '''

if rule in self.rules:

self.rules[rule][method] = target

if name: self.builder[name] = self.builder[rule]

return

target = self.rules[rule] = {method: target}

# Build pattern and other structures for dynamic routes

anons = 0 # Number of anonymous wildcards

pattern = '' # Regular expression pattern

filters = [] # Lists of wildcard input filters

builder = [] # Data structure for the URL builder

is_static = True

for key, mode, conf in self.parse_rule(rule):

if mode:

is_static = False

mask, in_filter, out_filter = self.filters[mode](conf)

if key:

pattern += '(?P<%s>%s)' % (key, mask)

else:

pattern += '(?:%s)' % mask

key = 'anon%d' % anons; anons += 1

if in_filter: filters.append((key, in_filter))

builder.append((key, out_filter or str))

elif key:

pattern += re.escape(key)

builder.append((None, key))

self.builder[rule] = builder

if name: self.builder[name] = builder

if is_static and not self.strict_order:

self.static[self.build(rule)] = target

return

def fpat_sub(m):

return m.group(0) if len(m.group(1)) % 2 else m.group(1) + '(?:'

flat_pattern = re.sub(r'(\\*)(\(\?P<[^>]*>|\((?!\?))', fpat_sub, pattern)

try:

re_match = re.compile('^(%s)$' % pattern).match

except re.error:

raise RouteSyntaxError("Could not add Route: %s (%s)" % (rule, _e()))

def match(path):

""" Return an url-argument dictionary. """

url_args = re_match(path).groupdict()

for name, wildcard_filter in filters:

try:

url_args[name] = wildcard_filter(url_args[name])

except ValueError:

raise HTTPError(400, 'Path has wrong format.')

return url_args

try:

combined = '%s|(^%s$)' % (self.dynamic[-1][0].pattern, flat_pattern)

self.dynamic[-1] = (re.compile(combined), self.dynamic[-1][1])

self.dynamic[-1][1].append((match, target))

except (AssertionError, IndexError): # AssertionError: Too many groups

self.dynamic.append((re.compile('(^%s$)' % flat_pattern),

[(match, target)]))

return match

def build(self, _name, *anons, **query):

''' Build an URL by filling the wildcards in a rule. '''

builder = self.builder.get(_name)

if not builder: raise RouteBuildError("No route with that name.", _name)

try:

for i, value in enumerate(anons): query['anon%d'%i] = value

url = ''.join([f(query.pop(n)) if n else f for (n,f) in builder])

return url if not query else url+'?'+urlencode(query)

except KeyError:

raise RouteBuildError('Missing URL argument: %r' % _e().args[0])

def match(self, environ):

''' Return a (target, url_agrs) tuple or raise HTTPError(400/404/405). '''

path, targets, urlargs = environ['PATH_INFO'] or '/', None, {}

if path in self.static:

targets = self.static[path]

else:

for combined, rules in self.dynamic:

match = combined.match(path)

if not match: continue

getargs, targets = rules[match.lastindex - 1]

urlargs = getargs(path) if getargs else {}

break

if not targets:

raise HTTPError(404, "Not found: " + repr(environ['PATH_INFO']))

method = environ['REQUEST_METHOD'].upper()

if method in targets:

return targets[method], urlargs

if method == 'HEAD' and 'GET' in targets:

return targets['GET'], urlargs

if 'ANY' in targets:

return targets['ANY'], urlargs

allowed = [verb for verb in targets if verb != 'ANY']

if 'GET' in allowed and 'HEAD' not in allowed:

allowed.append('HEAD')

raise HTTPError(405, "Method not allowed.",

''' This class wraps a route callback along with route specific metadata and

def __init__(self, app, rule, method, callback, name=None,

plugins=None, skiplist=None, **config):

#: The application this route is installed to.

self.app = app

#: The path-rule string (e.g. ``/wiki/:page``).

self.rule = rule

#: The HTTP method as a string (e.g. ``GET``).

self.method = method

#: The original callback with no plugins applied. Useful for introspection.

self.callback = callback

#: The name of the route (if specified) or ``None``.

self.name = name or None

#: A list of route-specific plugins (see :meth:`Bottle.route`).

self.plugins = plugins or []

#: A list of plugins to not apply to this route (see :meth:`Bottle.route`).

self.skiplist = skiplist or []

#: Additional keyword arguments passed to the :meth:`Bottle.route`

#: decorator are stored in this dictionary. Used for route-specific

#: plugin configuration and meta-data.

self.config = ConfigDict(config)

def __call__(self, *a, **ka):

depr("Some APIs changed to return Route() instances instead of"\

" callables. Make sure to use the Route.call method and not to"\

" call Route instances directly.")

return self.call(*a, **ka)

@cached_property

def call(self):

''' The route callback with all plugins applied. This property is

return self._make_callback()

def reset(self):

''' Forget any cached values. The next time :attr:`call` is accessed,

self.__dict__.pop('call', None)

def prepare(self):

''' Do all on-demand work immediately (useful for debugging).'''

self.call

@property

def _context(self):

depr('Switch to Plugin API v2 and access the Route object directly.')

return dict(rule=self.rule, method=self.method, callback=self.callback,

name=self.name, app=self.app, config=self.config,

apply=self.plugins, skip=self.skiplist)

def all_plugins(self):

''' Yield all Plugins affecting this route. '''

unique = set()

for p in reversed(self.app.plugins + self.plugins):

if True in self.skiplist: break

name = getattr(p, 'name', False)

if name and (name in self.skiplist or name in unique): continue

if p in self.skiplist or type(p) in self.skiplist: continue

if name: unique.add(name)

yield p

def _make_callback(self):

callback = self.callback

for plugin in self.all_plugins():

try:

if hasattr(plugin, 'apply'):

api = getattr(plugin, 'api', 1)

context = self if api > 1 else self._context

callback = plugin.apply(callback, context)

else:

callback = plugin(callback)

except RouteReset: # Try again with changed configuration.

return self._make_callback()

if not callback is self.callback:

update_wrapper(callback, self.callback)

return callback

def __repr__(self):

""" Each Bottle object represents a single, distinct web application and

def __init__(self, catchall=True, autojson=True):

#: If true, most exceptions are caught and returned as :exc:`HTTPError`

self.catchall = catchall

#: A :cls:`ResourceManager` for application files

self.resources = ResourceManager()

#: A :cls:`ConfigDict` for app specific configuration.

self.config = ConfigDict()

self.config.autojson = autojson

self.routes = [] # List of installed :class:`Route` instances.

self.router = Router() # Maps requests to :class:`Route` instances.

self.error_handler = {}

# Core plugins

self.plugins = [] # List of installed plugins.

self.hooks = HooksPlugin()

self.install(self.hooks)

if self.config.autojson:

self.install(JSONPlugin())

self.install(TemplatePlugin())

def mount(self, prefix, app, **options):

''' Mount an application (:class:`Bottle` or plain WSGI) to a specific

if isinstance(app, basestring):

prefix, app = app, prefix

depr('Parameter order of Bottle.mount() changed.') # 0.10

segments = [p for p in prefix.split('/') if p]

if not segments: raise ValueError('Empty path prefix.')

path_depth = len(segments)

def mountpoint_wrapper():

try:

request.path_shift(path_depth)

rs = BaseResponse([], 200)

def start_response(status, header):

rs.status = status

for name, value in header: rs.add_header(name, value)

return rs.body.append

body = app(request.environ, start_response)

body = itertools.chain(rs.body, body)

return HTTPResponse(body, rs.status_code, rs.headers)

finally:

request.path_shift(-path_depth)

options.setdefault('skip', True)

options.setdefault('method', 'ANY')

options.setdefault('mountpoint', {'prefix': prefix, 'target': app})

options['callback'] = mountpoint_wrapper

self.route('/%s/<:re:.*>' % '/'.join(segments), **options)

if not prefix.endswith('/'):

self.route('/' + '/'.join(segments), **options)

def merge(self, routes):

''' Merge the routes of another :class:`Bottle` application or a list of

if isinstance(routes, Bottle):

routes = routes.routes

for route in routes:

self.add_route(route)

def install(self, plugin):

''' Add a plugin to the list of plugins and prepare it for being

if hasattr(plugin, 'setup'): plugin.setup(self)

if not callable(plugin) and not hasattr(plugin, 'apply'):

raise TypeError("Plugins must be callable or implement .apply()")

self.plugins.append(plugin)

self.reset()

return plugin

def uninstall(self, plugin):

''' Uninstall plugins. Pass an instance to remove a specific plugin, a type

removed, remove = [], plugin

for i, plugin in list(enumerate(self.plugins))[::-1]:

if remove is True or remove is plugin or remove is type(plugin) \

or getattr(plugin, 'name', True) == remove:

removed.append(plugin)

del self.plugins[i]

if hasattr(plugin, 'close'): plugin.close()

if removed: self.reset()

return removed

def run(self, **kwargs):

''' Calls :func:`run` with the same parameters. '''

run(self, **kwargs)

def reset(self, route=None):

''' Reset all routes (force plugins to be re-applied) and clear all

if route is None: routes = self.routes

elif isinstance(route, Route): routes = [route]

else: routes = [self.routes[route]]

for route in routes: route.reset()

if DEBUG:

for route in routes: route.prepare()

self.hooks.trigger('app_reset')

def close(self):

''' Close the application and all installed plugins. '''

for plugin in self.plugins:

if hasattr(plugin, 'close'): plugin.close()

self.stopped = True

def match(self, environ):

""" Search for a matching route and return a (:class:`Route` , urlargs)

return self.router.match(environ)

def get_url(self, routename, **kargs):

""" Return a string that matches a named route """

scriptname = request.environ.get('SCRIPT_NAME', '').strip('/') + '/'

location = self.router.build(routename, **kargs).lstrip('/')

return urljoin(urljoin('/', scriptname), location)

def add_route(self, route):

''' Add a route object, but do not change the :data:`Route.app`

self.routes.append(route)

self.router.add(route.rule, route.method, route, name=route.name)

if DEBUG: route.prepare()

def route(self, path=None, method='GET', callback=None, name=None,

apply=None, skip=None, **config):

""" A decorator to bind a function to a request URL. Example::

if callable(path): path, callback = None, path

plugins = makelist(apply)

skiplist = makelist(skip)

def decorator(callback):

# TODO: Documentation and tests

if isinstance(callback, basestring): callback = load(callback)

for rule in makelist(path) or yieldroutes(callback):

for verb in makelist(method):

verb = verb.upper()

route = Route(self, rule, verb, callback, name=name,

plugins=plugins, skiplist=skiplist, **config)

self.add_route(route)

return callback

return decorator(callback) if callback else decorator

def get(self, path=None, method='GET', **options):

""" Equals :meth:`route`. """

return self.route(path, method, **options)

def post(self, path=None, method='POST', **options):

""" Equals :meth:`route` with a ``POST`` method parameter. """

return self.route(path, method, **options)

def put(self, path=None, method='PUT', **options):

""" Equals :meth:`route` with a ``PUT`` method parameter. """

return self.route(path, method, **options)

def delete(self, path=None, method='DELETE', **options):

""" Equals :meth:`route` with a ``DELETE`` method parameter. """

return self.route(path, method, **options)

def error(self, code=500):

""" Decorator: Register an output handler for a HTTP error code"""

def wrapper(handler):

self.error_handler[int(code)] = handler

return handler

return wrapper

def hook(self, name):

""" Return a decorator that attaches a callback to a hook. Three hooks

def wrapper(func):

self.hooks.add(name, func)

return func

return wrapper

def handle(self, path, method='GET'):

""" (deprecated) Execute the first matching route callback and return

depr("This method will change semantics in 0.10. Try to avoid it.")

if isinstance(path, dict):

return self._handle(path)

return self._handle({'PATH_INFO': path, 'REQUEST_METHOD': method.upper()})

def _handle(self, environ):

try:

environ['bottle.app'] = self

request.bind(environ)

response.bind()

route, args = self.router.match(environ)

environ['route.handle'] = environ['bottle.route'] = route

environ['route.url_args'] = args

return route.call(**args)

except HTTPResponse:

return _e()

except RouteReset:

route.reset()

return self._handle(environ)

except (KeyboardInterrupt, SystemExit, MemoryError):

raise

except Exception:

if not self.catchall: raise

stacktrace = format_exc()

environ['wsgi.errors'].write(stacktrace)

return HTTPError(500, "Internal Server Error", _e(), stacktrace)

def _cast(self, out, peek=None):

""" Try to convert the parameter into something WSGI compatible and set

# Empty output is done here

if not out:

response['Content-Length'] = 0

return []

# Join lists of byte or unicode strings. Mixed lists are NOT supported

if isinstance(out, (tuple, list))\

and isinstance(out[0], (bytes, unicode)):

out = out[0][0:0].join(out) # b'abc'[0:0] -> b''

# Encode unicode strings

if isinstance(out, unicode):

out = out.encode(response.charset)

# Byte Strings are just returned

if isinstance(out, bytes):

response['Content-Length'] = len(out)

return [out]

# HTTPError or HTTPException (recursive, because they may wrap anything)

# TODO: Handle these explicitly in handle() or make them iterable.

if isinstance(out, HTTPError):

out.apply(response)

out = self.error_handler.get(out.status, repr)(out)

if isinstance(out, HTTPResponse):

depr('Error handlers must not return :exc:`HTTPResponse`.') #0.9

return self._cast(out)

if isinstance(out, HTTPResponse):

out.apply(response)

return self._cast(out.output)

# File-like objects.

if hasattr(out, 'read'):

if 'wsgi.file_wrapper' in request.environ:

return request.environ['wsgi.file_wrapper'](out)

elif hasattr(out, 'close') or not hasattr(out, '__iter__'):

return WSGIFileWrapper(out)

# Handle Iterables. We peek into them to detect their inner type.

try:

out = iter(out)

first = next(out)

while not first:

first = next(out)

except StopIteration:

return self._cast('')

except HTTPResponse:

first = _e()

except (KeyboardInterrupt, SystemExit, MemoryError):

raise

except Exception:

if not self.catchall: raise

first = HTTPError(500, 'Unhandled exception', _e(), format_exc())

# These are the inner types allowed in iterator or generator objects.

if isinstance(first, HTTPResponse):

return self._cast(first)

if isinstance(first, bytes):

return itertools.chain([first], out)

if isinstance(first, unicode):

return imap(lambda x: x.encode(response.charset),

itertools.chain([first], out))

return self._cast(HTTPError(500, 'Unsupported response type: %s'\

% type(first)))

def wsgi(self, environ, start_response):

""" The bottle WSGI-interface. """

try:

out = self._cast(self._handle(environ))

# rfc2616 section 4.3

if response._status_code in (100, 101, 204, 304)\

or request.method == 'HEAD':

if hasattr(out, 'close'): out.close()

out = []

if isinstance(response._status_line, unicode):

response._status_line = str(response._status_line)

start_response(response._status_line, list(response.iter_headers()))

return out

except (KeyboardInterrupt, SystemExit, MemoryError):

raise

except Exception:

if not self.catchall: raise

err = '<h1>Critical error while processing request: %s</h1>' \

% html_escape(environ.get('PATH_INFO', '/'))

if DEBUG:

err += '<h2>Error:</h2>\n<pre>\n%s\n</pre>\n' \

'<h2>Traceback:</h2>\n<pre>\n%s\n</pre>\n' \

% (html_escape(repr(_e())), html_escape(format_exc()))

environ['wsgi.errors'].write(err)

headers = [('Content-Type', 'text/html; charset=UTF-8')]

start_response('500 INTERNAL SERVER ERROR', headers)

return [tob(err)]

def __call__(self, environ, start_response):

''' Each instance of :class:'Bottle' is a WSGI application. '''

""" A wrapper for WSGI environment dictionaries that adds a lot of

__slots__ = ('environ')

#: Maximum size of memory buffer for :attr:`body` in bytes.

MEMFILE_MAX = 102400

#: Maximum number pr GET or POST parameters per request

MAX_PARAMS = 100

def __init__(self, environ=None):

""" Wrap a WSGI environ dictionary. """

#: The wrapped WSGI environ dictionary. This is the only real attribute.

#: All other attributes actually are read-only properties.

self.environ = {} if environ is None else environ

self.environ['bottle.request'] = self

@DictProperty('environ', 'bottle.app', read_only=True)

def app(self):

''' Bottle application handling this request. '''

raise RuntimeError('This request is not connected to an application.')

@property

def path(self):

''' The value of ``PATH_INFO`` with exactly one prefixed slash (to fix

return '/' + self.environ.get('PATH_INFO','').lstrip('/')

@property

def method(self):

''' The ``REQUEST_METHOD`` value as an uppercase string. '''

return self.environ.get('REQUEST_METHOD', 'GET').upper()

@DictProperty('environ', 'bottle.request.headers', read_only=True)

def headers(self):

''' A :class:`WSGIHeaderDict` that provides case-insensitive access to

return WSGIHeaderDict(self.environ)

def get_header(self, name, default=None):

''' Return the value of a request header, or a given default value. '''

return self.headers.get(name, default)

@DictProperty('environ', 'bottle.request.cookies', read_only=True)

def cookies(self):

""" Cookies parsed into a :class:`FormsDict`. Signed cookies are NOT

cookies = SimpleCookie(self.environ.get('HTTP_COOKIE',''))

cookies = list(cookies.values())[:self.MAX_PARAMS]

return FormsDict((c.key, c.value) for c in cookies)

def get_cookie(self, key, default=None, secret=None):

""" Return the content of a cookie. To read a `Signed Cookie`, the

value = self.cookies.get(key)

if secret and value:

dec = cookie_decode(value, secret) # (key, value) tuple or None

return dec[1] if dec and dec[0] == key else default

return value or default

@DictProperty('environ', 'bottle.request.query', read_only=True)

def query(self):

''' The :attr:`query_string` parsed into a :class:`FormsDict`. These

get = self.environ['bottle.get'] = FormsDict()

pairs = _parse_qsl(self.environ.get('QUERY_STRING', ''))

for key, value in pairs[:self.MAX_PARAMS]:

get[key] = value

return get

@DictProperty('environ', 'bottle.request.forms', read_only=True)

def forms(self):

""" Form values parsed from an `url-encoded` or `multipart/form-data`

forms = FormsDict()

for name, item in self.POST.allitems():

if not hasattr(item, 'filename'):

forms[name] = item