#源码样例-1

from flask import Flask

app = Flask(__name__)

@app.route('/')

def hello_world():

    return 'Hello World!'

if __name__ == '__main__':

    app.run()

#源码样例-2

# 1. 可调用对象是一个函数

def application(environ, start_response):

  response_body = 'The request method was %s' % environ['REQUEST_METHOD']

  # HTTP response code and message

  status = '200 OK'

  # 应答的头部是一个列表，每对键值都必须是一个 tuple。

  response_headers = [('Content-Type', 'text/plain'),

                      ('Content-Length', str(len(response_body)))]

  # 调用服务器程序提供的 start_response，填入两个参数

  start_response(status, response_headers)

  # 返回必须是 iterable

  return [response_body]

#2. 可调用对象是一个类实例

class AppClass:

    """这里的可调用对象就是 AppClass 的实例，使用方法类似于：

        app = AppClass()

        for result in app(environ, start_response):

            do_somthing(result)

    """

    def __init__(self):

        pass

    def __call__(self, environ, start_response):

        status = '200 OK'

        response_headers = [('Content-type', 'text/plain')]

        self.start(status, response_headers)

        yield "Hello world!\n"

#源码样例-3

def run(self, host=None, port=None, debug=None, **options):

        from werkzeug.serving import run_simple

        if host is None:

            host = '127.0.0.1'

        if port is None:

            server_name = self.config['SERVER_NAME']

            if server_name and ':' in server_name:

                port = int(server_name.rsplit(':', 1)[1])

            else:

                port = 5000

        if debug is not None:

            self.debug = bool(debug)

        options.setdefault('use_reloader', self.debug)

        options.setdefault('use_debugger', self.debug)

        try:

            run_simple(host, port, self, **options)

        finally:

            # reset the first request information if the development server

            # reset normally.  This makes it possible to restart the server

            # without reloader and that stuff from an interactive shell.

            self._got_first_request = False

#源码样例-4

def inner():

        try:

            fd = int(os.environ['WERKZEUG_SERVER_FD'])

        except (LookupError, ValueError):

            fd = None

        srv = make_server(hostname, port, application, threaded,

                          processes, request_handler,

                          passthrough_errors, ssl_context,

                          fd=fd)

        if fd is None:

            log_startup(srv.socket)

        srv.serve_forever()

#源码样例-5

def execute(app):

    application_iter = app(environ, start_response)

    try:

        for data in application_iter:

            write(data)

        if not headers_sent:

            write(b'')

    finally:

        if hasattr(application_iter, 'close'):

            application_iter.close()

        application_iter = None

#源码样例-6

def __call__(self, environ, start_response):

        """Shortcut for :attr:`wsgi_app`."""

        return self.wsgi_app(environ, start_response)

#源码样例-7

def wsgi_app(self, environ, start_response):

    ctx = self.request_context(environ)

    ctx.push()

    error = None

    try:

        try:

            response = self.full_dispatch_request()

        except Exception as e:

            error = e

            response = self.handle_exception(e)

        return response(environ, start_response)

    finally:

        if self.should_ignore_error(error):

            error = None

        ctx.auto_pop(error)

#源码样例-8

from werkzeug.routing import Map, Rule, NotFound, RequestRedirect

url_map = Map([

    Rule('/', endpoint='blog/index'),

    Rule('/<int:year>/', endpoint='blog/archive'),

    Rule('/<int:year>/<int:month>/', endpoint='blog/archive'),

    Rule('/<int:year>/<int:month>/<int:day>/', endpoint='blog/archive'),

    Rule('/<int:year>/<int:month>/<int:day>/<slug>',

        endpoint='blog/show_post'),

    Rule('/about', endpoint='blog/about_me'),

    Rule('/feeds/', endpoint='blog/feeds'),

    Rule('/feeds/<feed_name>.rss', endpoint='blog/show_feed')

])

def application(environ, start_response):

    urls = url_map.bind_to_environ(environ)

    try:

        endpoint, args = urls.match()

    except HTTPException, e:

        return e(environ, start_response)

    start_response('200 OK', [('Content-Type', 'text/plain')])

    return ['Rule points to %r with arguments %r' % (endpoint, args)]

#源码样例-9

def add_url_rule(self, rule, endpoint=None, view_func=None, **options):

    if endpoint is None:

        endpoint = _endpoint_from_view_func(view_func)

    options['endpoint'] = endpoint

    methods = options.pop('methods', None)

    if methods is None:

        methods = getattr(view_func, 'methods', None) or ('GET',)

    if isinstance(methods, string_types):

        raise TypeError('Allowed methods have to be iterables of strings, '

                        'for example: @app.route(…, methods=["POST"])')

    methods = set(item.upper() for item in methods)

    required_methods = set(getattr(view_func, 'required_methods', ()))

    provide_automatic_options = getattr(view_func,

                                        'provide_automatic_options', None)

    if provide_automatic_options is None:

        if 'OPTIONS' not in methods:

            provide_automatic_options = True

            required_methods.add('OPTIONS')

        else:

            provide_automatic_options = False

    methods |= required_methods

    rule = self.url_rule_class(rule, methods=methods, **options)

    rule.provide_automatic_options = provide_automatic_options

    self.url_map.add(rule)

    if view_func is not None:

        old_func = self.view_functions.get(endpoint)

        if old_func is not None and old_func != view_func:

            raise AssertionError('View function mapping is overwriting an '

                                'existing endpoint function: %s' % endpoint)

        self.view_functions[endpoint] = view_func

#源码样例-10

def full_dispatch_request(self):

    self.try_trigger_before_first_request_functions()

    try:

        request_started.send(self)

        rv = self.preprocess_request()

        if rv is None:

            rv = self.dispatch_request()

    except Exception as e:

        rv = self.handle_user_exception(e)

    return self.finalize_request(rv)

#源码样例-11

def dispatch_request(self):

    req = _request_ctx_stack.top.request

    if req.routing_exception is not None:

        self.raise_routing_exception(req)

    rule = req.url_rule

    if getattr(rule, 'provide_automatic_options', False) \

            and req.method == 'OPTIONS':

        return self.make_default_options_response()

    return self.view_functions[rule.endpoint](**req.view_args)

#源码样例-12

class Local(object):

    __slots__ = ('__storage__', '__ident_func__')

    def __init__(self):

        object.__setattr__(self, '__storage__', {})

        object.__setattr__(self, '__ident_func__', get_ident)

   def __getattr__(self, name):

        try:

            return self.__storage__[self.__ident_func__()][name]

        except KeyError:

            raise AttributeError(name)

    def __setattr__(self, name, value):

        ident = self.__ident_func__()

        storage = self.__storage__

        try:

            storage[ident][name] = value

        except KeyError:

            storage[ident] = {name: value}

#源码样例-13

class LocalStack(object):

    def __init__(self):

        self._local = Local()

   def push(self, obj):

        rv = getattr(self._local, 'stack', None)

        if rv is None:

            self._local.stack = rv = []

        rv.append(obj)

        return rv

    def pop(self):

        stack = getattr(self._local, 'stack', None)

        if stack is None:

            return None

        elif len(stack) == 1:

            release_local(self._local)

            return stack[-1]

        else:

            return stack.pop()

    @property

    def top(self):

        """The topmost item on the stack.  If the stack is empty,

        `None` is returned.

        """

        try:

            return self._local.stack[-1]

        except (AttributeError, IndexError):

            return None

>>> ls = LocalStack()

>>> ls.push(42)

>>> ls.top

42

>>> ls.push(23)

>>> ls.top

23

>>> ls.pop()

23

>>> ls.top

42

#源码样例-14

def __init__(self, local, name=None):

    object.__setattr__(self, '_LocalProxy__local', local)

    object.__setattr__(self, '__name__', name)

#源码样例-15

def __call__(self):

    def _lookup():

        rv = self.top

        if rv is None:

            raise RuntimeError('object unbound')

        return rv

    return LocalProxy(_lookup)

#源码样例-16

_response_local = LocalStack()

response = _response_local()

#源码样例-17

def _lookup_req_object(name):

    top = _request_ctx_stack.top

    if top is None:

        raise RuntimeError(_request_ctx_err_msg)

    return getattr(top, name)

def _lookup_app_object(name):

    top = _app_ctx_stack.top

    if top is None:

        raise RuntimeError(_app_ctx_err_msg)

    return getattr(top, name)

def _find_app():

    top = _app_ctx_stack.top

    if top is None:

        raise RuntimeError(_app_ctx_err_msg)

    return top.app

# context locals

_request_ctx_stack = LocalStack()

_app_ctx_stack = LocalStack()

current_app = LocalProxy(_find_app)

request = LocalProxy(partial(_lookup_req_object, 'request'))

session = LocalProxy(partial(_lookup_req_object, 'session'))

g = LocalProxy(partial(_lookup_app_object, 'g'))

#源码样例-18

from flask import current_app

app = create_app()

admin_app = create_admin_app()

def do_something():

    with app.app_context():

        work_on(current_app)

        with admin_app.app_context():

            work_on(current_app)

#源码样例-19

class RequestContext(object):

    def __init__(self, app, environ, request=None):

        self.app = app

        if request is None:

            request = app.request_class(environ)

        self.request = request

        self.url_adapter = app.create_url_adapter(self.request)

        self.flashes = None

        self.session = None

        self.preserved = False

        self._preserved_exc = None

        self._after_request_functions = []

        self.match_request()

    def push(self):

        top = _request_ctx_stack.top

        if top is not None and top.preserved:

            top.pop(top._preserved_exc)

        app_ctx = _app_ctx_stack.top

        if app_ctx is None or app_ctx.app != self.app:

            app_ctx = self.app.app_context()

            app_ctx.push()

            self._implicit_app_ctx_stack.append(app_ctx)

        else:

            self._implicit_app_ctx_stack.append(None)

        if hasattr(sys, 'exc_clear'):

            sys.exc_clear()

        _request_ctx_stack.push(self)

        self.session = self.app.open_session(self.request)

        if self.session is None:

            self.session = self.app.make_null_session()

from flask import request,session,request,g

"wsgi.multiprocess":"False"

"SERVER_SOFTWARE":"Werkzeug/0.11.15"

"SCRIPT_NAME":""

"REQUEST_METHOD":"GET"

"PATH_INFO":"/favicon.ico"

"SERVER_PROTOCOL":"HTTP/1.1"

"QUERY_STRING":""

"werkzeug.server.shutdown":"<function shutdown_server at 0x0000000003F4FAC8>"

"CONTENT_LENGTH":""

"HTTP_USER_AGENT":"Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:55.0) Gecko/20100101 Firefox/55.0"

"HTTP_CONNECTION":"keep-alive"

"SERVER_NAME":"127.0.0.1"

"REMOTE_PORT":"12788"

"wsgi.url_scheme":"http"

"SERVER_PORT":"5000"

"wsgi.input":"<socket._fileobject object at 0x0000000003E18408>"

"HTTP_HOST":"127.0.0.1:5000"

"wsgi.multithread":"False"

"HTTP_ACCEPT":"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"

"wsgi.version":"(1, 0)"

"wsgi.run_once":"False"

"wsgi.errors":"<open file '<stderr>', mode 'w' at 0x0000000001DD2150>"

"REMOTE_ADDR":"127.0.0.1"

"HTTP_ACCEPT_LANGUAGE":"zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3"

"CONTENT_TYPE":""

"HTTP_ACCEPT_ENCODING":"gzip, deflate"

#源码样例-20

class Request(BaseRequest, AcceptMixin, ETagRequestMixin,

              UserAgentMixin, AuthorizationMixin,

              CommonRequestDescriptorsMixin):

    """Full featured request object implementing the following mixins:

    - :class:`AcceptMixin` for accept header parsing

    - :class:`ETagRequestMixin` for etag and cache control handling

    - :class:`UserAgentMixin` for user agent introspection

    - :class:`AuthorizationMixin` for http auth handling

    - :class:`CommonRequestDescriptorsMixin` for common headers

    """

#源码样例-21

def finalize_request(self, rv, from_error_handler=False):

    response = self.make_response(rv)

    try:

        response = self.process_response(response)

        request_finished.send(self, response=response)

    except Exception:

        if not from_error_handler:

            raise

        self.logger.exception('Request finalizing failed with an '

                              'error while handling an error')

    return response

def make_response(self, rv):

    status_or_headers = headers = None

    if isinstance(rv, tuple):

        rv, status_or_headers, headers = rv + (None,) * (3 - len(rv))

    if rv is None:

        raise ValueError('View function did not return a response')

    if isinstance(status_or_headers, (dict, list)):

        headers, status_or_headers = status_or_headers, None

    if not isinstance(rv, self.response_class):

        if isinstance(rv, (text_type, bytes, bytearray)):

            rv = self.response_class(rv, headers=headers,

                                    status=status_or_headers)

            headers = status_or_headers = None

        else:

            rv = self.response_class.force_type(rv, request.environ)

    if status_or_headers is not None:

        if isinstance(status_or_headers, string_types):

            rv.status = status_or_headers

        else:

            rv.status_code = status_or_headers

    if headers:

        rv.headers.extend(headers)

    return rv

def process_response(self, response):

    ctx = _request_ctx_stack.top

    bp = ctx.request.blueprint

    funcs = ctx._after_request_functions

    if bp is not None and bp in self.after_request_funcs:

        funcs = chain(funcs, reversed(self.after_request_funcs[bp]))

    if None in self.after_request_funcs:

        funcs = chain(funcs, reversed(self.after_request_funcs[None]))

    for handler in funcs:

        response = handler(response)

    if not self.session_interface.is_null_session(ctx.session):

        self.save_session(ctx.session, response)

    return response

#源码样例-22

class Response(ResponseBase):

    """The response object that is used by default in Flask.  Works like the

    response object from Werkzeug but is set to have an HTML mimetype by

    default.  Quite often you don't have to create this object yourself because

    :meth:`~flask.Flask.make_response` will take care of that for you.

    If you want to replace the response object used you can subclass this and

    set :attr:`~flask.Flask.response_class` to your subclass.

    """

    default_mimetype = 'text/html'

#源码样例-23

#Type1: 直接配置参数

app.config['SECRET_KEY'] = 'YOUCANNOTGUESSME'

#Type2: 从环境变量中获得配置文件名并导入配置参数

export MyAppConfig=/path/to/settings.cfg #linux

set MyAppConfig=d:\settings.cfg#不能立即生效，不建议windows下通过这种方式获得环境变量。

app.config.from_envvar('MyAppConfig')

#Type3: 从对象中获得配置

class Config(object):

    DEBUG = False

    TESTING = False

    DATABASE_URI = 'sqlite://:memory:'

class ProductionConfig(Config):

    DATABASE_URI = 'mysql://user@localhost/foo'

app.config.from_object(ProductionConfig)

print app.config.get('DATABASE_URI')

#Type4: 从文件中获得配置参数

# default_config.py

HOST = 'localhost'

PORT = 5000

DEBUG = True

# flask中使用

app.config.from_pyfile('default_config.py')

#源码样例-24

app.config["DEBUG"]=True

@app.route('/')

def hello_world():

    a=3/0

    return 'Hello World!'

#源码样例-25

#: The configuration dictionary as :class:`Config`.  This behaves

#: exactly like a regular dictionary but supports additional methods

#: to load a config from files.

self.config = self.make_config(instance_relative_config)

#源码样例-26

def make_config(self, instance_relative=False):

    """Used to create the config attribute by the Flask constructor.

    The `instance_relative` parameter is passed in from the constructor

    of Flask (there named `instance_relative_config`) and indicates if

    the config should be relative to the instance path or the root path

    of the application.

    .. versionadded:: 0.8

    """

    root_path = self.root_path

    if instance_relative:

        root_path = self.instance_path

    return self.config_class(root_path, self.default_config)

config_class = Config

#源码样例-27

class Config(dict):

      def __init__(self, root_path, defaults=None):

        dict.__init__(self, defaults or {})

        self.root_path = root_path

#源码样例-28

#: Default configuration parameters.

    default_config = ImmutableDict({

        'DEBUG':                                get_debug_flag(default=False),

        'TESTING':                              False,

        'PROPAGATE_EXCEPTIONS':                None,

        'PRESERVE_CONTEXT_ON_EXCEPTION':        None,

        'SECRET_KEY':                          None,

        'PERMANENT_SESSION_LIFETIME':          timedelta(days=31),

        'USE_X_SENDFILE':                      False,

        'LOGGER_NAME':                          None,

        'LOGGER_HANDLER_POLICY':              'always',

        'SERVER_NAME':                          None,

        'APPLICATION_ROOT':                    None,

        'SESSION_COOKIE_NAME':                  'session',

        'SESSION_COOKIE_DOMAIN':                None,

        'SESSION_COOKIE_PATH':                  None,

        'SESSION_COOKIE_HTTPONLY':              True,

        'SESSION_COOKIE_SECURE':                False,

        'SESSION_REFRESH_EACH_REQUEST':        True,

        'MAX_CONTENT_LENGTH':                  None,

        'SEND_FILE_MAX_AGE_DEFAULT':            timedelta(hours=12),

        'TRAP_BAD_REQUEST_ERRORS':              False,

        'TRAP_HTTP_EXCEPTIONS':                False,

        'EXPLAIN_TEMPLATE_LOADING':            False,

        'PREFERRED_URL_SCHEME':                'http',

        'JSON_AS_ASCII':                        True,

        'JSON_SORT_KEYS':                      True,

        'JSONIFY_PRETTYPRINT_REGULAR':          True,

        'JSONIFY_MIMETYPE':                    'application/json',

        'TEMPLATES_AUTO_RELOAD':                None,

    })

exec (compile(config_file.read(), filename, 'exec'), d.__dict__)

#源码样例-29

import types

var2=types.ModuleType("test")

exec("A='bb'",var2.__dict__)

#源码样例-30

def from_envvar(self, variable_name, silent=False):

    rv = os.environ.get(variable_name)

    if not rv:

        if silent:

            return False

        raise RuntimeError('The environment variable %r is not set '

                          'and as such configuration could not be '

                          'loaded.  Set this variable and make it '

                          'point to a configuration file' %

                          variable_name)

    return self.from_pyfile(rv, silent=silent)

def from_pyfile(self, filename, silent=False):

    filename = os.path.join(self.root_path, filename)

    d = types.ModuleType('config')

    d.__file__ = filename

    try:

        with open(filename) as config_file:

            exec (compile(config_file.read(), filename, 'exec'), d.__dict__)

    except IOError as e:

        if silent and e.errno in (errno.ENOENT, errno.EISDIR):

            return False

        e.strerror = 'Unable to load configuration file (%s)' % e.strerror

        raise

    self.from_object(d)

    return True

def from_object(self, obj):

    """Updates the values from the given object.  An object can be of one

    of the following two types:

    -  a string: in this case the object with that name will be imported

    -  an actual object reference: that object is used directly

    Objects are usually either modules or classes. :meth:`from_object`

    loads only the uppercase attributes of the module/class. A ``dict``

    object will not work with :meth:`from_object` because the keys of a

    ``dict`` are not attributes of the ``dict`` class.

    Example of module-based configuration::

        app.config.from_object('yourapplication.default_config')

        from yourapplication import default_config

        app.config.from_object(default_config)

    You should not use this function to load the actual configuration but

    rather configuration defaults.  The actual config should be loaded

    with :meth:`from_pyfile` and ideally from a location not within the

    package because the package might be installed system wide.

    See :ref:`config-dev-prod` for an example of class-based configuration

    using :meth:`from_object`.

    :param obj: an import name or object

    """

    if isinstance(obj, string_types):

        obj = import_string(obj)

    for key in dir(obj):

        if key.isupper():

            self[key] = getattr(obj, key)