Werkzeug Turorial「建议收藏」

大家好，又见面了，我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

Werkzeug不是一个framework,而是一个library,用来开发application和framework

一个WSGI的application像这样:

def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/plain')])
    return ['Hello World!']

一个WSGI application是一个可以调用的,有environ(dict)和start_response(callable)
environ包含了所有的请求的信息
start_response方法表明开始处理response

request允许包含environ
response是一个WSGI application

from werkzeug.wrappers import Response

def application(environ, start_response):
    response = Response('Hello World!', mimetype='text/plain')
    return response(environ, start_response)

处理请求,并返回

from werkzeug.wrappers import Request, Response

def application(environ, start_response):
    request = Request(environ)
    text = 'Hello %s!' % request.args.get('name', 'World')
    response = Response(text, mimetype='text/plain')
    return response(environ, start_response)

1.创建文件夹

/shortly
    /static
    /templates

2.基础的架构
在/shortly文件夹下创建shortly.py文件

import os
import redis
import urlparse
from werkzeug.wrappers import Request, Response
from werkzeug.routing import Map, Rule
from werkzeug.exceptions import HTTPException, NotFound
from werkzeug.middleware.shared_data import SharedDataMiddleware
from werkzeug.utils import redirect
from jinja2 import Environment, FileSystemLoader

class Shortly(object): # 一个WSGI application

    def __init__(self, config):
        self.redis = redis.Redis(config['redis_host'], config['redis_port'])

    def dispatch_request(self, request):
        return Response('Hello World!')

    def wsgi_app(self, environ, start_response):
        request = Request(environ) # 获得request
        response = self.dispatch_request(request) # 处理request,并返回response,嗲用的是`dispatch_request`方法
        return response(environ, start_response) # 返回

    def __call__(self, environ, start_response): # 调用wsgi_app
        return self.wsgi_app(environ, start_response)


def create_app(redis_host='localhost', redis_port=6379, with_static=True): # 用来创建application,添加配置文件和middleware
    app = Shortly({ 
    # 连接数据库的配置
        'redis_host':       redis_host,
        'redis_port':       redis_port
    })
    if with_static: # 添加中间件,用来处理static文件
        app.wsgi_app = SharedDataMiddleware(app.wsgi_app, { 
   
            '/static':  os.path.join(os.path.dirname(__file__), 'static')
        })
    return app

if __name__ == '__main__':
    from werkzeug.serving import run_simple
    app = create_app()
    run_simple('127.0.0.1', 5000, app, use_debugger=True, use_reloader=True)

运行

$ python shortly.py
 * Running on http://127.0.0.1:5000/
 * Restarting with reloader: stat() polling

3.环境
添加渲染模板和连接到redis

def __init__(self, config):
    self.redis = redis.Redis(config['redis_host'], config['redis_port'])
    template_path = os.path.join(os.path.dirname(__file__), 'templates')
    self.jinja_env = Environment(loader=FileSystemLoader(template_path),
                                 autoescape=True)

def render_template(self, template_name, **context):
    t = self.jinja_env.get_template(template_name)
    return Response(t.render(context), mimetype='text/html')

4.路由
Routing is the process of matching and parsing the URL.
先创建Map实例,然后添加Rule实例.每个rule可以匹配URL和endpoint
The endpoint is typically a string and can be used to uniquely identify the URL.

self.url_map = Map([
    Rule('/', endpoint='new_url'),
    Rule('/<short_id>', endpoint='follow_short_link'),
    Rule('/<short_id>+', endpoint='short_link_details')
])

Werkzeug提供了URL和Endpoint的绑定,那么如何绑定Endpoint和function呢?这完全由我们自己来决定.

def dispatch_request(self, request):
    adapter = self.url_map.bind_to_environ(request.environ)
    try:
        endpoint, values = adapter.match()
        return getattr(self, 'on_' + endpoint)(request, **values)
    except HTTPException, e:
        return e

我们将URL map和当前的环境进行绑定,形成了URLAdapter.adapter可以用来匹配request并且可以处理URLs.匹配到的方法将返回endpoint,和URL中的值.
当我们访问http://localhost:5000/foo,我们将获得

endpoint = 'follow_short_link'
values = { 
   'short_id': u'foo'}

如果没有匹配到任何东西,则会报NotFound exception, which is an HTTPException

5.视图

def on_new_url(self, request):
    error = None
    url = ''
    if request.method == 'POST':
        url = request.form['url']
        if not is_valid_url(url):
            error = 'Please enter a valid URL'
        else:
            short_id = self.insert_url(url)
            return redirect('/%s+' % short_id)
    return self.render_template('new_url.html', error=error, url=url)

首先检查请求方法是post,因此我们要验证url,所以我们要先做一个验证url的方法

def is_valid_url(url):
    parts = urlparse.urlparse(url)
    return parts.scheme in ('http', 'https')

添加url的方法

def insert_url(self, url):
    short_id = self.redis.get('reverse-url:' + url)
    if short_id is not None:
        return short_id
    url_num = self.redis.incr('last-url-id')
    short_id = base36_encode(url_num)
    self.redis.set('url-target:' + short_id, url)
    self.redis.set('reverse-url:' + url, short_id)
    return short_id

如果url存在,我们就直接返回,如果没有我们要传入redis,先转成base36

def base36_encode(number):
    assert number >= 0, 'positive integer required'
    if number == 0:
        return '0'
    base36 = []
    while number != 0:
        number, i = divmod(number, 36)
        base36.append('0123456789abcdefghijklmnopqrstuvwxyz'[i])
    return ''.join(reversed(base36))

6.redirect
我们查看每个链接被点击的次数

def on_follow_short_link(self, request, short_id):
    link_target = self.redis.get('url-target:' + short_id)
    if link_target is None:
        raise NotFound()
    self.redis.incr('click-count:' + short_id)
    return redirect(link_target)

7.Detail View

def on_short_link_details(self, request, short_id):
    link_target = self.redis.get('url-target:' + short_id)
    if link_target is None:
        raise NotFound()
    click_count = int(self.redis.get('click-count:' + short_id) or 0)
    return self.render_template('short_link_details.html',
        link_target=link_target,
        short_id=short_id,
        click_count=click_count
    )