#!/usr/bin/env python# -*- coding: utf-8 -*-"""Bottle is a fast and simple micro-framework for small web applications. Itoffers request dispatching (Routes) with URL parameter support, templates,a built-in HTTP Server and adapters for many third party WSGI/HTTP-server andtemplate engines - all in a single file and with no dependencies other than thePython Standard Library.Homepage and documentation: http://bottlepy.org/Copyright (c) 2009-2025, Marcel Hellkamp.License: MIT (see LICENSE for details)"""importsys__author__='Marcel Hellkamp'__version__='0.14-dev'__license__='MIT'################################################################################ Command-line interface ###################################################################################################################################### INFO: Some server adapters need to monkey-patch std-lib modules before they# are imported. This is why some of the command-line handling is done here, but# the actual call to _main() is at the end of the file.def_cli_parse(args):# pragma: no coveragefromargparseimportArgumentParserparser=ArgumentParser(prog=args[0],usage="%(prog)s [options] package.module:app")opt=parser.add_argumentopt("--version",action="store_true",help="show version number.")opt("-b","--bind",metavar="ADDRESS",help="bind socket to ADDRESS.")opt("-s","--server",default='wsgiref',help="use SERVER as backend.")opt("-p","--plugin",action="append",help="install additional plugin/s.")opt("-c","--conf",action="append",metavar="FILE",help="load config values from FILE.")opt("-C","--param",action="append",metavar="NAME=VALUE",help="override config values.")opt("--debug",action="store_true",help="start server in debug mode.")opt("--reload",action="store_true",help="auto-reload on file changes.")opt('app',help='WSGI app entry point.',nargs='?')cli_args=parser.parse_args(args[1:])returncli_args,parserdef_cli_patch(cli_args):# pragma: no coverageparsed_args,_=_cli_parse(cli_args)opts=parsed_argsifopts.server:ifopts.server.startswith('gevent'):importgevent.monkeygevent.monkey.patch_all()elifopts.server.startswith('eventlet'):importeventleteventlet.monkey_patch()if__name__=='__main__':_cli_patch(sys.argv)################################################################################ Imports and Helpers used everywhere else ####################################################################################################################importbase64,calendar,email.utils,functools,hmac,itertools,\mimetypes,os,re,tempfile,threading,time,warnings,weakref,hashlibfromtypesimportFunctionTypefromdatetimeimportdateasdatedate,datetime,timedeltafromtempfileimportNamedTemporaryFilefromtracebackimportformat_exc,print_excfromunicodedataimportnormalizetry:fromujsonimportdumpsasjson_dumps,loadsasjson_ldsexceptImportError:fromjsonimportdumpsasjson_dumps,loadsasjson_ldspy=sys.version_infoimporthttp.clientashttplibimport_threadasthreadfromurllib.parseimporturljoin,SplitResultasUrlSplitResultfromurllib.parseimporturlencode,quoteasurlquote,unquoteasurlunquotefromhttp.cookiesimportSimpleCookie,Morsel,CookieErrorfromcollections.abcimportMutableMappingasDictMixinfromtypesimportModuleTypeasnew_moduleimportpicklefromioimportBytesIOimportconfigparserfromdatetimeimporttimezoneUTC=timezone.utcimportinspectjson_loads=lambdas:json_lds(touni(s))callable=lambdax:hasattr(x,'__call__')_UNSET=object()def_wsgi_recode(src):""" Translate a PEP-3333 latin1-string to utf8+surrogateescape """ifsrc.isascii():returnsrcreturnsrc.encode('latin1').decode('utf8','surrogateescape')def_raise(*a):raisea[0](a[1]).with_traceback(a[2])# Some helpers for string/byte handlingdeftob(s,enc='utf8'):ifisinstance(s,str):returns.encode(enc)returnb''ifsisNoneelsebytes(s)deftouni(s,enc='utf8',err='strict'):ifisinstance(s,(bytes,bytearray)):returnstr(s,enc,err)return""ifsisNoneelsestr(s)def_stderr(*args):try:print(*args,file=sys.stderr)except(IOError,AttributeError):pass# Some environments do not allow printing (mod_wsgi)# A bug in functools causes it to break if the wrapper is an instance methoddefupdate_wrapper(wrapper,wrapped,*a,**ka):try:functools.update_wrapper(wrapper,wrapped,*a,**ka)exceptAttributeError:pass# These helpers are used at module level and need to be defined first.# And yes, I know PEP-8, but sometimes a lower-case classname makes more sense.defdepr(major,minor,cause,fix,stacklevel=3):text="Use of feature or API deprecated since Bottle-%d.%d\n"\"Cause:%s\n"\"Fix:%s\n"%(major,minor,cause,fix)ifDEBUG=='strict':raiseDeprecationWarning(text)warnings.warn(text,DeprecationWarning,stacklevel=stacklevel)returnDeprecationWarning(text)defmakelist(data):# This is just too handyifisinstance(data,(tuple,list,set,dict)):returnlist(data)elifdata:return[data]else:return[][docs]classDictProperty:""" Property that maps to a key in a local dict-like attribute. """[docs]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.keyorfunc.__name__returnselfdef__get__(self,obj,cls):ifobjisNone:returnselfkey,storage=self.key,getattr(obj,self.attr)ifkeynotinstorage:storage[key]=self.getter(obj)returnstorage[key]def__set__(self,obj,value):ifself.read_only:raiseAttributeError("Read-Only property.")getattr(obj,self.attr)[self.key]=valuedef__delete__(self,obj):ifself.read_only:raiseAttributeError("Read-Only property.")delgetattr(obj,self.attr)[self.key] [docs]classcached_property:""" A property that is only computed once per instance and then replaces itself with an ordinary attribute. Deleting the attribute resets the property. """[docs]def__init__(self,func):update_wrapper(self,func)self.func=func def__get__(self,obj,cls):ifobjisNone:returnselfvalue=obj.__dict__[self.func.__name__]=self.func(obj)returnvalue [docs]classlazy_attribute:""" A property that caches itself to the class object. """[docs]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)returnvalue ################################################################################ Exceptions and Events ######################################################################################################################################[docs]classBottleException(Exception):""" A base class for exceptions used by bottle. """pass ################################################################################ Routing #####################################################################################################################################################classRouteError(BottleException):""" This is a base class for all routing related exceptions """classRouterUnknownModeError(RouteError):passclassRouteSyntaxError(RouteError):""" The route parser found something not supported by this router. """classRouteBuildError(RouteError):""" The route could not be built. """def_re_flatten(p):""" Turn all capturing groups in a regular expression pattern into non-capturing groups. """if'('notinp:returnpreturnre.sub(r'(\\*)(\(\?P<[^>]+>|\((?!\?))',lambdam:m.group(0)iflen(m.group(1))%2elsem.group(1)+'(?:',p)[docs]classRouter:""" A Router is an ordered collection of route->target pairs. It is used to efficiently match WSGI requests against a number of routes and return the first target that satisfies the request. The target may be anything, usually a string, ID or callable object. A route consists of a path-rule and a HTTP method. The path-rule is either a static path (e.g. `/contact`) or a dynamic path that contains wildcards (e.g. `/wiki/<page>`). The wildcard syntax and details on the matching order are described in docs:`routing`. """default_pattern='[^/]+'default_filter='re'#: The current CPython regexp implementation does not allow more#: than 99 matching groups per regular expression._MAX_GROUPS_PER_PATTERN=99[docs]def__init__(self,strict=False):self.rules=[]# All rules in orderself._groups={}# index of regexes to find them in dyna_routesself.builder={}# Data structure for the url builderself.static={}# Search structure for static routesself.dyna_routes={}self.dyna_regexes={}# Search structure for dynamic routes#: If true, static routes are no longer checked first.self.strict_order=strictself.filters={'re':lambdaconf:(_re_flatten(conforself.default_pattern),None,None),'int':lambdaconf:(r'-?\d+',int,lambdax:str(int(x))),'float':lambdaconf:(r'-?[\d.]+',float,lambdax:str(float(x))),'path':lambdaconf:(r'.+?',None,None)} [docs]defadd_filter(self,name,func):""" Add a filter. The provided function is called with the configuration string as parameter and must return a (regexp, to_python, to_url) tuple. The first element is a string, the last two are callables or None. """self.filters[name]=func rule_syntax=re.compile('(\\\\*)''(?:(?::([a-zA-Z_][a-zA-Z_0-9]*)?()(?:#(.*?)#)?)''|(?:<([a-zA-Z_][a-zA-Z_0-9]*)?(?::([a-zA-Z_]*)''(?::((?:\\\\.|[^\\\\>])+)?)?)?>))')def_itertokens(self,rule):offset,prefix=0,''formatchinself.rule_syntax.finditer(rule):prefix+=rule[offset:match.start()]g=match.groups()ifg[2]isnotNone:depr(0,13,"Use of old route syntax.","Use <name> instead of :name in routes.",stacklevel=4)iflen(g[0])%2:# Escaped wildcardprefix+=match.group(0)[len(g[0]):]offset=match.end()continueifprefix:yieldprefix,None,Nonename,filtr,conf=g[4:7]ifg[2]isNoneelseg[1:4]yieldname,filtror'default',conforNoneoffset,prefix=match.end(),''ifoffset<=len(rule)orprefix:yieldprefix+rule[offset:],None,None[docs]defadd(self,rule,method,target,name=None):""" Add a new rule or replace the target for an existing rule. """anons=0# Number of anonymous wildcards foundkeys=[]# Names of keyspattern=''# Regular expression pattern with named groupsfilters=[]# Lists of wildcard input filtersbuilder=[]# Data structure for the URL builderis_static=Trueforkey,mode,confinself._itertokens(rule):ifmode:is_static=Falseifmode=='default':mode=self.default_filtermask,in_filter,out_filter=self.filters[mode](conf)ifnotkey:pattern+='(?:%s)'%maskkey='anon%d'%anonsanons+=1else:pattern+='(?P<%s>%s)'%(key,mask)keys.append(key)ifin_filter:filters.append((key,in_filter))builder.append((key,out_filterorstr))elifkey:pattern+=re.escape(key)builder.append((None,key))self.builder[rule]=builderifname:self.builder[name]=builderifis_staticandnotself.strict_order:self.static.setdefault(method,{})self.static[method][self.build(rule)]=(target,None)returntry:re_pattern=re.compile('^(%s)$'%pattern)re_match=re_pattern.matchexceptre.errorase:raiseRouteSyntaxError("Could not add Route:%s (%s)"%(rule,e))iffilters:defgetargs(path):url_args=re_match(path).groupdict()forname,wildcard_filterinfilters:try:url_args[name]=wildcard_filter(url_args[name])exceptValueError:raiseHTTPError(400,'Path has wrong format.')returnurl_argselifre_pattern.groupindex:defgetargs(path):returnre_match(path).groupdict()else:getargs=Noneflatpat=_re_flatten(pattern)whole_rule=(rule,flatpat,target,getargs)if(flatpat,method)inself._groups:ifDEBUG:msg='Route <%s%s> overwrites a previously defined route'warnings.warn(msg%(method,rule),RuntimeWarning,stacklevel=3)self.dyna_routes[method][self._groups[flatpat,method]]=whole_ruleelse:self.dyna_routes.setdefault(method,[]).append(whole_rule)self._groups[flatpat,method]=len(self.dyna_routes[method])-1self._compile(method) def_compile(self,method):all_rules=self.dyna_routes[method]comborules=self.dyna_regexes[method]=[]maxgroups=self._MAX_GROUPS_PER_PATTERNforxinrange(0,len(all_rules),maxgroups):some=all_rules[x:x+maxgroups]combined=(flatpatfor(_,flatpat,_,_)insome)combined='|'.join('(^%s$)'%flatpatforflatpatincombined)combined=re.compile(combined).matchrules=[(target,getargs)for(_,_,target,getargs)insome]comborules.append((combined,rules))[docs]defbuild(self,_name,*anons,**query):""" Build an URL by filling the wildcards in a rule. """builder=self.builder.get(_name)ifnotbuilder:raiseRouteBuildError("No route with that name.",_name)try:fori,valueinenumerate(anons):query['anon%d'%i]=valueurl=''.join([f(query.pop(n))ifnelseffor(n,f)inbuilder])returnurlifnotqueryelseurl+'?'+urlencode(query,doseq=True)exceptKeyErrorasE:raiseRouteBuildError('Missing URL argument:%r'%E.args[0]) [docs]defmatch(self,environ):""" Return a (target, url_args) tuple or raise HTTPError(400/404/405). """verb=environ['REQUEST_METHOD'].upper()path=environ['PATH_INFO']or'/'methods=('PROXY','HEAD','GET','ANY')ifverb=='HEAD'else('PROXY',verb,'ANY')formethodinmethods:ifmethodinself.staticandpathinself.static[method]:target,getargs=self.static[method][path]returntarget,getargs(path)ifgetargselse{}elifmethodinself.dyna_regexes:forcombined,rulesinself.dyna_regexes[method]:match=combined(path)ifmatch:target,getargs=rules[match.lastindex-1]returntarget,getargs(path)ifgetargselse{}# No matching route found. Collect alternative methods for 405 responseallowed=set([])nocheck=set(methods)formethodinset(self.static)-nocheck:ifpathinself.static[method]:allowed.add(method)formethodinset(self.dyna_regexes)-allowed-nocheck:forcombined,rulesinself.dyna_regexes[method]:match=combined(path)ifmatch:allowed.add(method)ifallowed:allow_header=",".join(sorted(allowed))raiseHTTPError(405,"Method not allowed.",Allow=allow_header)# No matching route and no alternative method found. We give upraiseHTTPError(404,"Not found: "+repr(path)) [docs]classRoute:""" This class wraps a route callback along with route specific metadata and configuration and applies Plugins on demand. It is also responsible for turning an URL path rule into a regular expression usable by the Router. """[docs]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=nameorNone#: A list of route-specific plugins (see :meth:`Bottle.route`).self.plugins=pluginsor[]#: A list of plugins to not apply to this route (see :meth:`Bottle.route`).self.skiplist=skiplistor[]#: 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=app.config._make_overlay()self.config.load_dict(config) [docs]@cached_propertydefcall(self):""" The route callback with all plugins applied. This property is created on demand and then cached to speed up subsequent requests."""returnself._make_callback() [docs]defreset(self):""" Forget any cached values. The next time :attr:`call` is accessed, all plugins are re-applied. """self.__dict__.pop('call',None) [docs]defprepare(self):""" Do all on-demand work immediately (useful for debugging)."""self.call [docs]defall_plugins(self):""" Yield all Plugins affecting this route. """unique=set()forpinreversed(self.app.plugins+self.plugins):ifTrueinself.skiplist:breakname=getattr(p,'name',False)ifnameand(nameinself.skiplistornameinunique):continueifpinself.skiplistortype(p)inself.skiplist:continueifname:unique.add(name)yieldp def_make_callback(self):callback=self.callbackforplugininself.all_plugins():ifhasattr(plugin,'apply'):callback=plugin.apply(callback,self)else:callback=plugin(callback)ifcallbackisnotself.callback:update_wrapper(callback,self.callback)returncallback[docs]defget_undecorated_callback(self):""" Return the callback. If the callback is a decorated function, try to recover the original function. """func=self.callbackwhileTrue:ifgetattr(func,'__wrapped__',False):func=func.__wrapped__elifgetattr(func,'__func__',False):func=func.__func__elifgetattr(func,'__closure__',False):depr(0,14,"Decorated callback without __wrapped__","When applying decorators to route callbacks, make sure"" the decorator uses @functools.wraps or update_wrapper."" This warning may also trigger if you reference callables"" from a nonlocal scope.")cells_values=(cell.cell_contentsforcellinfunc.__closure__)isfunc=lambdax:isinstance(x,FunctionType)orhasattr(x,'__call__')func=next(filter(isfunc,cells_values),func)else:returnfunc [docs]defget_callback_args(self):""" Return a list of argument names the callback (most likely) accepts as keyword arguments. If the callback is a decorated function, try to recover the original function before inspection. """sig=inspect.signature(self.get_undecorated_callback())return[p.nameforpinsig.parameters.values()ifp.kindin(p.POSITIONAL_OR_KEYWORD,p.KEYWORD_ONLY)] [docs]defget_config(self,key,default=None):""" Lookup a config field and return its value, first checking the route.config, then route.app.config."""depr(0,13,"Route.get_config() is deprecated.","The Route.config property already includes values from the"" application config for missing keys. Access it directly.")returnself.config.get(key,default) def__repr__(self):cb=self.get_undecorated_callback()return'<%s%s ->%s:%s>'%(self.method,self.rule,cb.__module__,getattr(cb,'__name__','?')) ################################################################################ Application Object ##########################################################################################################################################[docs]classBottle:""" Each Bottle object represents a single, distinct web application and consists of routes, callbacks, plugins, resources and configuration. Instances are callable WSGI applications. :param catchall: If true (default), handle all exceptions. Turn off to let debugging middleware handle exceptions. """@lazy_attributedef_global_config(cls):cfg=ConfigDict()cfg.meta_set('catchall','validate',bool)returncfg[docs]def__init__(self,**kwargs):#: A :class:`ConfigDict` for app specific configuration.self.config=self._global_config._make_overlay()self.config._add_change_listener(functools.partial(self.trigger_hook,'config'))self.config.update({"catchall":True})ifkwargs.get('catchall')isFalse:depr(0,13,"Bottle(catchall) keyword argument.","The 'catchall' setting is now part of the app ""configuration. Fix: `app.config['catchall'] = False`")self.config['catchall']=Falseifkwargs.get('autojson')isFalse:depr(0,13,"Bottle(autojson) keyword argument.","The 'autojson' setting is now part of the app ""configuration. Fix: `app.config['json.enable'] = False`")self.config['json.enable']=Falseself._mounts=[]#: A :class:`ResourceManager` for application filesself.resources=ResourceManager()self.routes=[]# List of installed :class:`Route` instances.self.router=Router()# Maps requests to :class:`Route` instances.self.error_handler={}# Core pluginsself.plugins=[]# List of installed plugins.self.install(JSONPlugin())self.install(TemplatePlugin()) #: If true, most exceptions are caught and returned as :exc:`HTTPError`catchall=DictProperty('config','catchall')__hook_names='before_request','after_request','app_reset','config'__hook_reversed={'after_request'}@cached_propertydef_hooks(self):returndict((name,[])fornameinself.__hook_names)[docs]defadd_hook(self,name,func):""" Attach a callback to a hook. Three hooks are currently implemented: before_request Executed once before each request. The request context is available, but no routing has happened yet. after_request Executed once after each request regardless of its outcome. app_reset Called whenever :meth:`Bottle.reset` is called. """ifnameinself.__hook_reversed:self._hooks[name].insert(0,func)else:self._hooks[name].append(func) [docs]defremove_hook(self,name,func):""" Remove a callback from a hook. """ifnameinself._hooksandfuncinself._hooks[name]:self._hooks[name].remove(func)returnTrue [docs]deftrigger_hook(self,__name,*args,**kwargs):""" Trigger a hook and return a list of results. """return[hook(*args,**kwargs)forhookinself._hooks[__name][:]] [docs]defhook(self,name):""" Return a decorator that attaches a callback to a hook. See :meth:`add_hook` for details."""defdecorator(func):self.add_hook(name,func)returnfuncreturndecorator def_mount_wsgi(self,prefix,app,**options):segments=[pforpinprefix.split('/')ifp]ifnotsegments:raiseValueError('WSGI applications cannot be mounted to "/".')path_depth=len(segments)defmountpoint_wrapper():try:request.path_shift(path_depth)rs=HTTPResponse([])defstart_response(status,headerlist,exc_info=None):ifexc_info:_raise(*exc_info)status=_wsgi_recode(status)headerlist=[(k,_wsgi_recode(v))for(k,v)inheaderlist]rs.status=statusforname,valueinheaderlist:rs.add_header(name,value)returnrs.body.appendbody=app(request.environ,start_response)rs.body=itertools.chain(rs.body,body)ifrs.bodyelsebodyreturnrsfinally:request.path_shift(-path_depth)options.setdefault('skip',True)options.setdefault('method','PROXY')options.setdefault('mountpoint',{'prefix':prefix,'target':app})options['callback']=mountpoint_wrapperself.route('/%s/<:re:.*>'%'/'.join(segments),**options)ifnotprefix.endswith('/'):self.route('/'+'/'.join(segments),**options)def_mount_app(self,prefix,app,**options):ifappinself._mountsor'_mount.app'inapp.config:depr(0,13,"Application mounted multiple times. Falling back to WSGI mount.","Clone application before mounting to a different location.")returnself._mount_wsgi(prefix,app,**options)ifoptions:depr(0,13,"Unsupported mount options. Falling back to WSGI mount.","Do not specify any route options when mounting bottle application.")returnself._mount_wsgi(prefix,app,**options)ifnotprefix.endswith("/"):depr(0,13,"Prefix must end in '/'. Falling back to WSGI mount.","Consider adding an explicit redirect from '/prefix' to '/prefix/' in the parent application.")returnself._mount_wsgi(prefix,app,**options)self._mounts.append(app)app.config['_mount.prefix']=prefixapp.config['_mount.app']=selfforrouteinapp.routes:route.rule=prefix+route.rule.lstrip('/')self.add_route(route)[docs]defmount(self,prefix,app,**options):""" Mount an application (:class:`Bottle` or plain WSGI) to a specific URL prefix. Example:: parent_app.mount('/prefix/', child_app) :param prefix: path prefix or `mount-point`. :param app: an instance of :class:`Bottle` or a WSGI application. Plugins from the parent application are not applied to the routes of the mounted child application. If you need plugins in the child application, install them separately. While it is possible to use path wildcards within the prefix path (:class:`Bottle` childs only), it is highly discouraged. The prefix path must end with a slash. If you want to access the root of the child application via `/prefix` in addition to `/prefix/`, consider adding a route with a 307 redirect to the parent application. """ifnotprefix.startswith('/'):raiseValueError("Prefix must start with '/'")ifisinstance(app,Bottle):returnself._mount_app(prefix,app,**options)else:returnself._mount_wsgi(prefix,app,**options) [docs]defmerge(self,routes):""" Merge the routes of another :class:`Bottle` application or a list of :class:`Route` objects into this application. The routes keep their 'owner', meaning that the :data:`Route.app` attribute is not changed. """ifisinstance(routes,Bottle):routes=routes.routesforrouteinroutes:self.add_route(route) [docs]definstall(self,plugin):""" Add a plugin to the list of plugins and prepare it for being applied to all routes of this application. A plugin may be a simple decorator or an object that implements the :class:`Plugin` API. """ifhasattr(plugin,'setup'):plugin.setup(self)ifnotcallable(plugin)andnothasattr(plugin,'apply'):raiseTypeError("Plugins must be callable or implement .apply()")self.plugins.append(plugin)self.reset()returnplugin [docs]defuninstall(self,plugin):""" Uninstall plugins. Pass an instance to remove a specific plugin, a type object to remove all plugins that match that type, a string to remove all plugins with a matching ``name`` attribute or ``True`` to remove all plugins. Return the list of removed plugins. """removed,remove=[],pluginfori,plugininlist(enumerate(self.plugins))[::-1]:ifremoveisTrueorremoveispluginorremoveistype(plugin) \
orgetattr(plugin,'name',True)==remove:removed.append(plugin)delself.plugins[i]ifhasattr(plugin,'close'):plugin.close()ifremoved:self.reset()returnremoved [docs]defreset(self,route=None):""" Reset all routes (force plugins to be re-applied) and clear all caches. If an ID or route object is given, only that specific route is affected. """ifrouteisNone:routes=self.routeselifisinstance(route,Route):routes=[route]else:routes=[self.routes[route]]forrouteinroutes:route.reset()ifDEBUG:forrouteinroutes:route.prepare()self.trigger_hook('app_reset') [docs]defclose(self):""" Close the application and all installed plugins. """forplugininself.plugins:ifhasattr(plugin,'close'):plugin.close() [docs]defrun(self,**kwargs):""" Calls :func:`run` with the same parameters. """run(self,**kwargs) [docs]defmatch(self,environ):""" Search for a matching route and return a (:class:`Route`, urlargs) tuple. The second value is a dictionary with parameters extracted from the URL. Raise :exc:`HTTPError` (404/405) on a non-match."""returnself.router.match(environ) [docs]defget_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('/')returnurljoin(urljoin('/',scriptname),location) [docs]defadd_route(self,route):""" Add a route object, but do not change the :data:`Route.app` attribute."""self.routes.append(route)self.router.add(route.rule,route.method,route,name=route.name)ifDEBUG:route.prepare() [docs]defroute(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:: @app.route('/hello/<name>') def hello(name): return 'Hello %s' % name The ``<name>`` part is a wildcard. See :class:`Router` for syntax details. :param path: Request path or a list of paths to listen to. If no path is specified, it is automatically generated from the signature of the function. :param method: HTTP method (`GET`, `POST`, `PUT`, ...) or a list of methods to listen to. (default: `GET`) :param callback: An optional shortcut to avoid the decorator syntax. ``route(..., callback=func)`` equals ``route(...)(func)`` :param name: The name for this route. (default: None) :param apply: A decorator or plugin or a list of plugins. These are applied to the route callback in addition to installed plugins. :param skip: A list of plugins, plugin classes or names. Matching plugins are not installed to this route. ``True`` skips all. Any additional keyword arguments are stored as route-specific configuration and passed to plugins (see :meth:`Plugin.apply`). """ifcallable(path):path,callback=None,pathplugins=makelist(apply)skiplist=makelist(skip)defdecorator(callback):ifisinstance(callback,str):callback=load(callback)# type: Callableforruleinmakelist(path)oryieldroutes(callback):forverbinmakelist(method):verb=verb.upper()route=Route(self,rule,verb,callback,name=name,plugins=plugins,skiplist=skiplist,**config)self.add_route(route)returncallbackreturndecorator(callback)ifcallbackelsedecorator [docs]defget(self,path=None,method='GET',**options):""" Equals :meth:`route`. """returnself.route(path,method,**options) [docs]defpost(self,path=None,method='POST',**options):""" Equals :meth:`route` with a ``POST`` method parameter. """returnself.route(path,method,**options) [docs]defput(self,path=None,method='PUT',**options):""" Equals :meth:`route` with a ``PUT`` method parameter. """returnself.route(path,method,**options) [docs]defdelete(self,path=None,method='DELETE',**options):""" Equals :meth:`route` with a ``DELETE`` method parameter. """returnself.route(path,method,**options) [docs]defpatch(self,path=None,method='PATCH',**options):""" Equals :meth:`route` with a ``PATCH`` method parameter. """returnself.route(path,method,**options) [docs]deferror(self,code=500,callback=None):""" Register an output handler for a HTTP error code. Can be used as a decorator or called directly :: def error_handler_500(error): return 'error_handler_500' app.error(code=500, callback=error_handler_500) @app.error(404) def error_handler_404(error): return 'error_handler_404' """defdecorator(callback):ifisinstance(callback,str):callback=load(callback)self.error_handler[int(code)]=callbackreturncallbackreturndecorator(callback)ifcallbackelsedecorator defdefault_error_handler(self,res):returntob(template(ERROR_PAGE_TEMPLATE,e=res,template_settings=dict(name='__ERROR_PAGE_TEMPLATE')))def_handle(self,environ):path=environ['bottle.raw_path']=environ['PATH_INFO']environ['PATH_INFO']=_wsgi_recode(path)environ['bottle.app']=selfrequest.bind(environ)response.bind()out=Nonetry:try:self.trigger_hook('before_request')route,args=self.router.match(environ)environ['route.handle']=routeenviron['bottle.route']=routeenviron['route.url_args']=argsout=route.call(**args)exceptHTTPResponseasE:out=Efinally:ifisinstance(out,HTTPResponse):out.apply(response)try:self.trigger_hook('after_request')exceptHTTPResponseasE:out=Eout.apply(response)except(KeyboardInterrupt,SystemExit,MemoryError):raiseexceptExceptionasE:_try_close(out)ifnotself.catchall:raisestacktrace=format_exc()environ['wsgi.errors'].write(stacktrace)environ['wsgi.errors'].flush()environ['bottle.exc_info']=sys.exc_info()out=HTTPError(500,"Internal Server Error",E,stacktrace)out.apply(response)returnoutdef_cast(self,out,peek=None):""" Try to convert the parameter into something WSGI compatible and set correct HTTP headers when possible. Support: False, bytes/bytearray, str, dict, HTTPResponse, HTTPError, file-like, iterable of bytes/bytearray or str instances. """# Empty output is done hereifnotout:if'Content-Length'notinresponse:response['Content-Length']=0return[]# Join lists of byte or unicode strings. Mixed lists are NOT supportedifisinstance(out,(tuple,list))andisinstance(out[0],(bytes,str)):out=out[0][0:0].join(out)# b'abc'[0:0] -> b''# Encode unicode stringsifisinstance(out,str):out=out.encode(response.charset)# Byte Strings are just returnedifisinstance(out,bytes):if'Content-Length'notinresponse: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.ifisinstance(out,HTTPError):out.apply(response)out=self.error_handler.get(out.status_code,self.default_error_handler)(out)returnself._cast(out)ifisinstance(out,HTTPResponse):out.apply(response)returnself._cast(out.body)# File-like objects.ifhasattr(out,'read'):if'wsgi.file_wrapper'inrequest.environ:returnrequest.environ['wsgi.file_wrapper'](out)elifhasattr(out,'close')ornothasattr(out,'__iter__'):returnWSGIFileWrapper(out)# Handle Iterables. We peek into them to detect their inner type.try:iout=iter(out)first=next(iout)whilenotfirst:first=next(iout)exceptStopIteration:_try_close(out)returnself._cast('')exceptHTTPResponseasE:first=Eexcept(KeyboardInterrupt,SystemExit,MemoryError):raiseexceptExceptionaserror:_try_close(out)ifnotself.catchall:raisefirst=HTTPError(500,'Unhandled exception',error,format_exc())# These are the inner types allowed in iterator or generator objects.ifisinstance(first,HTTPResponse):returnself._cast(first)elifisinstance(first,bytes):new_iter=itertools.chain([first],iout)elifisinstance(first,str):encoder=lambdax:x.encode(response.charset)new_iter=map(encoder,itertools.chain([first],iout))else:_try_close(out)msg='Unsupported response type:%s'%type(first)returnself._cast(HTTPError(500,msg))ifhasattr(out,'close'):new_iter=_closeiter(new_iter,out.close)returnnew_iter[docs]defwsgi(self,environ,start_response):""" The bottle WSGI-interface. """out=Nonetry:out=self._cast(self._handle(environ))# rfc2616 section 4.3ifresponse._status_codein(100,101,204,304) \
orenviron['REQUEST_METHOD']=='HEAD':ifhasattr(out,'close'):out.close()out=[]exc_info=environ.get('bottle.exc_info')ifexc_infoisnotNone:delenviron['bottle.exc_info']start_response(response._wsgi_status_line(),response.headerlist,exc_info)returnoutexcept(KeyboardInterrupt,SystemExit,MemoryError):raiseexceptExceptionasE:_try_close(out)ifnotself.catchall:raiseerr='<h1>Critical error while processing request:%s</h1>' \
%html_escape(environ.get('PATH_INFO','/'))ifDEBUG: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)environ['wsgi.errors'].flush()headers=[('Content-Type','text/html; charset=UTF-8')]start_response('500 INTERNAL SERVER ERROR',headers,sys.exc_info())return[tob(err)] def__call__(self,environ,start_response):""" Each instance of :class:'Bottle' is a WSGI application. """returnself.wsgi(environ,start_response)def__enter__(self):""" Use this application as default for all module-level shortcuts. """default_app.push(self)returnselfdef__exit__(self,exc_type,exc_value,traceback):default_app.pop()def__setattr__(self,name,value):ifnameinself.__dict__:raiseAttributeError("Attribute%s already defined. Plugin conflict?"%name)object.__setattr__(self,name,value) ################################################################################ HTTP and WSGI Tools #########################################################################################################################################[docs]classBaseRequest:""" A wrapper for WSGI environment dictionaries that adds a lot of convenient access methods and properties. Most of them are read-only. Adding new attributes to a request actually adds them to the environ dictionary (as 'bottle.request.ext.<name>'). This is the recommended way to store and access request-specific data. """__slots__=('environ',)#: Maximum size of memory buffer for :attr:`body` in bytes.MEMFILE_MAX=102400[docs]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={}ifenvironisNoneelseenvironself.environ['bottle.request']=self [docs]@DictProperty('environ','bottle.app',read_only=True)defapp(self):""" Bottle application handling this request. """raiseRuntimeError('This request is not connected to an application.') [docs]@DictProperty('environ','bottle.route',read_only=True)defroute(self):""" The bottle :class:`Route` object that matches this request. """raiseRuntimeError('This request is not connected to a route.') [docs]@DictProperty('environ','route.url_args',read_only=True)defurl_args(self):""" The arguments extracted from the URL. """raiseRuntimeError('This request is not connected to a route.') @propertydefpath(self):""" The value of ``PATH_INFO`` with exactly one prefixed slash (to fix broken clients and avoid the "empty path" edge case). """return'/'+self.environ.get('PATH_INFO','').lstrip('/')@propertydefmethod(self):""" The ``REQUEST_METHOD`` value as an uppercase string. """returnself.environ.get('REQUEST_METHOD','GET').upper()[docs]@DictProperty('environ','bottle.request.headers',read_only=True)defheaders(self):""" A :class:`WSGIHeaderDict` that provides case-insensitive access to HTTP request headers. """returnWSGIHeaderDict(self.environ) [docs]defget_header(self,name,default=None):""" Return the value of a request header, or a given default value. """returnself.headers.get(name,default) [docs]@DictProperty('environ','bottle.request.cookies',read_only=True)defcookies(self):""" Cookies parsed into a :class:`FormsDict`. Signed cookies are NOT decoded. Use :meth:`get_cookie` if you expect signed cookies. """cookie_header=_wsgi_recode(self.environ.get('HTTP_COOKIE',''))cookies=SimpleCookie(cookie_header).values()returnFormsDict((c.key,c.value)forcincookies) [docs]defget_cookie(self,key,default=None,secret=None,digestmod=hashlib.sha256):""" Return the content of a cookie. To read a `Signed Cookie`, the `secret` must match the one used to create the cookie (see :meth:`Response.set_cookie <BaseResponse.set_cookie>`). If anything goes wrong (missing cookie or wrong signature), return a default value. """value=self.cookies.get(key)ifsecret:# See BaseResponse.set_cookie for details on signed cookies.ifvalueandvalue.startswith('!')and'?'invalue:sig,msg=map(tob,value[1:].split('?',1))hash=hmac.new(tob(secret),msg,digestmod=digestmod).digest()if_lscmp(sig,base64.b64encode(hash)):dst=pickle.loads(base64.b64decode(msg))ifdstanddst[0]==key:returndst[1]returndefaultreturnvalueordefault [docs]@DictProperty('environ','bottle.request.query',read_only=True)defquery(self):""" The :attr:`query_string` parsed into a :class:`FormsDict`. These values are sometimes called "URL arguments" or "GET parameters", but not to be confused with "URL wildcards" as they are provided by the :class:`Router`. """get=self.environ['bottle.get']=FormsDict()pairs=_parse_qsl(self.environ.get('QUERY_STRING',''),'utf8')forkey,valueinpairs:get[key]=valuereturnget [docs]@DictProperty('environ','bottle.request.forms',read_only=True)defforms(self):""" Form values parsed from an `url-encoded` or `multipart/form-data` encoded POST or PUT request body. The result is returned as a :class:`FormsDict`. All keys and values are strings. File uploads are stored separately in :attr:`files`. """forms=FormsDict()forname,iteminself.POST.allitems():ifnotisinstance(item,FileUpload):forms[name]=itemreturnforms [docs]@DictProperty('environ','bottle.request.params',read_only=True)defparams(self):""" A :class:`FormsDict` with the combined values of :attr:`query` and :attr:`forms`. File uploads are stored in :attr:`files`. """params=FormsDict()forkey,valueinself.query.allitems():params[key]=valueforkey,valueinself.forms.allitems():params[key]=valuereturnparams [docs]@DictProperty('environ','bottle.request.files',read_only=True)deffiles(self):""" File uploads parsed from `multipart/form-data` encoded POST or PUT request body. The values are instances of :class:`FileUpload`. """files=FormsDict()forname,iteminself.POST.allitems():ifisinstance(item,FileUpload):files[name]=itemreturnfiles [docs]@DictProperty('environ','bottle.request.json',read_only=True)defjson(self):""" If the ``Content-Type`` header is ``application/json`` or ``application/json-rpc``, this property holds the parsed content of the request body. Only requests smaller than :attr:`MEMFILE_MAX` are processed to avoid memory exhaustion. Invalid JSON raises a 400 error response. """ctype=self.environ.get('CONTENT_TYPE','').lower().split(';')[0]ifctypein('application/json','application/json-rpc'):b=self._get_body_string(self.MEMFILE_MAX)ifnotb:returnNonetry:returnjson_loads(b)except(ValueError,TypeError)aserr:raiseHTTPError(400,'Invalid JSON',exception=err)returnNone def_iter_body(self,read,bufsize):maxread=max(0,self.content_length)whilemaxread:part=read(min(maxread,bufsize))ifnotpart:breakyieldpartmaxread-=len(part)@staticmethoddef_iter_chunked(read,bufsize):err=HTTPError(400,'Error while parsing chunked transfer body.')rn,sem,bs=b'\r\n',b';',b''whileTrue:header=read(1)whileheader[-2:]!=rn:c=read(1)header+=cifnotc:raiseerriflen(header)>bufsize:raiseerrsize,_,_=header.partition(sem)try:maxread=int(size.strip(),16)exceptValueError:raiseerrifmaxread==0:breakbuff=bswhilemaxread>0:ifnotbuff:buff=read(min(maxread,bufsize))part,buff=buff[:maxread],buff[maxread:]ifnotpart:raiseerryieldpartmaxread-=len(part)ifread(2)!=rn:raiseerr@DictProperty('environ','bottle.request.body',read_only=True)def_body(self):try:read_func=self.environ['wsgi.input'].readexceptKeyError:self.environ['wsgi.input']=BytesIO()returnself.environ['wsgi.input']body_iter=self._iter_chunkedifself.chunkedelseself._iter_bodybody,body_size,is_temp_file=BytesIO(),0,Falseforpartinbody_iter(read_func,self.MEMFILE_MAX):body.write(part)body_size+=len(part)ifnotis_temp_fileandbody_size>self.MEMFILE_MAX:body,tmp=NamedTemporaryFile(mode='w+b'),bodybody.write(tmp.getvalue())deltmpis_temp_file=Trueself.environ['wsgi.input']=bodybody.seek(0)returnbodydef_get_body_string(self,maxread):""" Read body into a string. Raise HTTPError(413) on requests that are too large. """ifself.content_length>maxread:raiseHTTPError(413,'Request entity too large')data=self.body.read(maxread+1)iflen(data)>maxread:raiseHTTPError(413,'Request entity too large')returndata@propertydefbody(self):""" The HTTP request body as a seek-able file-like object. Depending on :attr:`MEMFILE_MAX`, this is either a temporary file or a :class:`io.BytesIO` instance. Accessing this property for the first time reads and replaces the ``wsgi.input`` environ variable. Subsequent accesses just do a `seek(0)` on the file object. """self._body.seek(0)returnself._body@propertydefchunked(self):""" True if Chunked transfer encoding was. """return'chunked'inself.environ.get('HTTP_TRANSFER_ENCODING','').lower()#: An alias for :attr:`query`.GET=query[docs]@DictProperty('environ','bottle.request.post',read_only=True)defPOST(self):""" The values of :attr:`forms` and :attr:`files` combined into a single :class:`FormsDict`. Values are either strings (form values) or instances of :class:`FileUpload`. """post=FormsDict()content_type=self.environ.get('CONTENT_TYPE','')content_type,options=_parse_http_header(content_type)[0]# We default to application/x-www-form-urlencoded for everything that# is not multipart and take the fast path (also: 3.1 workaround)ifnotcontent_type.startswith('multipart/'):body=self._get_body_string(self.MEMFILE_MAX).decode('utf8','surrogateescape')forkey,valuein_parse_qsl(body,'utf8'):post[key]=valuereturnpostcharset=options.get("charset","utf8")boundary=options.get("boundary")ifnotboundary:raiseMultipartError("Invalid content type header, missing boundary")parser=_MultipartParser(self.body,boundary,self.content_length,mem_limit=self.MEMFILE_MAX,memfile_limit=self.MEMFILE_MAX,charset=charset)forpartinparser.parse():ifnotpart.filenameandpart.is_buffered():post[part.name]=part.valueelse:post[part.name]=FileUpload(part.file,part.name,part.filename,part.headerlist)returnpost @propertydefurl(self):""" The full request URI including hostname and scheme. If your app lives behind a reverse proxy or load balancer and you get confusing results, make sure that the ``X-Forwarded-Host`` header is set correctly. """returnself.urlparts.geturl()[docs]@DictProperty('environ','bottle.request.urlparts',read_only=True)defurlparts(self):""" The :attr:`url` string as an :class:`urlparse.SplitResult` tuple. The tuple contains (scheme, host, path, query_string and fragment), but the fragment is always empty because it is not visible to the server. """env=self.environhttp=env.get('HTTP_X_FORWARDED_PROTO')orenv.get('wsgi.url_scheme','http')host=env.get('HTTP_X_FORWARDED_HOST')orenv.get('HTTP_HOST')ifnothost:# HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients.host=env.get('SERVER_NAME','127.0.0.1')port=env.get('SERVER_PORT')ifportandport!=('80'ifhttp=='http'else'443'):host+=':'+portpath=urlquote(self.fullpath)returnUrlSplitResult(http,host,path,env.get('QUERY_STRING'),'') @propertydeffullpath(self):""" Request path including :attr:`script_name` (if present). """returnurljoin(self.script_name,self.path.lstrip('/'))@propertydefquery_string(self):""" The raw :attr:`query` part of the URL (everything in between ``?`` and ``#``) as a string. """returnself.environ.get('QUERY_STRING','')@propertydefscript_name(self):""" The initial portion of the URL's `path` that was removed by a higher level (server or routing middleware) before the application was called. This script path is returned with leading and tailing slashes. """script_name=self.environ.get('SCRIPT_NAME','').strip('/')return'/'+script_name+'/'ifscript_nameelse'/'[docs]defpath_shift(self,shift=1):""" Shift path segments from :attr:`path` to :attr:`script_name` and vice versa. :param shift: The number of path segments to shift. May be negative to change the shift direction. (default: 1) """script,path=path_shift(self.environ.get('SCRIPT_NAME','/'),self.path,shift)self['SCRIPT_NAME'],self['PATH_INFO']=script,path @propertydefcontent_length(self):""" The request body length as an integer. The client is responsible to set this header. Otherwise, the real length of the body is unknown and -1 is returned. In this case, :attr:`body` will be empty. """returnint(self.environ.get('CONTENT_LENGTH')or-1)@propertydefcontent_type(self):""" The Content-Type header as a lowercase-string (default: empty). """returnself.environ.get('CONTENT_TYPE','').lower()@propertydefis_xhr(self):""" True if the request was triggered by a XMLHttpRequest. This only works with JavaScript libraries that support the `X-Requested-With` header (most of the popular libraries do). """requested_with=self.environ.get('HTTP_X_REQUESTED_WITH','')returnrequested_with.lower()=='xmlhttprequest'@propertydefis_ajax(self):""" Alias for :attr:`is_xhr`. "Ajax" is not the right term. """returnself.is_xhr@propertydefauth(self):""" HTTP authentication data as a (user, password) tuple. This implementation currently supports basic (not digest) authentication only. If the authentication happened at a higher level (e.g. in the front web-server or a middleware), the password field is None, but the user field is looked up from the ``REMOTE_USER`` environ variable. On any errors, None is returned. """basic=parse_auth(self.environ.get('HTTP_AUTHORIZATION',''))ifbasic:returnbasicruser=self.environ.get('REMOTE_USER')ifruser:return(ruser,None)returnNone@propertydefremote_route(self):""" A list of all IPs that were involved in this request, starting with the client IP and followed by zero or more proxies. This does only work if all proxies support the ```X-Forwarded-For`` header. Note that this information can be forged by malicious clients. """proxy=self.environ.get('HTTP_X_FORWARDED_FOR')ifproxy:return[ip.strip()foripinproxy.split(',')]remote=self.environ.get('REMOTE_ADDR')return[remote]ifremoteelse[]@propertydefremote_addr(self):""" The client IP as a string. Note that this information can be forged by malicious clients. """route=self.remote_routereturnroute[0]ifrouteelseNone[docs]defcopy(self):""" Return a new :class:`Request` with a shallow :attr:`environ` copy. """returnRequest(self.environ.copy()) defget(self,key,default=None):returnself.environ.get(key,default)def__getitem__(self,key):returnself.environ[key]def__delitem__(self,key):self[key]=""del(self.environ[key])def__iter__(self):returniter(self.environ)def__len__(self):returnlen(self.environ)defkeys(self):returnself.environ.keys()def__setitem__(self,key,value):""" Change an environ value and clear all caches that depend on it. """ifself.environ.get('bottle.request.readonly'):raiseKeyError('The environ dictionary is read-only.')self.environ[key]=valuetodelete=()ifkey=='wsgi.input':todelete=('body','forms','files','params','post','json')elifkey=='QUERY_STRING':todelete=('query','params')elifkey.startswith('HTTP_'):todelete=('headers','cookies')forkeyintodelete:self.environ.pop('bottle.request.'+key,None)def__repr__(self):return'<%s:%s%s>'%(self.__class__.__name__,self.method,self.url)def__getattr__(self,name):""" Search in self.environ for additional user defined attributes. """try:var=self.environ['bottle.request.ext.%s'%name]returnvar.__get__(self)ifhasattr(var,'__get__')elsevarexceptKeyError:raiseAttributeError('Attribute%r not defined.'%name)[docs]def__setattr__(self,name,value):""" Define new attributes that are local to the bound request environment. """ifname=='environ':returnobject.__setattr__(self,name,value)key='bottle.request.ext.%s'%nameifhasattr(self,name):raiseAttributeError("Attribute already defined:%s"%name)self.environ[key]=value def__delattr__(self,name):try:delself.environ['bottle.request.ext.%s'%name]exceptKeyError:raiseAttributeError("Attribute not defined:%s"%name) def_hkey(key):key=touni(key)if'\n'inkeyor'\r'inkeyor'\0'inkey:raiseValueError("Header names must not contain control characters:%r"%key)returnkey.title().replace('_','-')def_hval(value):value=touni(value)if'\n'invalueor'\r'invalueor'\0'invalue:raiseValueError("Header value must not contain control characters:%r"%value)returnvalueclassHeaderProperty:def__init__(self,name,reader=None,writer=None,default=''):self.name,self.default=name,defaultself.reader,self.writer=reader,writerself.__doc__='Current value of the%r header.'%name.title()def__get__(self,obj,_):ifobjisNone:returnselfvalue=obj.get_header(self.name,self.default)returnself.reader(value)ifself.readerelsevaluedef__set__(self,obj,value):obj[self.name]=self.writer(value)ifself.writerelsevaluedef__delete__(self,obj):delobj[self.name][docs]classBaseResponse:""" Storage class for a response body as well as headers and cookies. This class does support dict-like case-insensitive item-access to headers, but is NOT a dict. Most notably, iterating over a response yields parts of the body and not the headers. """default_status=200default_content_type='text/html; charset=UTF-8'# Header denylist for specific response codes# (rfc2616 section 10.2.3 and 10.3.5)bad_headers={204:frozenset(('Content-Type','Content-Length')),304:frozenset(('Allow','Content-Encoding','Content-Language','Content-Length','Content-Range','Content-Type','Content-Md5','Last-Modified'))}[docs]def__init__(self,body='',status=None,headers=None,**more_headers):""" Create a new response object. :param body: The response body as one of the supported types. :param status: Either an HTTP status code (e.g. 200) or a status line including the reason phrase (e.g. '200 OK'). :param headers: A dictionary or a list of name-value pairs. Additional keyword arguments are added to the list of headers. Underscores in the header name are replaced with dashes. """self._cookies=Noneself._headers={}self.body=bodyself.status=statusorself.default_statusifheaders:ifisinstance(headers,dict):headers=headers.items()forname,valueinheaders:self.add_header(name,value)ifmore_headers:forname,valueinmore_headers.items():self.add_header(name,value) [docs]defcopy(self,cls=None):""" Returns a copy of self. """cls=clsorBaseResponseassertissubclass(cls,BaseResponse)copy=cls()copy.status=self.statuscopy._headers=dict((k,v[:])for(k,v)inself._headers.items())ifself._cookies:cookies=copy._cookies=SimpleCookie()fork,vinself._cookies.items():cookies[k]=v.valuecookies[k].update(v)# also copy cookie attributesreturncopy def__iter__(self):returniter(self.body)defclose(self):ifhasattr(self.body,'close'):self.body.close()@propertydefstatus_line(self):""" The HTTP status line as a string (e.g. ``404 Not Found``)."""returnself._status_line@propertydefstatus_code(self):""" The HTTP status code as an integer (e.g. 404)."""returnself._status_codedef_set_status(self,status):ifisinstance(status,int):code,status=status,_HTTP_STATUS_LINES.get(status)elif' 'instatus:if'\n'instatusor'\r'instatusor'\0'instatus:raiseValueError('Status line must not include control chars.')status=status.strip()code=int(status.split()[0])else:raiseValueError('String status line without a reason phrase.')ifnot100<=code<=999:raiseValueError('Status code out of range.')self._status_code=codeself._status_line=str(statusor('%d Unknown'%code))def_get_status(self):returnself._status_linestatus=property(_get_status,_set_status,None,''' A writeable property to change the HTTP response status. It accepts either a numeric code (100-999) or a string with a custom reason phrase (e.g. "404 Brain not found"). Both :data:`status_line` and :data:`status_code` are updated accordingly. The return value is always a status string. ''')del_get_status,_set_status@propertydefheaders(self):""" An instance of :class:`HeaderDict`, a case-insensitive dict-like view on the response headers. """hdict=HeaderDict()hdict.dict=self._headersreturnhdictdef__contains__(self,name):return_hkey(name)inself._headersdef__delitem__(self,name):delself._headers[_hkey(name)]def__getitem__(self,name):returnself._headers[_hkey(name)][-1]def__setitem__(self,name,value):self._headers[_hkey(name)]=[_hval(value)][docs]defget_header(self,name,default=None):""" Return the value of a previously defined header. If there is no header with that name, return a default value. """returnself._headers.get(_hkey(name),[default])[-1] [docs]defset_header(self,name,value):""" Create a new response header, replacing any previously defined headers with the same name. """self._headers[_hkey(name)]=[_hval(value)] [docs]defadd_header(self,name,value):""" Add an additional response header, not removing duplicates. """self._headers.setdefault(_hkey(name),[]).append(_hval(value)) [docs]defiter_headers(self):""" Yield (header, value) tuples, skipping headers that are not allowed with the current response status code. """returnself.headerlist def_wsgi_status_line(self):""" WSGI conform status line (latin1-encodeable) """returnself._status_line.encode('utf8','surrogateescape').decode('latin1')@propertydefheaderlist(self):""" WSGI conform list of (header, value) tuples. """out=[]headers=list(self._headers.items())if'Content-Type'notinself._headers:headers.append(('Content-Type',[self.default_content_type]))ifself._status_codeinself.bad_headers:bad_headers=self.bad_headers[self._status_code]headers=[hforhinheadersifh[0]notinbad_headers]out+=[(name,val)for(name,vals)inheadersforvalinvals]ifself._cookies:forcinself._cookies.values():out.append(('Set-Cookie',_hval(c.OutputString())))out=[(k,v.encode('utf8','surrogateescape').decode('latin1'))for(k,v)inout]returnoutcontent_type=HeaderProperty('Content-Type')content_length=HeaderProperty('Content-Length',reader=int,default=-1)expires=HeaderProperty('Expires',reader=lambdax:datetime.fromtimestamp(parse_date(x),UTC),writer=lambdax:http_date(x))@propertydefcharset(self,default='UTF-8'):""" Return the charset specified in the content-type header (default: utf8). """if'charset='inself.content_type:returnself.content_type.split('charset=')[-1].split(';')[0].strip()returndefault[docs]defset_cookie(self,name,value,secret=None,digestmod=hashlib.sha256,**options):""" Create a new cookie or replace an old one. If the `secret` parameter is set, create a `Signed Cookie` (described below). :param name: the name of the cookie. :param value: the value of the cookie. :param secret: a signature key required for signed cookies. Additionally, this method accepts all RFC 2109 attributes that are supported by :class:`cookie.Morsel`, including: :param maxage: maximum age in seconds. (default: None) :param expires: a datetime object or UNIX timestamp. (default: None) :param domain: the domain that is allowed to read the cookie. (default: current domain) :param path: limits the cookie to a given path (default: current path) :param secure: limit the cookie to HTTPS connections (default: off). :param httponly: prevents client-side javascript to read this cookie (default: off, requires Python 2.6 or newer). :param samesite: Control or disable third-party use for this cookie. Possible values: `lax`, `strict` or `none` (default). If neither `expires` nor `maxage` is set (default), the cookie will expire at the end of the browser session (as soon as the browser window is closed). Signed cookies may store any pickle-able object and are cryptographically signed to prevent manipulation. Keep in mind that cookies are limited to 4kb in most browsers. Warning: Pickle is a potentially dangerous format. If an attacker gains access to the secret key, he could forge cookies that execute code on server side if unpickled. Using pickle is discouraged and support for it will be removed in later versions of bottle. Warning: Signed cookies are not encrypted (the client can still see the content) and not copy-protected (the client can restore an old cookie). The main intention is to make pickling and unpickling save, not to store secret information at client side. """ifnotself._cookies:self._cookies=SimpleCookie()# Monkey-patch Cookie lib to support 'SameSite' parameter# https://tools.ietf.org/html/draft-west-first-party-cookies-07#section-4.1ifpy<(3,8,0):Morsel._reserved.setdefault('samesite','SameSite')ifsecret:ifnotisinstance(value,str):depr(0,13,"Pickling of arbitrary objects into cookies is ""deprecated.","Only store strings in cookies. ""JSON strings are fine, too.")encoded=base64.b64encode(pickle.dumps([name,value],-1))sig=base64.b64encode(hmac.new(tob(secret),encoded,digestmod=digestmod).digest())value=touni(b'!'+sig+b'?'+encoded)elifnotisinstance(value,str):raiseTypeError('Secret key required for non-string cookies.')# Cookie size plus options must not exceed 4kb.iflen(name)+len(value)>3800:raiseValueError('Content does not fit into a cookie.')self._cookies[name]=valueforkey,valueinoptions.items():ifkeyin('max_age','maxage'):# 'maxage' variant added in 0.13key='max-age'ifisinstance(value,timedelta):value=value.seconds+value.days*24*3600ifkey=='expires':value=http_date(value)ifkeyin('same_site','samesite'):# 'samesite' variant added in 0.13key,value='samesite',(valueor"none").lower()ifvaluenotin('lax','strict','none'):raiseCookieError("Invalid value for SameSite")ifkeyin('secure','httponly')andnotvalue:continueself._cookies[name][key]=value [docs]defdelete_cookie(self,key,**kwargs):""" Delete a cookie. Be sure to use the same `domain` and `path` settings as used to create the cookie. """kwargs['max_age']=-1kwargs['expires']=0self.set_cookie(key,'',**kwargs) def__repr__(self):out=''forname,valueinself.headerlist:out+='%s:%s\n'%(name.title(),value.strip())returnout def_local_property():ls=threading.local()deffget(_):try:returnls.varexceptAttributeError:raiseRuntimeError("Request context not initialized.")deffset(_,value):ls.var=valuedeffdel(_):dells.varreturnproperty(fget,fset,fdel,'Thread-local property')[docs]classLocalRequest(BaseRequest):""" A thread-local subclass of :class:`BaseRequest` with a different set of attributes for each thread. There is usually only one global instance of this class (:data:`request`). If accessed during a request/response cycle, this instance always refers to the *current* request (even on a multithreaded server). """bind=BaseRequest.__init__environ=_local_property() [docs]classLocalResponse(BaseResponse):""" A thread-local subclass of :class:`BaseResponse` with a different set of attributes for each thread. There is usually only one global instance of this class (:data:`response`). Its attributes are used to build the HTTP response at the end of the request/response cycle. """bind=BaseResponse.__init___status_line=_local_property()_status_code=_local_property()_cookies=_local_property()_headers=_local_property()body=_local_property() Request=BaseRequestResponse=BaseResponse[docs]classHTTPResponse(Response,BottleException):""" A subclass of :class:`Response` that can be raised or returned from request handlers to short-curcuit request processing and override changes made to the global :data:`request` object. This bypasses error handlers, even if the status code indicates an error. Return or raise :class:`HTTPError` to trigger error handlers. """[docs]def__init__(self,body='',status=None,headers=None,**more_headers):super(HTTPResponse,self).__init__(body,status,headers,**more_headers) [docs]defapply(self,other):""" Copy the state of this response to a different :class:`Response` object. """other._status_code=self._status_codeother._status_line=self._status_lineother._headers=self._headersother._cookies=self._cookiesother.body=self.body [docs]classHTTPError(HTTPResponse):""" A subclass of :class:`HTTPResponse` that triggers error handlers. """default_status=500[docs]def__init__(self,status=None,body=None,exception=None,traceback=None,**more_headers):self.exception=exceptionself.traceback=tracebacksuper(HTTPError,self).__init__(body,status,**more_headers) ################################################################################ Plugins #####################################################################################################################################################classPluginError(BottleException):passclassJSONPlugin:name='json'api=2def__init__(self,json_dumps=json_dumps):self.json_dumps=json_dumpsdefsetup(self,app):app.config._define('json.enable',default=True,validate=bool,help="Enable or disable automatic dict->json filter.")app.config._define('json.ascii',default=False,validate=bool,help="Use only 7-bit ASCII characters in output.")app.config._define('json.indent',default=True,validate=bool,help="Add whitespace to make json more readable.")app.config._define('json.dump_func',default=None,help="If defined, use this function to transform"" dict into json. The other options no longer"" apply.")defapply(self,callback,route):dumps=self.json_dumpsifnotself.json_dumps:returncallback@functools.wraps(callback)defwrapper(*a,**ka):try:rv=callback(*a,**ka)exceptHTTPResponseasresp:rv=respifisinstance(rv,dict):# Attempt to serialize, raises exception on failurejson_response=dumps(rv)# Set content type only if serialization successfulresponse.content_type='application/json'returnjson_responseelifisinstance(rv,HTTPResponse)andisinstance(rv.body,dict):rv.body=dumps(rv.body)rv.content_type='application/json'returnrvreturnwrapperclassTemplatePlugin:""" This plugin applies the :func:`view` decorator to all routes with a `template` config parameter. If the parameter is a tuple, the second element must be a dict with additional options (e.g. `template_engine`) or default variables for the template. """name='template'api=2defsetup(self,app):app.tpl=selfdefapply(self,callback,route):conf=route.config.get('template')ifisinstance(conf,(tuple,list))andlen(conf)==2:returnview(conf[0],**conf[1])(callback)elifisinstance(conf,str):returnview(conf)(callback)else:returncallback#: Not a plugin, but part of the plugin API. TODO: Find a better place.class_ImportRedirect:def__init__(self,name,impmask):""" Create a virtual package that redirects imports (see PEP 302). """self.name=nameself.impmask=impmaskself.module=sys.modules.setdefault(name,new_module(name))self.module.__dict__.update({'__file__':__file__,'__path__':[],'__all__':[],'__loader__':self})sys.meta_path.append(self)deffind_spec(self,fullname,path,target=None):if'.'notinfullname:returniffullname.rsplit('.',1)[0]!=self.name:returnfromimportlib.utilimportspec_from_loaderreturnspec_from_loader(fullname,self)deffind_module(self,fullname,path=None):if'.'notinfullname:returniffullname.rsplit('.',1)[0]!=self.name:returnreturnselfdefcreate_module(self,spec):returnself.load_module(spec.name)defexec_module(self,module):pass# This probably breaks importlib.reload() :/defload_module(self,fullname):iffullnameinsys.modules:returnsys.modules[fullname]modname=fullname.rsplit('.',1)[1]realname=self.impmask%modname__import__(realname)module=sys.modules[fullname]=sys.modules[realname]setattr(self.module,modname,module)module.__loader__=selfreturnmodule################################################################################ Common Utilities ############################################################################################################################################[docs]classMultiDict(DictMixin):""" This dict stores multiple values per key, but behaves exactly like a normal dict in that it returns only the newest value for any given key. There are special methods available to access the full list of values. """[docs]def__init__(self,*a,**k):self.dict=dict((k,[v])for(k,v)indict(*a,**k).items()) def__len__(self):returnlen(self.dict)def__iter__(self):returniter(self.dict)def__contains__(self,key):returnkeyinself.dictdef__delitem__(self,key):delself.dict[key]def__getitem__(self,key):returnself.dict[key][-1]def__setitem__(self,key,value):self.append(key,value)[docs]defkeys(self):returnself.dict.keys() [docs]defvalues(self):return(v[-1]forvinself.dict.values()) [docs]defitems(self):return((k,v[-1])fork,vinself.dict.items()) defallitems(self):return((k,v)fork,vlinself.dict.items()forvinvl)iterkeys=keysitervalues=valuesiteritems=itemsiterallitems=allitems[docs]defget(self,key,default=None,index=-1,type=None):""" Return the most recent value for a key. :param default: The default value to be returned if the key is not present or the type conversion fails. :param index: An index for the list of available values. :param type: If defined, this callable is used to cast the value into a specific type. Exception are suppressed and result in the default value to be returned. """try:val=self.dict[key][index]returntype(val)iftypeelsevalexceptException:passreturndefault [docs]defappend(self,key,value):""" Add a new value to the list of values for this key. """self.dict.setdefault(key,[]).append(value) [docs]defreplace(self,key,value):""" Replace the list of values with a single value. """self.dict[key]=[value] [docs]defgetall(self,key):""" Return a (possibly empty) list of values for a key. """returnself.dict.get(key)or[] #: Aliases for WTForms to mimic other multi-dict APIs (Django)getone=getgetlist=getall [docs]classFormsDict(MultiDict):""" This :class:`MultiDict` subclass is used to store request form data. Additionally to the normal dict-like item access methods, this container also supports attribute-like access to its values. Missing attributes default to an empty string. .. versionchanged:: 0.14 All keys and values are now decoded as utf8 by default, item and attribute access will return the same string. """[docs]defdecode(self,encoding=None):""" (deprecated) Starting with 0.13 all keys and values are already correctly decoded. """copy=FormsDict()forkey,valueinself.allitems():copy[key]=valuereturncopy [docs]defgetunicode(self,name,default=None,encoding=None):""" (deprecated) Return the value as a unicode string, or the default. """returnself.get(name,default) def__getattr__(self,name,default=str()):# Without this guard, pickle generates a cryptic TypeError:ifname.startswith('__')andname.endswith('__'):returnsuper(FormsDict,self).__getattr__(name)returnself.get(name,default=default) [docs]classHeaderDict(MultiDict):""" A case-insensitive version of :class:`MultiDict` that defaults to replace the old value instead of appending it. """[docs]def__init__(self,*a,**ka):self.dict={}ifaorka:self.update(*a,**ka) def__contains__(self,key):return_hkey(key)inself.dictdef__delitem__(self,key):delself.dict[_hkey(key)]def__getitem__(self,key):returnself.dict[_hkey(key)][-1]def__setitem__(self,key,value):self.dict[_hkey(key)]=[_hval(value)][docs]defappend(self,key,value):self.dict.setdefault(_hkey(key),[]).append(_hval(value)) [docs]defreplace(self,key,value):self.dict[_hkey(key)]=[_hval(value)] [docs]defgetall(self,key):returnself.dict.get(_hkey(key))or[] [docs]defget(self,key,default=None,index=-1):returnMultiDict.get(self,_hkey(key),default,index) deffilter(self,names):fornamein(_hkey(n)forninnames):ifnameinself.dict:delself.dict[name] [docs]classWSGIHeaderDict(DictMixin):""" This dict-like class wraps a WSGI environ dict and provides convenient access to HTTP_* fields. Header names are case-insensitive and titled by default. """#: List of keys that do not have a ``HTTP_`` prefix.cgikeys=('CONTENT_TYPE','CONTENT_LENGTH')[docs]def__init__(self,environ):self.environ=environ def_ekey(self,key):""" Translate header field name to CGI/WSGI environ key. """key=key.replace('-','_').upper()ifkeyinself.cgikeys:returnkeyreturn'HTTP_'+key[docs]defraw(self,key,default=None):""" Return the header value as is (not utf8-translated). """returnself.environ.get(self._ekey(key),default) def__getitem__(self,key):return_wsgi_recode(self.environ[self._ekey(key)])def__setitem__(self,key,value):raiseTypeError("%s is read-only."%self.__class__)def__delitem__(self,key):raiseTypeError("%s is read-only."%self.__class__)def__iter__(self):forkeyinself.environ:ifkey[:5]=='HTTP_':yield_hkey(key[5:])elifkeyinself.cgikeys:yield_hkey(key)[docs]defkeys(self):return[xforxinself] def__len__(self):returnlen(self.keys())def__contains__(self,key):returnself._ekey(key)inself.environ [docs]classConfigDict(dict):""" A dict-like configuration storage with additional support for namespaces, validators, meta-data and overlays. This dict-like class is heavily optimized for read access. Read-only methods and item access should be as fast as a native dict. """__slots__=('_meta','_change_listener','_overlays','_virtual_keys','_source','__weakref__')[docs]def__init__(self):self._meta={}self._change_listener=[]#: Weak references of overlays that need to be kept in sync.self._overlays=[]#: Config that is the source for this overlay.self._source=None#: Keys of values copied from the source (values we do not own)self._virtual_keys=set() [docs]defload_module(self,name,squash=True):"""Load values from a Python module. Import a python module by name and add all upper-case module-level variables to this config dict. :param name: Module name to import and load. :param squash: If true (default), nested dicts are assumed to represent namespaces and flattened (see :meth:`load_dict`). """config_obj=load(name)obj={key:getattr(config_obj,key)forkeyindir(config_obj)ifkey.isupper()}ifsquash:self.load_dict(obj)else:self.update(obj)returnself [docs]defload_config(self,filename,**options):""" Load values from ``*.ini`` style config files using configparser. INI style sections (e.g. ``[section]``) are used as namespace for all keys within that section. Both section and key names may contain dots as namespace separators and are converted to lower-case. The special sections ``[bottle]`` and ``[ROOT]`` refer to the root namespace and the ``[DEFAULT]`` section defines default values for all other sections. :param filename: The path of a config file, or a list of paths. :param options: All keyword parameters are passed to the underlying :class:`python:configparser.ConfigParser` constructor call. """options.setdefault('allow_no_value',True)options.setdefault('interpolation',configparser.ExtendedInterpolation())conf=configparser.ConfigParser(**options)conf.read(filename)forsectioninconf.sections():forkeyinconf.options(section):value=conf.get(section,key)ifsectionnotin('bottle','ROOT'):key=section+'.'+keyself[key.lower()]=valuereturnself [docs]defload_dict(self,source,namespace=''):""" Load values from a dictionary structure. Nesting can be used to represent namespaces. >>> c = ConfigDict() >>> c.load_dict({'some': {'namespace': {'key': 'value'} } }) {'some.namespace.key': 'value'} """forkey,valueinsource.items():ifisinstance(key,str):nskey=(namespace+'.'+key).strip('.')ifisinstance(value,dict):self.load_dict(value,namespace=nskey)else:self[nskey]=valueelse:raiseTypeError('Key has type%r (not a string)'%type(key))returnself [docs]defupdate(self,*a,**ka):""" If the first parameter is a string, all keys are prefixed with this namespace. Apart from that it works just as the usual dict.update(). >>> c = ConfigDict() >>> c.update('some.namespace', key='value') """prefix=''ifaandisinstance(a[0],str):prefix=a[0].strip('.')+'.'a=a[1:]forkey,valueindict(*a,**ka).items():self[prefix+key]=value [docs]defsetdefault(self,key,value=None):ifkeynotinself:self[key]=valuereturnself[key] def__setitem__(self,key,value):ifnotisinstance(key,str):raiseTypeError('Key has type%r (not a string)'%type(key))self._virtual_keys.discard(key)value=self.meta_get(key,'filter',lambdax:x)(value)ifkeyinselfandself[key]isvalue:returnself._on_change(key,value)dict.__setitem__(self,key,value)foroverlayinself._iter_overlays():overlay._set_virtual(key,value)def__delitem__(self,key):ifkeynotinself:raiseKeyError(key)ifkeyinself._virtual_keys:raiseKeyError("Virtual keys cannot be deleted:%s"%key)ifself._sourceandkeyinself._source:# Not virtual, but present in source -> Restore virtual valuedict.__delitem__(self,key)self._set_virtual(key,self._source[key])else:# not virtual, not present in source. This is OUR valueself._on_change(key,None)dict.__delitem__(self,key)foroverlayinself._iter_overlays():overlay._delete_virtual(key)def_set_virtual(self,key,value):""" Recursively set or update virtual keys. """ifkeyinselfandkeynotinself._virtual_keys:return# Do nothing for non-virtual keys.self._virtual_keys.add(key)ifkeyinselfandself[key]isnotvalue:self._on_change(key,value)dict.__setitem__(self,key,value)foroverlayinself._iter_overlays():overlay._set_virtual(key,value)def_delete_virtual(self,key):""" Recursively delete virtual entry. """ifkeynotinself._virtual_keys:return# Do nothing for non-virtual keys.ifkeyinself:self._on_change(key,None)dict.__delitem__(self,key)self._virtual_keys.discard(key)foroverlayinself._iter_overlays():overlay._delete_virtual(key)def_on_change(self,key,value):forcbinself._change_listener:ifcb(self,key,value):returnTruedef_add_change_listener(self,func):self._change_listener.append(func)returnfunc[docs]defmeta_get(self,key,metafield,default=None):""" Return the value of a meta field for a key. """returnself._meta.get(key,{}).get(metafield,default) [docs]defmeta_set(self,key,metafield,value):""" Set the meta field for a key to a new value. Meta-fields are shared between all members of an overlay tree. """self._meta.setdefault(key,{})[metafield]=value [docs]defmeta_list(self,key):""" Return an iterable of meta field names defined for a key. """returnself._meta.get(key,{}).keys() def_define(self,key,default=_UNSET,help=_UNSET,validate=_UNSET):""" (Unstable) Shortcut for plugins to define own config parameters. """ifdefaultisnot_UNSET:self.setdefault(key,default)ifhelpisnot_UNSET:self.meta_set(key,'help',help)ifvalidateisnot_UNSET:self.meta_set(key,'validate',validate)def_iter_overlays(self):forrefinself._overlays:overlay=ref()ifoverlayisnotNone:yieldoverlaydef_make_overlay(self):""" (Unstable) Create a new overlay that acts like a chained map: Values missing in the overlay are copied from the source map. Both maps share the same meta entries. Entries that were copied from the source are called 'virtual'. You can not delete virtual keys, but overwrite them, which turns them into non-virtual entries. Setting keys on an overlay never affects its source, but may affect any number of child overlays. Other than collections.ChainMap or most other implementations, this approach does not resolve missing keys on demand, but instead actively copies all values from the source to the overlay and keeps track of virtual and non-virtual keys internally. This removes any lookup-overhead. Read-access is as fast as a build-in dict for both virtual and non-virtual keys. Changes are propagated recursively and depth-first. A failing on-change handler in an overlay stops the propagation of virtual values and may result in an partly updated tree. Take extra care here and make sure that on-change handlers never fail. Used by Route.config """# Cleanup dead referencesself._overlays[:]=[refforrefinself._overlaysifref()isnotNone]overlay=ConfigDict()overlay._meta=self._metaoverlay._source=selfself._overlays.append(weakref.ref(overlay))forkeyinself:overlay._set_virtual(key,self[key])returnoverlay [docs]classAppStack(list):""" A stack-like list. Calling it returns the head of the stack. """def__call__(self):""" Return the current default application. """returnself.default[docs]defpush(self,value=None):""" Add a new :class:`Bottle` instance to the stack """ifnotisinstance(value,Bottle):value=Bottle()self.append(value)returnvalue new_app=push@propertydefdefault(self):try:returnself[-1]exceptIndexError:returnself.push() classWSGIFileWrapper:def__init__(self,fp,buffer_size=1024*64):self.fp,self.buffer_size=fp,buffer_sizeforattrin'fileno','close','read','readlines','tell','seek':ifhasattr(fp,attr):setattr(self,attr,getattr(fp,attr))def__iter__(self):buff,read=self.buffer_size,self.readpart=read(buff)whilepart:yieldpartpart=read(buff)class_closeiter:""" This only exists to be able to attach a .close method to iterators that do not support attribute assignment (most of itertools). """def__init__(self,iterator,close=None):self.iterator=iteratorself.close_callbacks=makelist(close)def__iter__(self):returniter(self.iterator)defclose(self):forfuncinself.close_callbacks:func()def_try_close(obj):""" Call obj.close() if present and ignore exceptions """try:ifhasattr(obj,'close'):obj.close()exceptException:passclassResourceManager:""" This class manages a list of search paths and helps to find and open application-bound resources (files). :param base: default value for :meth:`add_path` calls. :param opener: callable used to open resources. :param cachemode: controls which lookups are cached. One of 'all', 'found' or 'none'. """def__init__(self,base='./',opener=open,cachemode='all'):self.opener=openerself.base=baseself.cachemode=cachemode#: A list of search paths. See :meth:`add_path` for details.self.path=[]#: A cache for resolved paths. ``res.cache.clear()`` clears the cache.self.cache={}defadd_path(self,path,base=None,index=None,create=False):""" Add a new path to the list of search paths. Return False if the path does not exist. :param path: The new search path. Relative paths are turned into an absolute and normalized form. If the path looks like a file (not ending in `/`), the filename is stripped off. :param base: Path used to absolutize relative search paths. Defaults to :attr:`base` which defaults to ``os.getcwd()``. :param index: Position within the list of search paths. Defaults to last index (appends to the list). The `base` parameter makes it easy to reference files installed along with a python module or package:: res.add_path('./resources/', __file__) """base=os.path.abspath(os.path.dirname(baseorself.base))path=os.path.abspath(os.path.join(base,os.path.dirname(path)))path+=os.sepifpathinself.path:self.path.remove(path)ifcreateandnotos.path.isdir(path):os.makedirs(path)ifindexisNone:self.path.append(path)else:self.path.insert(index,path)self.cache.clear()returnos.path.exists(path)def__iter__(self):""" Iterate over all existing files in all registered paths. """search=self.path[:]whilesearch:path=search.pop()ifnotos.path.isdir(path):continuefornameinos.listdir(path):full=os.path.join(path,name)ifos.path.isdir(full):search.append(full)else:yieldfulldeflookup(self,name):""" Search for a resource and return an absolute file path, or `None`. The :attr:`path` list is searched in order. The first match is returned. Symlinks are followed. The result is cached to speed up future lookups. """ifnamenotinself.cacheorDEBUG:forpathinself.path:fpath=os.path.join(path,name)ifos.path.isfile(fpath):ifself.cachemodein('all','found'):self.cache[name]=fpathreturnfpathifself.cachemode=='all':self.cache[name]=Nonereturnself.cache[name]defopen(self,name,mode='r',*args,**kwargs):""" Find a resource and return a file object, or raise IOError. """fname=self.lookup(name)ifnotfname:raiseIOError("Resource%r not found."%name)returnself.opener(fname,mode=mode,*args,**kwargs)[docs]classFileUpload:[docs]def__init__(self,fileobj,name,filename,headers=None):""" Wrapper for a single file uploaded via ``multipart/form-data``. """#: Open file(-like) object (BytesIO buffer or temporary file)self.file=fileobj#: Name of the upload form fieldself.name=name#: Raw filename as sent by the client (may contain unsafe characters)self.raw_filename=filename#: A :class:`HeaderDict` with additional headers (e.g. content-type)self.headers=HeaderDict(headers)ifheaderselseHeaderDict() content_type=HeaderProperty('Content-Type')content_length=HeaderProperty('Content-Length',reader=int,default=-1)[docs]defget_header(self,name,default=None):""" Return the value of a header within the multipart part. """returnself.headers.get(name,default) [docs]@cached_propertydeffilename(self):""" Name of the file on the client file system, but normalized to ensure file system compatibility. An empty filename is returned as 'empty'. Only ASCII letters, digits, dashes, underscores and dots are allowed in the final filename. Accents are removed, if possible. Whitespace is replaced by a single dash. Leading or tailing dots or dashes are removed. The filename is limited to 255 characters. """fname=self.raw_filenamefname=normalize('NFKD',fname)fname=fname.encode('ASCII','ignore').decode('ASCII')fname=os.path.basename(fname.replace('\\',os.path.sep))fname=re.sub(r'[^a-zA-Z0-9-_.\s]','',fname).strip()fname=re.sub(r'[-\s]+','-',fname).strip('.-')returnfname[:255]or'empty' def_copy_file(self,fp,chunk_size=2**16):read,write,offset=self.file.read,fp.write,self.file.tell()while1:buf=read(chunk_size)ifnotbuf:breakwrite(buf)self.file.seek(offset)[docs]defsave(self,destination,overwrite=False,chunk_size=2**16):""" Save file to disk or copy its content to an open file(-like) object. If *destination* is a directory, :attr:`filename` is added to the path. Existing files are not overwritten by default (IOError). :param destination: File path, directory or file(-like) object. :param overwrite: If True, replace existing files. (default: False) :param chunk_size: Bytes to read at a time. (default: 64kb) """ifisinstance(destination,str):# Except file-likes hereifos.path.isdir(destination):destination=os.path.join(destination,self.filename)ifnotoverwriteandos.path.exists(destination):raiseIOError('File exists.')withopen(destination,'wb')asfp:self._copy_file(fp,chunk_size)else:self._copy_file(destination,chunk_size) ################################################################################ Application Helper ##########################################################################################################################################[docs]defabort(code=500,text='Unknown Error.'):""" Aborts execution and causes a HTTP error. """raiseHTTPError(code,text) [docs]defredirect(url,code=None):""" Aborts execution and causes a 303 or 302 redirect, depending on the HTTP protocol version. """ifnotcode:code=303ifrequest.get('SERVER_PROTOCOL')=="HTTP/1.1"else302res=response.copy(cls=HTTPResponse)res.status=coderes.body=""res.set_header('Location',urljoin(request.url,url))raiseres def_rangeiter(fp,offset,limit,bufsize=1024*1024):""" Yield chunks from a range in a file. """fp.seek(offset)whilelimit>0:part=fp.read(min(limit,bufsize))ifnotpart:breaklimit-=len(part)yieldpart[docs]defstatic_file(filename,root,mimetype=True,download=False,charset='UTF-8',etag=None,headers=None):""" Open a file in a safe way and return an instance of :exc:`HTTPResponse` that can be sent back to the client. :param filename: Name or path of the file to send, relative to ``root``. :param root: Root path for file lookups. Should be an absolute directory path. :param mimetype: Provide the content-type header (default: guess from file extension) :param download: If True, ask the browser to open a `Save as...` dialog instead of opening the file with the associated program. You can specify a custom filename as a string. If not specified, the original filename is used (default: False). :param charset: The charset for files with a ``text/*`` mime-type. (default: UTF-8) :param etag: Provide a pre-computed ETag header. If set to ``False``, ETag handling is disabled. (default: auto-generate ETag header) :param headers: Additional headers dict to add to the response. While checking user input is always a good idea, this function provides additional protection against malicious ``filename`` parameters from breaking out of the ``root`` directory and leaking sensitive information to an attacker. Read-protected files or files outside of the ``root`` directory are answered with ``403 Access Denied``. Missing files result in a ``404 Not Found`` response. Conditional requests (``If-Modified-Since``, ``If-None-Match``) are answered with ``304 Not Modified`` whenever possible. ``HEAD`` and ``Range`` requests (used by download managers to check or continue partial downloads) are also handled automatically. """root=os.path.join(os.path.abspath(root),'')filename=os.path.abspath(os.path.join(root,filename.strip('/\\')))headers=headers.copy()ifheaderselse{}getenv=request.environ.getifnotfilename.startswith(root):returnHTTPError(403,"Access denied.")ifnotos.path.isfile(filename):returnHTTPError(404,"File does not exist.")ifnotos.access(filename,os.R_OK):returnHTTPError(403,"You do not have permission to access this file.")ifmimetypeisTrue:name=downloadifisinstance(download,str)elsefilenamemimetype,encoding=mimetypes.guess_type(name)ifencoding=='gzip':mimetype='application/gzip'elifencoding:# e.g. bzip2 -> application/x-bzip2mimetype='application/x-'+encodingifcharsetandmimetypeand'charset='notinmimetype \
and(mimetype[:5]=='text/'ormimetype=='application/javascript'):mimetype+='; charset=%s'%charsetifmimetype:headers['Content-Type']=mimetypeifdownloadisTrue:download=os.path.basename(filename)ifdownload:download=download.replace('"','')headers['Content-Disposition']='attachment; filename="%s"'%downloadstats=os.stat(filename)headers['Content-Length']=clen=stats.st_sizeheaders['Last-Modified']=email.utils.formatdate(stats.st_mtime,usegmt=True)headers['Date']=email.utils.formatdate(time.time(),usegmt=True)ifetagisNone:etag='%d:%d:%d:%d:%s'%(stats.st_dev,stats.st_ino,stats.st_mtime,clen,filename)etag=hashlib.sha1(tob(etag)).hexdigest()ifetag:headers['ETag']=etagcheck=getenv('HTTP_IF_NONE_MATCH')ifcheckandcheck==etag:returnHTTPResponse(status=304,**headers)ims=getenv('HTTP_IF_MODIFIED_SINCE')ifims:ims=parse_date(ims.split(";")[0].strip())ifimsisnotNoneandims>=int(stats.st_mtime):returnHTTPResponse(status=304,**headers)body=''ifrequest.method=='HEAD'elseopen(filename,'rb')headers["Accept-Ranges"]="bytes"range_header=getenv('HTTP_RANGE')ifrange_header:ranges=list(parse_range_header(range_header,clen))ifnotranges:returnHTTPError(416,"Requested Range Not Satisfiable")offset,end=ranges[0]rlen=end-offsetheaders["Content-Range"]="bytes%d-%d/%d"%(offset,end-1,clen)headers["Content-Length"]=str(rlen)ifbody:body=_closeiter(_rangeiter(body,offset,rlen),body.close)returnHTTPResponse(body,status=206,**headers)returnHTTPResponse(body,**headers) ################################################################################ HTTP Utilities and MISC (TODO) ##############################################################################################################################[docs]defdebug(mode=True):""" Change the debug level. There is only one debug level supported at the moment."""globalDEBUGifmode:warnings.simplefilter('default')DEBUG=bool(mode) defhttp_date(value):ifisinstance(value,str):returnvalueifisinstance(value,datetime):# aware datetime.datetime is converted to UTC time# naive datetime.datetime is treated as UTC timevalue=value.utctimetuple()elifisinstance(value,datedate):# datetime.date is naive, and is treated as UTC timevalue=value.timetuple()ifnotisinstance(value,(int,float)):# convert struct_time in UTC to UNIX timestampvalue=calendar.timegm(value)returnemail.utils.formatdate(value,usegmt=True)[docs]defparse_date(ims):""" Parse rfc1123, rfc850 and asctime timestamps and return UTC epoch. """try:ts=email.utils.parsedate_tz(ims)returncalendar.timegm(ts[:8]+(0,))-(ts[9]or0)except(TypeError,ValueError,IndexError,OverflowError):returnNone [docs]defparse_auth(header):""" Parse rfc2617 HTTP authentication header string (basic) and return (user,pass) tuple or None"""try:method,data=header.split(None,1)ifmethod.lower()=='basic':user,pwd=touni(base64.b64decode(tob(data))).split(':',1)returnuser,pwdexcept(KeyError,ValueError):returnNone defparse_range_header(header,maxlen=0):""" Yield (start, end) ranges parsed from a HTTP Range header. Skip unsatisfiable ranges. The end index is non-inclusive."""ifnotheaderorheader[:6]!='bytes=':returnranges=[r.split('-',1)forrinheader[6:].split(',')if'-'inr]forstart,endinranges:try:ifnotstart:# bytes=-100 -> last 100 bytesstart,end=max(0,maxlen-int(end)),maxlenelifnotend:# bytes=100- -> all but the first 99 bytesstart,end=int(start),maxlenelse:# bytes=100-200 -> bytes 100-200 (inclusive)start,end=int(start),min(int(end)+1,maxlen)if0<=start<end<=maxlen:yieldstart,endexceptValueError:pass#: Header tokenizer used by _parse_http_header()_hsplit=re.compile('(?:(?:"((?:[^"\\\\]|\\\\.)*)")|([^;,=]+))([;,=]?)').findalldef_parse_http_header(h):""" Parses a typical multi-valued and parametrised HTTP header (e.g. Accept headers) and returns a list of values and parameters. For non-standard or broken input, this implementation may return partial results. :param h: A header string (e.g. ``text/html,text/plain;q=0.9,*/*;q=0.8``) :return: List of (value, params) tuples. The second element is a (possibly empty) dict. """values=[]if'"'notinh:# INFO: Fast path without regexp (~2x faster)forvalueinh.split(','):parts=value.split(';')values.append((parts[0].strip(),{}))forattrinparts[1:]:name,value=attr.split('=',1)values[-1][1][name.strip().lower()]=value.strip()else:lop,key,attrs=',',None,{}forquoted,plain,tokin_hsplit(h):value=plain.strip()ifplainelsequoted.replace('\\"','"')iflop==',':attrs={}values.append((value,attrs))eliflop==';':iftok=='=':key=valueelse:attrs[value.strip().lower()]=''eliflop=='='andkey:attrs[key.strip().lower()]=valuekey=Nonelop=tokreturnvaluesdef_parse_qsl(qs,encoding="utf8"):r=[]forpairinqs.split('&'):ifnotpair:continuenv=pair.split('=',1)iflen(nv)!=2:nv.append('')key=urlunquote(nv[0].replace('+',' '),encoding)value=urlunquote(nv[1].replace('+',' '),encoding)r.append((key,value))returnrdef_lscmp(a,b):""" Compares two strings in a cryptographically safe way: Runtime is not affected by length of common prefix. """returnnotsum(0ifx==yelse1forx,yinzip(a,b))andlen(a)==len(b)[docs]defcookie_encode(data,key,digestmod=None):""" Encode and sign a pickle-able object. Return a (byte) string """depr(0,13,"cookie_encode() will be removed soon.","Do not use this API directly.")digestmod=digestmodorhashlib.sha256msg=base64.b64encode(pickle.dumps(data,-1))sig=base64.b64encode(hmac.new(tob(key),msg,digestmod=digestmod).digest())returnb'!'+sig+b'?'+msg [docs]defcookie_decode(data,key,digestmod=None):""" Verify and decode an encoded string. Return an object or None."""depr(0,13,"cookie_decode() will be removed soon.","Do not use this API directly.")data=tob(data)ifcookie_is_encoded(data):sig,msg=data.split(b'?',1)digestmod=digestmodorhashlib.sha256hashed=hmac.new(tob(key),msg,digestmod=digestmod).digest()if_lscmp(sig[1:],base64.b64encode(hashed)):returnpickle.loads(base64.b64decode(msg))returnNone [docs]defcookie_is_encoded(data):""" Return True if the argument looks like a encoded cookie."""depr(0,13,"cookie_is_encoded() will be removed soon.","Do not use this API directly.")returnbool(data.startswith(b'!')andb'?'indata) defhtml_escape(string):""" Escape HTML special characters ``&<>`` and quotes ``'"``. """returnstring.replace('&','&').replace('<','<').replace('>','>')\.replace('"','"').replace("'",''')defhtml_quote(string):""" Escape and quote a string to be used as an HTTP attribute."""return'"%s"'%html_escape(string).replace('\n',' ') \.replace('\r',' ').replace('\t','	')[docs]defyieldroutes(func):""" Return a generator for routes that match the signature (name, args) of the func parameter. This may yield more than one route if the function takes optional keyword arguments. The output is best described by example:: a() -> '/a' b(x, y) -> '/b/<x>/<y>' c(x, y=5) -> '/c/<x>' and '/c/<x>/<y>' d(x=5, y=6) -> '/d' and '/d/<x>' and '/d/<x>/<y>' """path='/'+func.__name__.replace('__','/').lstrip('/')sig=inspect.signature(func,follow_wrapped=False)forpinsig.parameters.values():ifp.kind==p.POSITIONAL_ONLY:raiseValueError("Invalid signature for yieldroutes:%s"%sig)ifp.kindin(p.POSITIONAL_OR_KEYWORD,p.KEYWORD_ONLY):ifp.default!=p.empty:yieldpath# Yield path without this (optional) parameter.path+="/<%s>"%p.nameyieldpath [docs]defpath_shift(script_name,path_info,shift=1):""" Shift path fragments from PATH_INFO to SCRIPT_NAME and vice versa. :return: The modified paths. :param script_name: The SCRIPT_NAME path. :param script_name: The PATH_INFO path. :param shift: The number of path fragments to shift. May be negative to change the shift direction. (default: 1) """ifshift==0:returnscript_name,path_infopathlist=path_info.strip('/').split('/')scriptlist=script_name.strip('/').split('/')ifpathlistandpathlist[0]=='':pathlist=[]ifscriptlistandscriptlist[0]=='':scriptlist=[]if0<shift<=len(pathlist):moved=pathlist[:shift]scriptlist=scriptlist+movedpathlist=pathlist[shift:]elif0>shift>=-len(scriptlist):moved=scriptlist[shift:]pathlist=moved+pathlistscriptlist=scriptlist[:shift]else:empty='SCRIPT_NAME'ifshift<0else'PATH_INFO'raiseAssertionError("Cannot shift. Nothing left from%s"%empty)new_script_name='/'+'/'.join(scriptlist)new_path_info='/'+'/'.join(pathlist)ifpath_info.endswith('/')andpathlist:new_path_info+='/'returnnew_script_name,new_path_info defauth_basic(check,realm="private",text="Access denied"):""" Callback decorator to require HTTP auth (basic). TODO: Add route(check_auth=...) parameter. """defdecorator(func):@functools.wraps(func)defwrapper(*a,**ka):user,password=request.author(None,None)ifuserisNoneornotcheck(user,password):err=HTTPError(401,text)err.add_header('WWW-Authenticate','Basic realm="%s"'%realm)returnerrreturnfunc(*a,**ka)returnwrapperreturndecorator# Shortcuts for common Bottle methods.# They all refer to the current default application.defmake_default_app_wrapper(name):""" Return a callable that relays calls to the current default app. """@functools.wraps(getattr(Bottle,name))defwrapper(*a,**ka):returngetattr(app(),name)(*a,**ka)returnwrapperroute=make_default_app_wrapper('route')get=make_default_app_wrapper('get')post=make_default_app_wrapper('post')put=make_default_app_wrapper('put')delete=make_default_app_wrapper('delete')patch=make_default_app_wrapper('patch')error=make_default_app_wrapper('error')mount=make_default_app_wrapper('mount')hook=make_default_app_wrapper('hook')install=make_default_app_wrapper('install')uninstall=make_default_app_wrapper('uninstall')url=make_default_app_wrapper('get_url')################################################################################ Multipart Handling ########################################################################################################################################### cgi.FieldStorage was deprecated in Python 3.11 and removed in 3.13# This implementation is based on https://github.com/defnull/multipart/classMultipartError(HTTPError):def__init__(self,msg):HTTPError.__init__(self,400,"MultipartError: "+msg)class_MultipartParser:def__init__(self,stream,boundary,content_length=-1,disk_limit=2**30,mem_limit=2**20,memfile_limit=2**18,buffer_size=2**16,charset="latin1",):self.stream=streamself.boundary=boundaryself.content_length=content_lengthself.disk_limit=disk_limitself.memfile_limit=memfile_limitself.mem_limit=min(mem_limit,self.disk_limit)self.buffer_size=min(buffer_size,self.mem_limit)self.charset=charsetifnotboundary:raiseMultipartError("No boundary.")ifself.buffer_size-6<len(boundary):# "--boundary--\r\n"raiseMultipartError("Boundary does not fit into buffer_size.")def_lineiter(self):""" Iterate over a binary file-like object (crlf terminated) line by line. Each line is returned as a (line, crlf) tuple. Lines larger than buffer_size are split into chunks where all but the last chunk has an empty string instead of crlf. Maximum chunk size is twice the buffer size. """read=self.stream.readmaxread,maxbuf=self.content_length,self.buffer_sizepartial=b""# Contains the last (partial) linewhileTrue:chunk=read(maxbufifmaxread<0elsemin(maxbuf,maxread))maxread-=len(chunk)ifnotchunk:ifpartial:yieldpartial,b''breakifpartial:chunk=partial+chunkscanpos=0whileTrue:i=chunk.find(b'\r\n',scanpos)ifi>=0:yieldchunk[scanpos:i],b'\r\n'scanpos=i+2else:# CRLF not foundpartial=chunk[scanpos:]ifscanposelsechunkbreakiflen(partial)>maxbuf:yieldpartial[:-1],b""partial=partial[-1:]defparse(self):""" Return a MultiPart iterator. Can only be called once. """lines,line=self._lineiter(),""separator=b"--"+tob(self.boundary)terminator=separator+b"--"mem_used,disk_used=0,0# Track used resources to prevent DoSis_tail=False# True if the last line was incomplete (cutted)# Consume first boundary. Ignore any preamble, as required by RFC# 2046, section 5.1.1.forline,nlinlines:iflinein(separator,terminator):breakelse:raiseMultipartError("Stream does not contain boundary")# First line is termainating boundary -> empty multipart streamifline==terminator:for_inlines:raiseMultipartError("Found data after empty multipart stream")returnpart_options={"buffer_size":self.buffer_size,"memfile_limit":self.memfile_limit,"charset":self.charset,}part=_MultipartPart(**part_options)forline,nlinlines:ifnotis_tailand(line==separatororline==terminator):part.finish()ifpart.is_buffered():mem_used+=part.sizeelse:disk_used+=part.sizeyieldpartifline==terminator:breakpart=_MultipartPart(**part_options)else:is_tail=notnl# The next line continues this onetry:part.feed(line,nl)ifpart.is_buffered():ifpart.size+mem_used>self.mem_limit:raiseMultipartError("Memory limit reached.")elifpart.size+disk_used>self.disk_limit:raiseMultipartError("Disk limit reached.")exceptMultipartError:part.close()raiseelse:part.close()ifline!=terminator:raiseMultipartError("Unexpected end of multipart stream.")class_MultipartPart:def__init__(self,buffer_size=2**16,memfile_limit=2**18,charset="latin1"):self.headerlist=[]self.headers=Noneself.file=Falseself.size=0self._buf=b""self.disposition=Noneself.name=Noneself.filename=Noneself.content_type=Noneself.charset=charsetself.memfile_limit=memfile_limitself.buffer_size=buffer_sizedeffeed(self,line,nl=""):ifself.file:returnself.write_body(line,nl)returnself.write_header(line,nl)defwrite_header(self,line,nl):line=str(line,self.charset)ifnotnl:raiseMultipartError("Unexpected end of line in header.")ifnotline.strip():# blank line -> end of header segmentself.finish_header()elifline[0]in"\t"andself.headerlist:name,value=self.headerlist.pop()self.headerlist.append((name,value+line.strip()))else:if":"notinline:raiseMultipartError("Syntax error in header: No colon.")name,value=line.split(":",1)self.headerlist.append((name.strip(),value.strip()))defwrite_body(self,line,nl):ifnotlineandnotnl:return# This does not even flush the bufferself.size+=len(line)+len(self._buf)self.file.write(self._buf+line)self._buf=nlifself.content_length>0andself.size>self.content_length:raiseMultipartError("Size of body exceeds Content-Length header.")ifself.size>self.memfile_limitandisinstance(self.file,BytesIO):self.file,old=NamedTemporaryFile(mode="w+b"),self.fileold.seek(0)copied,maxcopy,chunksize=0,self.size,self.buffer_sizeread,write=old.read,self.file.writewhilecopied<maxcopy:chunk=read(min(chunksize,maxcopy-copied))write(chunk)copied+=len(chunk)deffinish_header(self):self.file=BytesIO()self.headers=HeaderDict(self.headerlist)content_disposition=self.headers.get("Content-Disposition")content_type=self.headers.get("Content-Type")ifnotcontent_disposition:raiseMultipartError("Content-Disposition header is missing.")self.disposition,self.options=_parse_http_header(content_disposition)[0]self.name=self.options.get("name")if"filename"inself.options:self.filename=self.options.get("filename")ifself.filename[1:3]==":\\"orself.filename[:2]=="\\\\":self.filename=self.filename.split("\\")[-1]# ie6 bugself.content_type,options=_parse_http_header(content_type)[0]ifcontent_typeelse(None,{})self.charset=options.get("charset")orself.charsetself.content_length=int(self.headers.get("Content-Length","-1"))deffinish(self):ifnotself.file:raiseMultipartError("Incomplete part: Header section not closed.")self.file.seek(0)defis_buffered(self):""" Return true if the data is fully buffered in memory."""returnisinstance(self.file,BytesIO)@propertydefvalue(self):""" Data decoded with the specified charset """returnstr(self.raw,self.charset)@propertydefraw(self):""" Data without decoding """pos=self.file.tell()self.file.seek(0)try:returnself.file.read()finally:self.file.seek(pos)defclose(self):ifself.file:self.file.close()self.file=False################################################################################ Server Adapter ############################################################################################################################################### Before you edit or add a server adapter, please read:# - https://github.com/bottlepy/bottle/pull/647#issuecomment-60152870# - https://github.com/bottlepy/bottle/pull/865#issuecomment-242795341classServerAdapter:quiet=Falsedef__init__(self,host='127.0.0.1',port=8080,**options):self.options=optionsself.host=hostself.port=int(port)defrun(self,handler):# pragma: no coverpass@propertydef_listen_url(self):ifself.host.startswith("unix:"):returnself.hostelif':'inself.host:return"http://[%s]:%d/"%(self.host,self.port)else:return"http://%s:%d/"%(self.host,self.port)def__repr__(self):args=', '.join('%s=%r'%kvforkvinself.options.items())return"%s(%s)"%(self.__class__.__name__,args)classCGIServer(ServerAdapter):quiet=Truedefrun(self,handler):# pragma: no coverfromwsgiref.handlersimportCGIHandlerdeffixed_environ(environ,start_response):environ.setdefault('PATH_INFO','')returnhandler(environ,start_response)CGIHandler().run(fixed_environ)classFlupFCGIServer(ServerAdapter):defrun(self,handler):# pragma: no coverimportflup.server.fcgiself.options.setdefault('bindAddress',(self.host,self.port))flup.server.fcgi.WSGIServer(handler,**self.options).run()classWSGIRefServer(ServerAdapter):defrun(self,app):# pragma: no coverfromwsgiref.simple_serverimportmake_serverfromwsgiref.simple_serverimportWSGIRequestHandler,WSGIServerimportsocketclassFixedHandler(WSGIRequestHandler):deflog_message(other,format,*args):ifnotself.quiet:returnWSGIRequestHandler.log_message(other,format,*args)handler_cls=self.options.get('handler_class',FixedHandler)server_cls=self.options.get('server_class',WSGIServer)if':'inself.host:# Fix wsgiref for IPv6 addresses.ifgetattr(server_cls,'address_family')==socket.AF_INET:classserver_cls(server_cls):address_family=socket.AF_INET6self.srv=make_server(self.host,self.port,app,server_cls,handler_cls)self.port=self.srv.server_port# update port actual port (0 means random)try:self.srv.serve_forever()exceptKeyboardInterrupt:self.srv.server_close()# Prevent ResourceWarning: unclosed socketraiseclassCherryPyServer(ServerAdapter):defrun(self,handler):# pragma: no coverdepr(0,13,"The wsgi server part of cherrypy was split into a new ""project called 'cheroot'.","Use the 'cheroot' server ""adapter instead of cherrypy.")fromcherrypyimportwsgiserver# This will fail for CherryPy >= 9self.options['bind_addr']=(self.host,self.port)self.options['wsgi_app']=handlercertfile=self.options.get('certfile')ifcertfile:delself.options['certfile']keyfile=self.options.get('keyfile')ifkeyfile:delself.options['keyfile']server=wsgiserver.CherryPyWSGIServer(**self.options)ifcertfile:server.ssl_certificate=certfileifkeyfile:server.ssl_private_key=keyfiletry:server.start()finally:server.stop()classCherootServer(ServerAdapter):defrun(self,handler):# pragma: no coverfromcherootimportwsgifromcheroot.sslimportbuiltinself.options['bind_addr']=(self.host,self.port)self.options['wsgi_app']=handlercertfile=self.options.pop('certfile',None)keyfile=self.options.pop('keyfile',None)chainfile=self.options.pop('chainfile',None)server=wsgi.Server(**self.options)ifcertfileandkeyfile:server.ssl_adapter=builtin.BuiltinSSLAdapter(certfile,keyfile,chainfile)try:server.start()finally:server.stop()classWaitressServer(ServerAdapter):defrun(self,handler):fromwaitressimportserveserve(handler,host=self.host,port=self.port,_quiet=self.quiet,**self.options)classPasteServer(ServerAdapter):defrun(self,handler):# pragma: no coverfrompasteimporthttpserverfrompaste.transloggerimportTransLoggerhandler=TransLogger(handler,setup_console_handler=(notself.quiet))httpserver.serve(handler,host=self.host,port=str(self.port),**self.options)classMeinheldServer(ServerAdapter):defrun(self,handler):frommeinheldimportserverserver.listen((self.host,self.port))server.run(handler)classFapwsServer(ServerAdapter):""" Extremely fast webserver using libev. See https://github.com/william-os4y/fapws3 """defrun(self,handler):# pragma: no coverdepr(0,13,"fapws3 is not maintained and support will be dropped.")importfapws._evwsgiasevwsgifromfapwsimportbase,configport=self.portiffloat(config.SERVER_IDENT[-2:])>0.4:# fapws3 silently changed its API in 0.5port=str(port)evwsgi.start(self.host,port)# fapws3 never releases the GIL. Complain upstream. I tried. No luck.if'BOTTLE_CHILD'inos.environandnotself.quiet:_stderr("WARNING: Auto-reloading does not work with Fapws3.")_stderr(" (Fapws3 breaks python thread support)")evwsgi.set_base_module(base)defapp(environ,start_response):environ['wsgi.multiprocess']=Falsereturnhandler(environ,start_response)evwsgi.wsgi_cb(('',app))evwsgi.run()classTornadoServer(ServerAdapter):""" The super hyped asynchronous server by facebook. Untested. """defrun(self,handler):# pragma: no coverimporttornado.wsgi,tornado.httpserver,tornado.ioloopcontainer=tornado.wsgi.WSGIContainer(handler)server=tornado.httpserver.HTTPServer(container)server.listen(port=self.port,address=self.host)tornado.ioloop.IOLoop.instance().start()classAppEngineServer(ServerAdapter):""" Adapter for Google App Engine. """quiet=Truedefrun(self,handler):depr(0,13,"AppEngineServer no longer required","Configure your application directly in your app.yaml")fromgoogle.appengine.ext.webappimportutil# A main() function in the handler script enables 'App Caching'.# Lets makes sure it is there. This _really_ improves performance.module=sys.modules.get('__main__')ifmoduleandnothasattr(module,'main'):module.main=lambda:util.run_wsgi_app(handler)util.run_wsgi_app(handler)classTwistedServer(ServerAdapter):""" Untested. """defrun(self,handler):fromtwisted.webimportserver,wsgifromtwisted.python.threadpoolimportThreadPoolfromtwisted.internetimportreactorthread_pool=ThreadPool()thread_pool.start()reactor.addSystemEventTrigger('after','shutdown',thread_pool.stop)factory=server.Site(wsgi.WSGIResource(reactor,thread_pool,handler))reactor.listenTCP(self.port,factory,interface=self.host)ifnotreactor.running:reactor.run()classDieselServer(ServerAdapter):""" Untested. """defrun(self,handler):depr(0,13,"Diesel is not tested or supported and will be removed.")fromdiesel.protocols.wsgiimportWSGIApplicationapp=WSGIApplication(handler,port=self.port)app.run()classGeventServer(ServerAdapter):""" Untested. Options: * See gevent.wsgi.WSGIServer() documentation for more options. """defrun(self,handler):fromgeventimportpywsgi,localifnotisinstance(threading.local(),local.local):msg="Bottle requires gevent.monkey.patch_all() (before import)"raiseRuntimeError(msg)ifself.quiet:self.options['log']=Noneaddress=(self.host,self.port)server=pywsgi.WSGIServer(address,handler,**self.options)if'BOTTLE_CHILD'inos.environ:importsignalsignal.signal(signal.SIGINT,lambdas,f:server.stop())server.serve_forever()classGunicornServer(ServerAdapter):""" Untested. See http://gunicorn.org/configure.html for options. """defrun(self,handler):fromgunicorn.app.baseimportBaseApplicationifself.host.startswith("unix:"):config={'bind':self.host}else:config={'bind':"%s:%d"%(self.host,self.port)}config.update(self.options)classGunicornApplication(BaseApplication):defload_config(self):forkey,valueinconfig.items():self.cfg.set(key,value)defload(self):returnhandlerGunicornApplication().run()classEventletServer(ServerAdapter):""" Untested. Options: * `backlog` adjust the eventlet backlog parameter which is the maximum number of queued connections. Should be at least 1; the maximum value is system-dependent. * `family`: (default is 2) socket family, optional. See socket documentation for available families. """defrun(self,handler):fromeventletimportwsgi,listen,patcherifnotpatcher.is_monkey_patched(os):msg="Bottle requires eventlet.monkey_patch() (before import)"raiseRuntimeError(msg)socket_args={}forargin('backlog','family'):try:socket_args[arg]=self.options.pop(arg)exceptKeyError:passaddress=(self.host,self.port)try:wsgi.server(listen(address,**socket_args),handler,log_output=(notself.quiet))exceptTypeError:# Fallback, if we have old version of eventletwsgi.server(listen(address),handler)classBjoernServer(ServerAdapter):""" Fast server written in C: https://github.com/jonashaag/bjoern """defrun(self,handler):frombjoernimportrunrun(handler,self.host,self.port,reuse_port=True)classAsyncioServerAdapter(ServerAdapter):""" Extend ServerAdapter for adding custom event loop """defget_event_loop(self):passclassAiohttpServer(AsyncioServerAdapter):""" Asynchronous HTTP client/server framework for asyncio https://pypi.python.org/pypi/aiohttp/ https://pypi.org/project/aiohttp-wsgi/ """defget_event_loop(self):importasyncioreturnasyncio.new_event_loop()defrun(self,handler):importasynciofromaiohttp_wsgi.wsgiimportserveself.loop=self.get_event_loop()asyncio.set_event_loop(self.loop)if'BOTTLE_CHILD'inos.environ:importsignalsignal.signal(signal.SIGINT,lambdas,f:self.loop.stop())serve(handler,host=self.host,port=self.port)classAiohttpUVLoopServer(AiohttpServer):"""uvloop https://github.com/MagicStack/uvloop """defget_event_loop(self):importuvloopreturnuvloop.new_event_loop()classAutoServer(ServerAdapter):""" Untested. """adapters=[WaitressServer,PasteServer,TwistedServer,CherryPyServer,CherootServer,WSGIRefServer]defrun(self,handler):forsainself.adapters:try:returnsa(self.host,self.port,**self.options).run(handler)exceptImportError:passserver_names={'cgi':CGIServer,'flup':FlupFCGIServer,'wsgiref':WSGIRefServer,'waitress':WaitressServer,'cherrypy':CherryPyServer,'cheroot':CherootServer,'paste':PasteServer,'fapws3':FapwsServer,'tornado':TornadoServer,'gae':AppEngineServer,'twisted':TwistedServer,'diesel':DieselServer,'meinheld':MeinheldServer,'gunicorn':GunicornServer,'eventlet':EventletServer,'gevent':GeventServer,'bjoern':BjoernServer,'aiohttp':AiohttpServer,'uvloop':AiohttpUVLoopServer,'auto':AutoServer,}################################################################################ Application Control #########################################################################################################################################[docs]defload(target,**namespace):""" Import a module or fetch an object from a module. * ``package.module`` returns `module` as a module object. * ``pack.mod:name`` returns the module variable `name` from `pack.mod`. * ``pack.mod:func()`` calls `pack.mod.func()` and returns the result. The last form accepts not only function calls, but any type of expression. Keyword arguments passed to this function are available as local variables. Example: ``import_string('re:compile(x)', x='[a-z]')`` """module,target=target.split(":",1)if':'intargetelse(target,None)ifmodulenotinsys.modules:__import__(module)ifnottarget:returnsys.modules[module]iftarget.isalnum():returngetattr(sys.modules[module],target)package_name=module.split('.')[0]namespace[package_name]=sys.modules[package_name]returneval('%s.%s'%(module,target),namespace) [docs]defload_app(target):""" Load a bottle application from a module and make sure that the import does not affect the current default application, but returns a separate application object. See :func:`load` for the target parameter. """globalNORUNNORUN,nr_old=True,NORUNtmp=default_app.push()# Create a new "default application"try:rv=load(target)# Import the target modulereturnrvifcallable(rv)elsetmpfinally:default_app.remove(tmp)# Remove the temporary added default applicationNORUN=nr_old _debug=debug[docs]defrun(app=None,server='wsgiref',host='127.0.0.1',port=8080,interval=1,reloader=False,quiet=False,plugins=None,debug=None,config=None,**kargs):""" Start a server instance. This method blocks until the server terminates. :param app: WSGI application or target string supported by :func:`load_app`. (default: :func:`default_app`) :param server: Server adapter to use. See :data:`server_names` keys for valid names or pass a :class:`ServerAdapter` subclass. (default: `wsgiref`) :param host: Server address to bind to. Pass ``0.0.0.0`` to listens on all interfaces including the external one. (default: 127.0.0.1) :param port: Server port to bind to. Values below 1024 require root privileges. (default: 8080) :param reloader: Start auto-reloading server? (default: False) :param interval: Auto-reloader interval in seconds (default: 1) :param quiet: Suppress output to stdout and stderr? (default: False) :param options: Options passed to the server adapter. """ifNORUN:returnifreloaderandnotos.environ.get('BOTTLE_CHILD'):importsubprocessfd,lockfile=tempfile.mkstemp(prefix='bottle.',suffix='.lock')environ=os.environ.copy()environ['BOTTLE_CHILD']='true'environ['BOTTLE_LOCKFILE']=lockfileargs=[sys.executable]+sys.argv# If a package was loaded with `python -m`, then `sys.argv` needs to be# restored to the original value, or imports might break. See #1336ifgetattr(sys.modules.get('__main__'),'__package__',None):args[1:1]=["-m",sys.modules['__main__'].__package__]try:os.close(fd)# We never write to this filewhileos.path.exists(lockfile):p=subprocess.Popen(args,env=environ)whilep.poll()isNone:os.utime(lockfile,None)# Tell child we are still alivetime.sleep(interval)ifp.returncode==3:# Child wants to be restartedcontinuesys.exit(p.returncode)exceptKeyboardInterrupt:passfinally:ifos.path.exists(lockfile):os.unlink(lockfile)returntry:ifdebugisnotNone:_debug(debug)app=appordefault_app()ifisinstance(app,str):app=load_app(app)ifnotcallable(app):raiseValueError("Application is not callable:%r"%app)forplugininpluginsor[]:ifisinstance(plugin,str):plugin=load(plugin)app.install(plugin)ifconfig:app.config.update(config)ifserverinserver_names:server=server_names.get(server)ifisinstance(server,str):server=load(server)ifisinstance(server,type):server=server(host=host,port=port,**kargs)ifnotisinstance(server,ServerAdapter):raiseValueError("Unknown or unsupported server:%r"%server)server.quiet=server.quietorquietifnotserver.quiet:_stderr("Bottle v%s server starting up (using%s)..."%(__version__,repr(server)))_stderr("Listening on%s"%server._listen_url)_stderr("Hit Ctrl-C to quit.\n")ifreloader:lockfile=os.environ.get('BOTTLE_LOCKFILE')bgcheck=FileCheckerThread(lockfile,interval)withbgcheck:server.run(app)ifbgcheck.status=='reload':sys.exit(3)else:server.run(app)exceptKeyboardInterrupt:passexcept(SystemExit,MemoryError):raiseexcept:# noqa: E722ifnotreloader:raiseifnotgetattr(server,'quiet',quiet):print_exc()time.sleep(interval)sys.exit(3) classFileCheckerThread(threading.Thread):""" Interrupt main-thread as soon as a changed module file is detected, the lockfile gets deleted or gets too old. """def__init__(self,lockfile,interval):threading.Thread.__init__(self)self.daemon=Trueself.lockfile,self.interval=lockfile,interval#: Is one of 'reload', 'error' or 'exit'self.status=Nonedefrun(self):exists=os.path.existsmtime=lambdap:os.stat(p).st_mtimefiles={}formoduleinlist(sys.modules.values()):path=getattr(module,'__file__','')or''ifpath[-4:]in('.pyo','.pyc'):path=path[:-1]ifpathandexists(path):files[path]=mtime(path)whilenotself.status:ifnotexists(self.lockfile)\ormtime(self.lockfile)<time.time()-self.interval-5:self.status='error'thread.interrupt_main()forpath,lmtimeinlist(files.items()):ifnotexists(path)ormtime(path)>lmtime:self.status='reload'thread.interrupt_main()breaktime.sleep(self.interval)def__enter__(self):self.start()def__exit__(self,exc_type,*_):ifnotself.status:self.status='exit'# silent exitself.join()returnexc_typeisnotNoneandissubclass(exc_type,KeyboardInterrupt)################################################################################ Template Adapters ###########################################################################################################################################classTemplateError(BottleException):pass[docs]classBaseTemplate:""" Base class and minimal API for template adapters """extensions=['tpl','html','thtml','stpl']settings={}# used in prepare()defaults={}# used in render()[docs]def__init__(self,source=None,name=None,lookup=None,encoding='utf8',**settings):""" Create a new template. If the source parameter (str or buffer) is missing, the name argument is used to guess a template filename. Subclasses can assume that self.source and/or self.filename are set. Both are strings. The lookup, encoding and settings parameters are stored as instance variables. The lookup parameter stores a list containing directory paths. The encoding parameter should be used to decode byte strings or files. The settings parameter contains a dict for engine-specific settings. """self.name=nameself.source=source.read()ifhasattr(source,'read')elsesourceself.filename=source.filenameifhasattr(source,'filename')elseNoneself.lookup=[os.path.abspath(x)forxinlookup]iflookupelse[]self.encoding=encodingself.settings=self.settings.copy()# Copy from class variableself.settings.update(settings)# Applyifnotself.sourceandself.name:self.filename=self.search(self.name,self.lookup)ifnotself.filename:raiseTemplateError('Template%s not found.'%repr(name))ifnotself.sourceandnotself.filename:raiseTemplateError('No template specified.')self.prepare(**self.settings) [docs]@classmethoddefsearch(cls,name,lookup=None):""" Search name in all directories specified in lookup. First without, then with common extensions. Return first hit. """ifnotlookup:raisedepr(0,12,"Empty template lookup path.","Configure a template lookup path.")ifos.path.isabs(name):raisedepr(0,12,"Use of absolute path for template name.","Refer to templates with names or paths relative to the lookup path.")forspathinlookup:spath=os.path.abspath(spath)+os.sepfname=os.path.abspath(os.path.join(spath,name))ifnotfname.startswith(spath):continueifos.path.isfile(fname):returnfnameforextincls.extensions:ifos.path.isfile('%s.%s'%(fname,ext)):return'%s.%s'%(fname,ext) [docs]@classmethoddefglobal_config(cls,key,*args):""" This reads or sets the global settings stored in class.settings. """ifargs:cls.settings=cls.settings.copy()# Make settings local to classcls.settings[key]=args[0]else:returncls.settings[key] [docs]defprepare(self,**options):""" Run preparations (parsing, caching, ...). It should be possible to call this again to refresh a template or to update settings. """raiseNotImplementedError [docs]defrender(self,*args,**kwargs):""" Render the template with the specified local variables and return a single byte or unicode string. If it is a byte string, the encoding must match self.encoding. This method must be thread-safe! Local variables may be provided in dictionaries (args) or directly, as keywords (kwargs). """raiseNotImplementedError classMakoTemplate(BaseTemplate):defprepare(self,**options):frommako.templateimportTemplatefrommako.lookupimportTemplateLookupoptions.update({'input_encoding':self.encoding})options.setdefault('format_exceptions',bool(DEBUG))lookup=TemplateLookup(directories=self.lookup,**options)ifself.source:self.tpl=Template(self.source,lookup=lookup,**options)else:self.tpl=Template(uri=self.name,filename=self.filename,lookup=lookup,**options)defrender(self,*args,**kwargs):fordictarginargs:kwargs.update(dictarg)_defaults=self.defaults.copy()_defaults.update(kwargs)returnself.tpl.render(**_defaults)classCheetahTemplate(BaseTemplate):defprepare(self,**options):fromCheetah.TemplateimportTemplateself.context=threading.local()self.context.vars={}options['searchList']=[self.context.vars]ifself.source:self.tpl=Template(source=self.source,**options)else:self.tpl=Template(file=self.filename,**options)defrender(self,*args,**kwargs):fordictarginargs:kwargs.update(dictarg)self.context.vars.update(self.defaults)self.context.vars.update(kwargs)out=str(self.tpl)self.context.vars.clear()returnoutclassJinja2Template(BaseTemplate):defprepare(self,filters=None,tests=None,globals={},**kwargs):fromjinja2importEnvironment,FunctionLoaderself.env=Environment(loader=FunctionLoader(self.loader),**kwargs)iffilters:self.env.filters.update(filters)iftests:self.env.tests.update(tests)ifglobals:self.env.globals.update(globals)ifself.source:self.tpl=self.env.from_string(self.source)else:self.tpl=self.env.get_template(self.name)defrender(self,*args,**kwargs):fordictarginargs:kwargs.update(dictarg)_defaults=self.defaults.copy()_defaults.update(kwargs)returnself.tpl.render(**_defaults)defloader(self,name):ifname==self.filename:fname=nameelse:fname=self.search(name,self.lookup)ifnotfname:returnwithopen(fname,"rb")asf:return(f.read().decode(self.encoding),fname,lambda:False)[docs]classSimpleTemplate(BaseTemplate):[docs]defprepare(self,escape_func=html_escape,noescape=False,syntax=None,**ka):self.cache={}enc=self.encodingself._str=lambdax:touni(x,enc)self._escape=lambdax:escape_func(touni(x,enc))self.syntax=syntaxifnoescape:self._str,self._escape=self._escape,self._str @cached_propertydefco(self):returncompile(self.code,self.filenameor'<string>','exec')@cached_propertydefcode(self):source=self.sourceifnotsource:withopen(self.filename,'rb')asf:source=f.read()try:source,encoding=touni(source),'utf8'exceptUnicodeError:raisedepr(0,11,'Unsupported template encodings.','Use utf-8 for templates.')parser=StplParser(source,encoding=encoding,syntax=self.syntax)code=parser.translate()self.encoding=parser.encodingreturncodedef_rebase(self,_env,_name=None,**kwargs):_env['_rebase']=(_name,kwargs)def_include(self,_env,_name=None,**kwargs):env=_env.copy()env.update(kwargs)if_namenotinself.cache:self.cache[_name]=self.__class__(name=_name,lookup=self.lookup,syntax=self.syntax)returnself.cache[_name].execute(env['_stdout'],env)defexecute(self,_stdout,kwargs):env=self.defaults.copy()env.update(kwargs)env.update({'_stdout':_stdout,'_printlist':_stdout.extend,'include':functools.partial(self._include,env),'rebase':functools.partial(self._rebase,env),'_rebase':None,'_str':self._str,'_escape':self._escape,'get':env.get,'setdefault':env.setdefault,'defined':env.__contains__})exec(self.co,env)ifenv.get('_rebase'):subtpl,rargs=env.pop('_rebase')rargs['base']=''.join(_stdout)# copy stdoutdel_stdout[:]# clear stdoutreturnself._include(env,subtpl,**rargs)returnenv[docs]defrender(self,*args,**kwargs):""" Render the template using keyword arguments as local variables. """env={}stdout=[]fordictarginargs:env.update(dictarg)env.update(kwargs)self.execute(stdout,env)return''.join(stdout) classStplSyntaxError(TemplateError):passclassStplParser:""" Parser for stpl templates. """_re_cache={}#: Cache for compiled re patterns# This huge pile of voodoo magic splits python code into 8 different tokens.# We use the verbose (?x) regex mode to make this more manageable_re_tok=r'''( [urbURB]* (?: ''(?!') |""(?!") |'{6} |"{6} |'(?:[^\\']|\\.)+?' |"(?:[^\\"]|\\.)+?" |'{3}(?:[^\\]|\\.|\n)+?'{3} |"{3}(?:[^\\]|\\.|\n)+?"{3} ) )'''_re_inl=_re_tok.replace(r'|\n','')# We re-use this string pattern later_re_tok+=r''' # 2: Comments (until end of line, but not the newline itself) |(\#.*) # 3: Open and close (4) grouping tokens |([\[\{\(]) |([\]\}\)]) # 5,6: Keywords that start or continue a python block (only start of line) |^([\ \t]*(?:if|for|while|with|try|def|class)\b) |^([\ \t]*(?:elif|else|except|finally)\b) # 7: Our special 'end' keyword (but only if it stands alone) |((?:^|;)[\ \t]*end[\ \t]*(?=(?:%(block_close)s[\ \t]*)?\r?$|;|\#)) # 8: A customizable end-of-code-block template token (only end of line) |(%(block_close)s[\ \t]*(?=\r?$)) # 9: And finally, a single newline. The 10th token is 'everything else' |(\r?\n) '''# Match the start tokens of code areas in a template_re_split=r'''(?m)^[ \t]*(\\?)((%(line_start)s)|(%(block_start)s))'''# Match inline statements (may contain python strings)_re_inl=r'''%%(inline_start)s((?:%s|[^'"\n])*?)%%(inline_end)s'''%_re_inl# add the flag in front of the regexp to avoid Deprecation warning (see Issue #949)# verbose and dot-matches-newline mode_re_tok='(?mx)'+_re_tok_re_inl='(?mx)'+_re_inldefault_syntax='<% %> % {{ }}'def__init__(self,source,syntax=None,encoding='utf8'):self.source,self.encoding=touni(source,encoding),encodingself.set_syntax(syntaxorself.default_syntax)self.code_buffer,self.text_buffer=[],[]self.lineno,self.offset=1,0self.indent,self.indent_mod=0,0self.paren_depth=0defget_syntax(self):""" Tokens as a space separated string (default: <% %> % {{ }}) """returnself._syntaxdefset_syntax(self,syntax):self._syntax=syntaxself._tokens=syntax.split()ifsyntaxnotinself._re_cache:names='block_start block_close line_start inline_start inline_end'etokens=map(re.escape,self._tokens)pattern_vars=dict(zip(names.split(),etokens))patterns=(self._re_split,self._re_tok,self._re_inl)patterns=[re.compile(p%pattern_vars)forpinpatterns]self._re_cache[syntax]=patternsself.re_split,self.re_tok,self.re_inl=self._re_cache[syntax]syntax=property(get_syntax,set_syntax)deftranslate(self):ifself.offset:raiseRuntimeError('Parser is a one time instance.')whileTrue:m=self.re_split.search(self.source,pos=self.offset)ifm:text=self.source[self.offset:m.start()]self.text_buffer.append(text)self.offset=m.end()ifm.group(1):# Escape syntaxline,sep,_=self.source[self.offset:].partition('\n')self.text_buffer.append(self.source[m.start():m.start(1)]+m.group(2)+line+sep)self.offset+=len(line+sep)continueself.flush_text()self.offset+=self.read_code(self.source[self.offset:],multiline=bool(m.group(4)))else:breakself.text_buffer.append(self.source[self.offset:])self.flush_text()return''.join(self.code_buffer)defread_code(self,pysource,multiline):code_line,comment='',''offset=0whileTrue:m=self.re_tok.search(pysource,pos=offset)ifnotm:code_line+=pysource[offset:]offset=len(pysource)self.write_code(code_line.strip(),comment)breakcode_line+=pysource[offset:m.start()]offset=m.end()_str,_com,_po,_pc,_blk1,_blk2,_end,_cend,_nl=m.groups()ifself.paren_depth>0and(_blk1or_blk2):# a if b else ccode_line+=_blk1or_blk2continueif_str:# Python stringcode_line+=_strelif_com:# Python comment (up to EOL)comment=_comifmultilineand_com.strip().endswith(self._tokens[1]):multiline=False# Allow end-of-block in commentselif_po:# open parenthesisself.paren_depth+=1code_line+=_poelif_pc:# close parenthesisifself.paren_depth>0:# we could check for matching parentheses here, but it's# easier to leave that to python - just check countsself.paren_depth-=1code_line+=_pcelif_blk1:# Start-block keyword (if/for/while/def/try/...)code_line=_blk1self.indent+=1self.indent_mod-=1elif_blk2:# Continue-block keyword (else/elif/except/...)code_line=_blk2self.indent_mod-=1elif_cend:# The end-code-block template token (usually '%>')ifmultiline:multiline=Falseelse:code_line+=_cendelif_end:self.indent-=1self.indent_mod+=1else:# \nself.write_code(code_line.strip(),comment)self.lineno+=1code_line,comment,self.indent_mod='','',0ifnotmultiline:breakreturnoffsetdefflush_text(self):text=''.join(self.text_buffer)delself.text_buffer[:]ifnottext:returnparts,pos,nl=[],0,'\\\n'+' '*self.indentforminself.re_inl.finditer(text):prefix,pos=text[pos:m.start()],m.end()ifprefix:parts.append(nl.join(map(repr,prefix.splitlines(True))))ifprefix.endswith('\n'):parts[-1]+=nlparts.append(self.process_inline(m.group(1).strip()))ifpos<len(text):prefix=text[pos:]lines=prefix.splitlines(True)iflines[-1].endswith('\\\\\n'):lines[-1]=lines[-1][:-3]eliflines[-1].endswith('\\\\\r\n'):lines[-1]=lines[-1][:-4]parts.append(nl.join(map(repr,lines)))code='_printlist((%s,))'%', '.join(parts)self.lineno+=code.count('\n')+1self.write_code(code)@staticmethoddefprocess_inline(chunk):ifchunk[0]=='!':return'_str(%s)'%chunk[1:]return'_escape(%s)'%chunkdefwrite_code(self,line,comment=''):code=' '*(self.indent+self.indent_mod)code+=line.lstrip()+comment+'\n'self.code_buffer.append(code)[docs]deftemplate(*args,**kwargs):""" Get a rendered template as a string iterator. You can use a name, a filename or a template string as first parameter. Template rendering arguments can be passed as dictionaries or directly (as keyword arguments). """tpl=args[0]ifargselseNonefordictarginargs[1:]:kwargs.update(dictarg)adapter=kwargs.pop('template_adapter',SimpleTemplate)lookup=kwargs.pop('template_lookup',TEMPLATE_PATH)tplid=(id(lookup),tpl)iftplidnotinTEMPLATESorDEBUG:settings=kwargs.pop('template_settings',{})ifisinstance(tpl,adapter):TEMPLATES[tplid]=tplifsettings:TEMPLATES[tplid].prepare(**settings)elif"\n"intplor"{"intplor"%"intplor'$'intpl:TEMPLATES[tplid]=adapter(source=tpl,lookup=lookup,**settings)else:TEMPLATES[tplid]=adapter(name=tpl,lookup=lookup,**settings)ifnotTEMPLATES[tplid]:abort(500,'Template (%s) not found'%tpl)returnTEMPLATES[tplid].render(kwargs) mako_template=functools.partial(template,template_adapter=MakoTemplate)cheetah_template=functools.partial(template,template_adapter=CheetahTemplate)jinja2_template=functools.partial(template,template_adapter=Jinja2Template)[docs]defview(tpl_name,**defaults):""" Decorator: renders a template for a handler. The handler can control its behavior like that: - return a dict of template vars to fill out the template - return something other than a dict and the view decorator will not process the template, but return the handler result as is. This includes returning a HTTPResponse(dict) to get, for instance, JSON with autojson or other castfilters. """defdecorator(func):@functools.wraps(func)defwrapper(*args,**kwargs):result=func(*args,**kwargs)ifisinstance(result,(dict,DictMixin)):tplvars=defaults.copy()tplvars.update(result)returntemplate(tpl_name,**tplvars)elifresultisNone:returntemplate(tpl_name,**defaults)returnresultreturnwrapperreturndecorator mako_view=functools.partial(view,template_adapter=MakoTemplate)cheetah_view=functools.partial(view,template_adapter=CheetahTemplate)jinja2_view=functools.partial(view,template_adapter=Jinja2Template)################################################################################ Constants and Globals #######################################################################################################################################TEMPLATE_PATH=['./','./views/']TEMPLATES={}DEBUG=FalseNORUN=False# If set, run() does nothing. Used by load_app()#: A dict to map HTTP status codes (e.g. 404) to phrases (e.g. 'Not Found')HTTP_CODES=httplib.responses.copy()HTTP_CODES[418]="I'm a teapot"# RFC 2324HTTP_CODES[428]="Precondition Required"HTTP_CODES[429]="Too Many Requests"HTTP_CODES[431]="Request Header Fields Too Large"HTTP_CODES[451]="Unavailable For Legal Reasons"# RFC 7725HTTP_CODES[511]="Network Authentication Required"_HTTP_STATUS_LINES=dict((k,'%d%s'%(k,v))for(k,v)inHTTP_CODES.items())#: The default template used for error pages. Override with @error()ERROR_PAGE_TEMPLATE="""%%try:%%from%s import DEBUG, request <!DOCTYPE HTML PUBLIC "-//IETF//DTD HTML 2.0//EN"> <html> <head> <title>Error: {{e.status}}</title> <style type="text/css"> html {background-color: #eee; font-family: sans-serif;} body {background-color: #fff; border: 1px solid #ddd; padding: 15px; margin: 15px;} pre {background-color: #eee; border: 1px solid #ddd; padding: 5px;} </style> </head> <body> <h1>Error: {{e.status}}</h1> <p>Sorry, the requested URL <tt>{{repr(request.url)}}</tt> caused an error:</p> <pre>{{e.body}}</pre>%%if DEBUG and e.exception: <h2>Exception:</h2>%%try:%%exc = repr(e.exception)%%except:%%exc = '<unprintable%%s object>'%% type(e.exception).__name__%%end <pre>{{exc}}</pre>%%end%%if DEBUG and e.traceback: <h2>Traceback:</h2> <pre>{{e.traceback}}</pre>%%end </body> </html>%%except ImportError: <b>ImportError:</b> Could not generate the error page. Please add bottle to the import path.%%end"""%__name__#: A thread-safe instance of :class:`LocalRequest`. If accessed from within a#: request callback, this instance always refers to the *current* request#: (even on a multi-threaded server).request=LocalRequest()#: A thread-safe instance of :class:`LocalResponse`. It is used to change the#: HTTP response for the *current* request.response=LocalResponse()#: A thread-safe namespace. Not used by Bottle.local=threading.local()# Initialize app stack (create first empty Bottle app now deferred until needed)# BC: 0.6.4 and needed for run()apps=app=default_app=AppStack()#: A virtual package that redirects import statements.#: Example: ``import bottle.ext.sqlite`` actually imports `bottle_sqlite`.ext=_ImportRedirect('bottle.ext'if__name__=='__main__'else__name__+".ext",'bottle_%s').moduledef_main(argv):# pragma: no coverageargs,parser=_cli_parse(argv)def_cli_error(cli_msg):parser.print_help()_stderr('\nError:%s\n'%cli_msg)sys.exit(1)ifargs.version:print(__version__)sys.exit(0)ifnotargs.app:_cli_error("No application entry point specified.")sys.path.insert(0,'.')sys.modules.setdefault('bottle',sys.modules['__main__'])host,port=(args.bindor'localhost'),8080if':'inhostandhost.rfind(']')<host.rfind(':'):host,port=host.rsplit(':',1)host=host.strip('[]')config=ConfigDict()forcfileinargs.confor[]:try:ifcfile.endswith('.json'):withopen(cfile,'rb')asfp:config.load_dict(json_loads(fp.read()))else:config.load_config(cfile)exceptconfigparser.Errorasparse_error:_cli_error(parse_error)exceptIOError:_cli_error("Unable to read config file%r"%cfile)except(UnicodeError,TypeError,ValueError)aserror:_cli_error("Unable to parse config file%r:%s"%(cfile,error))forcvalinargs.paramor[]:if'='incval:config.update((cval.split('=',1),))else:config[cval]=Truerun(args.app,host=host,port=int(port),server=args.server,reloader=args.reload,plugins=args.plugin,debug=args.debug,config=config)defmain():_main(sys.argv)if__name__=='__main__':# pragma: no coveragemain()