zoukankan      html  css  js  c++  java
  • 【flask】flask讲解

    flask接口调试利器

    下面是Flask主页给我们的第一个例子,我们现在就由它入手,深入理解“@app.route()”是如何工作的。

    Python

    import flask
    from flask import Flask
    app = Flask(__name__)
    
    @app.route("/")
    
    def hello():
    
        return "Hello World!"


    要想明白“@app.route()”的工作原理,我们首先需要看一看Python中的装饰器(就是以“@”开头的那玩意,下面接着函数定义)。@app.route和其它装饰器

    究竟什么是装饰器?没啥特别的。装饰器只是一种接受函数(就是那个你用“@”符号装饰的函数)的函数,并返回一个新的函数。

    当你装饰一个函数,意味着你告诉Python调用的是那个由你的装饰器返回的新函数,而不仅仅是直接返回原函数体的执行结果。

    还不是很明白?这里是一个简单的例子:

    Python

    # This is our decorator
    
    def simple_decorator(f):
        # This is the new function we're going to return
        # This function will be used in place of our original definition
        def wrapper():
            print "Entering Function"
            f()
            print "Exited Function"
        return wrapper
    
    @simple_decorator
    def hello():
        print "Hello World"
    hello()


    现在我们有点明白怎样创建我们自己的“@app.route()”装饰器了,但你可能会注意到有一个不同点,就是我们的simple_decorator不可以接受任何参数, 但“@app.route()”却可以。运行上述代码会输出以下结果:
    Entering Function
    Hello World
    Exited Function

    很好!

    那么我们怎样才能给我们的装饰器传参数?要实现这个我们只需创建一个“decorator_factory”函数,我们调用这个函数,返回适用于我们函数的装饰器。现在看看如果实现它。

    Python

    def decorator_factory(enter_message, exit_message):
        # We're going to return this decorator
        def simple_decorator(f):
            def wrapper():
                print enter_message
                f()
                print exit_message
            return wrapper
        return simple_decorator
    
    @decorator_factory("Start", "End")
    def hello():
        print "Hello World"
    
    hello()


    把“app”放进“app.route”给我们的输出是:
    Start
    Hello World
    End

    请注意在我们写@decorator_factory(“Start”, “End”)时,我们实际调用的是decorator_factory函数,实际返回的装饰器已经被用上了,代码很整洁,对吧?

    现在我们掌握了装饰器怎样工作的全部前置知识 ,可以重新实现Flask API的这个部分了,那么把我们的目光转移到“app”在我们Flask应用中的重要地位上面来。

    在开始解释Flask对象里面发生了什么之前,我们先创建我们自己的Python类NotFlask。

    Python

    class NotFlask():
        pass
    app = NotFlask()


    Python这不是个很有趣的类,不过有一样值得注意,就是这个类的方法也可以被用作装饰器,所以让我们把这个类写得更有趣一点,加一个称作 route的方法,它是一个简单的装饰器工厂。

    class NotFlask():
        def route(self, route_str):
            def decorator(f):
                return f
            return decorator
    
    app = NotFlask()
    @app.route("/")
    def hello():
        return "Hello World!"

    这个装饰器和我们之前创建的那些最大的不同,在于我们不想修改被我们装饰的函数的行为,我们只是想获得它的引用。

    所以,最后一步是我们打算去利用一个特性,就是用装饰器函数的副产品去保存一个提供给我们的路径之间的链接,装饰器函数应该与它关联起来。

    为了实现这个,我们给我们的NotFlask对象加一个“routes”字典,当我们的“decorator”函数被调用,路径将被插入新字典中函数对应的位置。

    Python

    class NotFlask():
    
        def __init__(self):
            self.routes = {}
        def route(self, route_str):
            def decorator(f):
                self.routes[route_str] = f
                return f
            return decorator
    
    
    
    app = NotFlask()
    @app.route("/")
    def hello():
        return "Hello World!"


    Python现在我们就要完成了!可如果没法访问内部的视图函数,保存路径的字典又有什么用?让我们加入一个方法serve(path),当给定的路径存在时运行一个函数并给们我结果,当路径尚未注册时则抛出一个异常。


    Python在这个系列我们只关注重现那些热门库提供的友好API,所以钩挂“serve”方法实现一个HTTP服务器其实有一点超出本文的范围,当然结果是确定的,运行下述片段:

    class NotFlask():
    
        def __init__(self):
            self.routes = {}
    
        def route(self, route_str):
            def decorator(f):
                self.routes[route_str] = f
                return f
            return decorator
    
        def serve(self, path):
            view_function = self.routes.get(path)
            if view_function:
                return view_function()
            else:
                raise ValueError('Route "{}"" has not been registered'.format(path))
    
    
    
    app = NotFlask()
    @app.route("/")
    def hello():
        return "Hello World!"
    app = NotFlask()
    
    @app.route("/")
    def hello():
        return "Hello World!"
    
    print app.serve("/")


    我们会看到:

    Hello World!

    我们已经完成了一个的Flask网页上第一个例子的非常简单的重现,让我们写一些快速测试检测我们简单重现的Flask的“@app.route()”是否正确。

    Python

    class TestNotFlask(unittest.TestCase):
    
        def setUp(self):
            self.app = NotFlask()
    
        def test_valid_route(self):
            @self.app.route('/')
            def index():
                return 'Hello World'
            self.assertEqual(self.app.serve('/'), 'Hello World')
    
    
        def test_invalid_route(self):
            with self.assertRaises(ValueError):
                self.app.serve('/invalid')


    完全正确!所以,仅仅是一个简单的包含一个字典的装饰器, 就重现了Flask的“app.route()”装饰器的基本的行为。吸口气。

    在本系列的下一篇,也是Flask的app.route()的最后一篇,将通过解析下面这个例子来解释动态URL模式是如何工作。

    Python

    app = Flask(__name__)
    
    
    @app.route("/hello/<username>")
    def hello_user(username):
        return "Hello {} !".format(username)

    装饰器 获取函数运行时间

    def metric(fn):
        @functools.wraps(fn)
        def f(arg,**kw):
            start=round(time.time()  1000)
            fun=fn(arg,**kw)
            end=round(time.time()  1000)
            print('%s executed in %s ms' % (fn.name, end-start))
            return fun
        return f
    
    @metric
    def test1():
        time.sleep(0.1)
        print('hello')
    
    @metric
    def test2():
        time.sleep(0.01)
        print('hello')
    test1()
    test2()
    from flask import Flask, request
    app = Flask(__name__)
    @app.route('/')
    def hello():
        return 'Hello world'
    if __name__ == '__main__':
        app.run(host='0.0.0.0', debug=True, port=5555)

    本地运行上面代码,打开浏览器访问 http://localhost:5555/ 就可以看到页面输出了 Hello World!

    flask 参考链接参考链接1flask中get活post,用postmanpostman简单教程

    注意 pi() 的返回值不能是浮点数,所以必须使用 str 转换成字符串,运行 python flask_pi.py ,打开浏览器访问 http://localhost:5000/pi?n=1000000 ,可以看到页面输出 3.14159169866 ,这个值同圆周率已经非常接近。

    再仔细观察代码,你还会注意到一个特殊的变量 request ,它看起来似乎是一个全局变量。从全局变量里拿当前请求参数,这非常奇怪。如果在多线程环境中,该如何保证每个线程拿到的都是当前线程正在处理的请求参数呢?所以它不能是全局变量,它是线程局部变量,线程局部变量外表上和全局变量没有差别,但是在访问线程局部变量时,每个线程得到的都是当前线程内部共享的对象。

    from flask import Flask, request,jsonify,url_for
    app = Flask(__name__)
    import math
    import threading
    class PiCache(object):
        def __init__(self):
            self.pis = {}
            self.lock = threading.RLock()
        def set(self,n,result):
            with self.lock:
                self.pis[n] = result
        def get(self,n):
            with self.lock:
                return self.pis.get(n)
    
    cache = PiCache()
    
    @app.route('/')
    def pi():
        n = int(request.args.get('n','100'))
        with app.test_request_context():
            print(url_for('pi')) # 返回的是pi函数的URL =route中第一个参数 /
        result = cache.get(n)
        if result:
            return jsonify({'cached':True,'result':result})
        s = 0.0
        for i in range(1,n):
            s += 1.0/i/i
        result = math.sqrt(6*s)
        cache.set(n,result)
        return jsonify({'cached':False,'result':result})
    
    if __name__ == '__main__':
        app.run()

    第一次:

    {"cached":false,"result":3.1319807472443624}

    第二次:

    {"cached":true,"result":3.1319807472443624}

    from flask import Flask, request,jsonify
    import redis
    app = Flask(__name__)
    import math
    import threading
    class PiCache(object):
        def __init__(self,client):
            self.client = client
        def set(self,n,result):
            self.client.hset('pis',str(n),str(result))
        def get(self,n):
            result = self.client.hget('pis',str(n))
            if not result:
                return
            return float(result)
    
    client = redis.StrictRedis(host='127.0.0.1', port=6379,db = '0')
    cache = PiCache(client)
    
    @app.route('/')
    def pi():
        n = int(request.args.get('n','100'))
        print(n)
        result = cache.get(n)
        if result:
            return jsonify({'cached':True,'result':result})
        s = 0.0
        for i in range(1,n):
            s += 1.0/i/i
        result = math.sqrt(6*s)
        cache.set(n,result)
        return jsonify({'cached':False,'result':result})
    
    if __name__ == '__main__':
        app.run()

    运行 python flask_pi.py ,打开浏览器访问 http://localhost:5000/pi?n=1000000 ,可以看到页面输出

    第一次结果:

    {"cached":false,"result":3.1319807472443624}

     第二次结果:

    {"cached":true,"result":3.1319807472443624}

    重启进程,再次刷新页面,可以看书页面输出的cached字段依然是true,说明缓存结果不再因为进程重启而丢失。

    MethodView

    写过Django的朋友们可能会问,Flask是否支持类形式的API编写方式,回答是肯定的。下面我们使用Flask原生支持的MethodView来改写一下上面的服务。

    from flask import Flask, request,jsonify
    from flask.views import MethodView
    app = Flask(__name__)
    import math
    import threading
    class PiCache(object):
        def __init__(self):
            self.pis = {}
            self.lock = threading.RLock()
        def set(self,n,result):
            with self.lock:
                self.pis[n] = result
        def get(self,n):
            with self.lock:
                return self.pis.get(n)
    
    cache = PiCache()
    
    class PiAPI(MethodView):
        def __init__(self,cache):
            self.cache = cache
        def get(self,n):
            result = self.cache.get(n)
            if result:
                return jsonify({'cached':True,'result':result})
            s = 0.0
            for i in range(1, n):
                s += 1.0 / i / i
            result = math.sqrt(6 * s)
            self.cache.set(n, result)
            return jsonify({'cached': False, 'result': result})
    
    '''
    as_view 提供了参数可以直接注入到MethodView的构造器中
    我们不再使用request.args.keys(),而是直接将参数放到URL里面,这就是RESTFUL风格的URL
    '''
    app.add_url_rule('/pi/<int:n>',view_func=PiAPI.as_view('pi',cache))
    
    if __name__ == '__main__':
        app.run()

    然后调用的时候直接在浏览器输入 : http://127.0.0.1:5000/pi/20
    结果:

    {"cached":false,"result":3.092245052300697}
    关注公众号 海量干货等你
  • 相关阅读:
    racktable安装过程
    racktables
    EM上的按钮是方框的问题
    install oracle
    记一次ORACLE无法启动登陆事故
    安装rlwrap-0.37.tar.gz
    centos7 安装oracle 11g数据库
    centos 7 安装mariadb
    centos7 lamp
    Linux安全之SSH 密钥创建及密钥登录
  • 原文地址:https://www.cnblogs.com/sowhat1412/p/12734348.html
Copyright © 2011-2022 走看看