Werkzeug Turorial「建议收藏」

全栈程序员站长

发布于 2022-11-10 14:36:41

1700

发布于 2022-11-10 14:36:41

文章被收录于专栏：全栈程序员必看

大家好，又见面了，我是你们的朋友全栈君。

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
)

工程文件: shortly in the example folder

参考： https://werkzeug.palletsprojects.com/en/1.0.x/tutorial/

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2022年9月28日，如有侵权请联系 cloudcommunity@tencent.com 删除

php