dinislage/CouchPotatoServer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Flask update

Browse Source
This commit is contained in:
Ruud
2011-08-24 20:59:59 +02:00
parent 970b400a0c
commit d7f758dff4
90 changed files with 3328 additions and 1236 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/flask/helpers.py
+187 -44
View File
@@ -9,12 +9,15 @@
:license: BSD, see LICENSE for more details.
"""
from __future__ import with_statement
import os
import sys
import posixpath
import mimetypes
from time import time
from zlib import adler32
from threading import RLock
# try to load the best simplejson implementation available. If JSON
# is not installed, we add a failing class.
@@ -33,9 +36,15 @@ except ImportError:
json_available = False
from werkzeug import Headers, wrap_file, cached_property
from werkzeug.datastructures import Headers
from werkzeug.exceptions import NotFound
# this was moved in 0.7
try:
from werkzeug.wsgi import wrap_file
except ImportError:
from werkzeug.utils import wrap_file
from jinja2 import FileSystemLoader
from .globals import session, _request_ctx_stack, current_app, request
@@ -58,6 +67,10 @@ else:
_tojson_filter = json.dumps
# sentinel
_missing = object()
# what separators does this operating system provide that are not a slash?
# this is used by the send_from_directory function to ensure that nobody is
# able to access files from outside the filesystem.
@@ -151,22 +164,16 @@ def make_response(*args):
def url_for(endpoint, **values):
"""Generates a URL to the given endpoint with the method provided.
The endpoint is relative to the active module if modules are in use.
Here are some examples:
==================== ======================= =============================
Active Module Target Endpoint Target Function
==================== ======================= =============================
`None` ``'index'`` `index` of the application
`None` ``'.index'`` `index` of the application
``'admin'`` ``'index'`` `index` of the `admin` module
any ``'.index'`` `index` of the application
any ``'admin.index'`` `index` of the `admin` module
==================== ======================= =============================
Variable arguments that are unknown to the target endpoint are appended
to the generated URL as query arguments.
to the generated URL as query arguments. If the value of a query argument
is `None`, the whole pair is skipped. In case blueprints are active
you can shortcut references to the same blueprint by prefixing the
local endpoint with a dot (``.``).
This will reference the index function local to the current blueprint::
url_for('.index')
For more information, head over to the :ref:`Quickstart <url-building>`.
@@ -175,13 +182,22 @@ def url_for(endpoint, **values):
:param _external: if set to `True`, an absolute URL is generated.
"""
ctx = _request_ctx_stack.top
if '.' not in endpoint:
mod = ctx.request.module
if mod is not None:
endpoint = mod + '.' + endpoint
elif endpoint.startswith('.'):
endpoint = endpoint[1:]
blueprint_name = request.blueprint
if not ctx.request._is_old_module:
if endpoint[:1] == '.':
if blueprint_name is not None:
endpoint = blueprint_name + endpoint
else:
endpoint = endpoint[1:]
else:
# TODO: get rid of this deprecated functionality in 1.0
if '.' not in endpoint:
if blueprint_name is not None:
endpoint = blueprint_name + '.' + endpoint
elif endpoint.startswith('.'):
endpoint = endpoint[1:]
external = values.pop('_external', False)
ctx.app.inject_url_defaults(endpoint, values)
return ctx.url_adapter.build(endpoint, values, force_external=external)
@@ -248,7 +264,8 @@ def get_flashed_messages(with_categories=False):
"""
flashes = _request_ctx_stack.top.flashes
if flashes is None:
_request_ctx_stack.top.flashes = flashes = session.pop('_flashes', [])
_request_ctx_stack.top.flashes = flashes = session.pop('_flashes') \
if '_flashes' in session else []
if not with_categories:
return [x[1] for x in flashes]
return flashes
@@ -321,7 +338,7 @@ def send_file(filename_or_fp, mimetype=None, as_attachment=False,
if not attachment_filename and not mimetype \
and isinstance(filename, basestring):
warn(DeprecationWarning('The filename support for file objects '
'passed to send_file is not deprecated. Pass an '
'passed to send_file is now deprecated. Pass an '
'attach_filename if you want mimetypes to be guessed.'),
stacklevel=2)
if add_etags:
@@ -377,7 +394,10 @@ def send_file(filename_or_fp, mimetype=None, as_attachment=False,
rv.set_etag('flask-%s-%s-%s' % (
os.path.getmtime(filename),
os.path.getsize(filename),
adler32(filename) & 0xffffffff
adler32(
filename.encode('utf8') if isinstance(filename, unicode)
else filename
) & 0xffffffff
))
if conditional:
rv = rv.make_conditional(request)
@@ -388,6 +408,31 @@ def send_file(filename_or_fp, mimetype=None, as_attachment=False,
return rv
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def send_from_directory(directory, filename, **options):
"""Send a file from a given directory with :func:`send_file`. This
is a secure way to quickly expose static files from an upload folder
@@ -415,35 +460,128 @@ def send_from_directory(directory, filename, **options):
:param options: optional keyword arguments that are directly
forwarded to :func:`send_file`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or filename.startswith('../'):
raise NotFound()
filename = os.path.join(directory, filename)
filename = safe_join(directory, filename)
if not os.path.isfile(filename):
raise NotFound()
return send_file(filename, conditional=True, **options)
def _get_package_path(name):
"""Returns the path to a package or cwd if that cannot be found."""
def get_root_path(import_name):
"""Returns the path to a package or cwd if that cannot be found. This
returns the path of a package or the folder that contains a module.
Not to be confused with the package path returned by :func:`find_package`.
"""
__import__(import_name)
try:
return os.path.abspath(os.path.dirname(sys.modules[name].__file__))
except (KeyError, AttributeError):
directory = os.path.dirname(sys.modules[import_name].__file__)
return os.path.abspath(directory)
except AttributeError:
return os.getcwd()
def find_package(import_name):
"""Finds a package and returns the prefix (or None if the package is
not installed) as well as the folder that contains the package or
module as a tuple. The package path returned is the module that would
have to be added to the pythonpath in order to make it possible to
import the module. The prefix is the path below which a UNIX like
folder structure exists (lib, share etc.).
"""
__import__(import_name)
root_mod = sys.modules[import_name.split('.')[0]]
package_path = getattr(root_mod, '__file__', None)
if package_path is None:
package_path = os.getcwd()
else:
package_path = os.path.abspath(os.path.dirname(package_path))
if hasattr(root_mod, '__path__'):
package_path = os.path.dirname(package_path)
# leave the egg wrapper folder or the actual .egg on the filesystem
test_package_path = package_path
if os.path.basename(test_package_path).endswith('.egg'):
test_package_path = os.path.dirname(test_package_path)
site_parent, site_folder = os.path.split(test_package_path)
py_prefix = os.path.abspath(sys.prefix)
if test_package_path.startswith(py_prefix):
return py_prefix, package_path
elif site_folder.lower() == 'site-packages':
parent, folder = os.path.split(site_parent)
# Windows like installations
if folder.lower() == 'lib':
base_dir = parent
# UNIX like installations
elif os.path.basename(parent).lower() == 'lib':
base_dir = os.path.dirname(parent)
else:
base_dir = site_parent
return base_dir, package_path
return None, package_path
class locked_cached_property(object):
"""A decorator that converts a function into a lazy property. The
function wrapped is called the first time to retrieve the result
and then that calculated result is used the next time you access
the value. Works like the one in Werkzeug but has a lock for
thread safety.
"""
def __init__(self, func, name=None, doc=None):
self.__name__ = name or func.__name__
self.__module__ = func.__module__
self.__doc__ = doc or func.__doc__
self.func = func
self.lock = RLock()
def __get__(self, obj, type=None):
if obj is None:
return self
with self.lock:
value = obj.__dict__.get(self.__name__, _missing)
if value is _missing:
value = self.func(obj)
obj.__dict__[self.__name__] = value
return value
class _PackageBoundObject(object):
def __init__(self, import_name):
def __init__(self, import_name, template_folder=None):
#: The name of the package or module. Do not change this once
#: it was set by the constructor.
self.import_name = import_name
#: location of the templates. `None` if templates should not be
#: exposed.
self.template_folder = template_folder
#: Where is the app root located?
self.root_path = _get_package_path(self.import_name)
self.root_path = get_root_path(self.import_name)
self._static_folder = None
self._static_url_path = None
def _get_static_folder(self):
if self._static_folder is not None:
return os.path.join(self.root_path, self._static_folder)
def _set_static_folder(self, value):
self._static_folder = value
static_folder = property(_get_static_folder, _set_static_folder)
del _get_static_folder, _set_static_folder
def _get_static_url_path(self):
if self._static_url_path is None:
if self.static_folder is None:
return None
return '/' + os.path.basename(self.static_folder)
return self._static_url_path
def _set_static_url_path(self, value):
self._static_url_path = value
static_url_path = property(_get_static_url_path, _set_static_url_path)
del _get_static_url_path, _set_static_url_path
@property
def has_static_folder(self):
@@ -452,15 +590,17 @@ class _PackageBoundObject(object):
.. versionadded:: 0.5
"""
return os.path.isdir(os.path.join(self.root_path, 'static'))
return self.static_folder is not None
@cached_property
@locked_cached_property
def jinja_loader(self):
"""The Jinja loader for this package bound object.
.. versionadded:: 0.5
"""
return FileSystemLoader(os.path.join(self.root_path, 'templates'))
if self.template_folder is not None:
return FileSystemLoader(os.path.join(self.root_path,
self.template_folder))
def send_static_file(self, filename):
"""Function used internally to send static files from the static
@@ -468,10 +608,11 @@ class _PackageBoundObject(object):
.. versionadded:: 0.5
"""
return send_from_directory(os.path.join(self.root_path, 'static'),
filename)
if not self.has_static_folder:
raise RuntimeError('No static folder for this object')
return send_from_directory(self.static_folder, filename)
def open_resource(self, resource):
def open_resource(self, resource, mode='rb'):
"""Opens a resource from the application's resource folder. To see
how this works, consider the following folder structure::
@@ -493,4 +634,6 @@ class _PackageBoundObject(object):
:param resource: the name of the resource. To access resources within
subfolders use forward slashes as separator.
"""
return open(os.path.join(self.root_path, resource), 'rb')
if mode not in ('r', 'rb'):
raise ValueError('Resources can only be opened for reading')
return open(os.path.join(self.root_path, resource), mode)