Python3教程Web開發實戰梳理-day5(Web框架)
Day5:編寫Web框架
前面完成了資料庫的部署與操作封裝,現在開始進入到伺服器方面的工作。我們前面提到了,在本專案中我們引入了非同步框架aiohttp,並且他自身提供了一個伺服器。aiohttp已經是一個框架了,為什麼我們還要再自己實現呢?
原因在於,從框架的使用者的角度來說,aiohttp還是相對比較底層,想要在使用框架時所需要的程式碼更簡潔,就需要我們在aithttp的基礎上增添一些其他的公用的功能,封裝出更高階的Web框架。Web框架的設計是完全從使用者出發,目的是讓框架使用者編寫儘可能少的程式碼。
編寫URL處理函式的大致流程
第一步,編寫一個用@asyncio.coroutine裝飾的函式:
@asyncio.coroutine
def handle_url_xxx(request):
pass
第二步,傳入的引數需要自己從request中獲取:
url_param = request.match_info['key']
query_params = parse_qs(request.query_string)
最後,需要自己構造Response物件:
text = render('template', data)
return web.Response(text.encode('utf-8'))
這些重複的工作可以由框架完成,我們的目標也是如此。
接下來在coroweb.py中編寫框架程式碼
Http定義了與伺服器互動的不同方法,最基本的方法有4種,分別是GET,POST,PUT,DELETE。
URL全稱是資源描述符,我們可以這樣認為:一個URL地址,它用於描述一個網路上的資源,而HTTP中的GET,POST,PUT,DELETE就對應著對這個資源的查,改,增,刪4個操作。
建議:
1、get方式的安全性較Post方式要差些,包含機密資訊的話,建議用Post資料提交方式;
2、在做資料查詢時,建議用Get方式;而在做資料新增、修改或刪除時,建議用Post方式;
要把一個函式對映為一個URL處理函式,我們先定義@get():
def get(path):
'''
Define decorator @get('/path')
'''
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kw):
return func(*args, **kw)
wrapper.__method__ = 'GET'
wrapper.__route__ = path
return wrapper
return decorator
這裡得到的get是一個裝飾器。這樣,一個函式通過@get()的裝飾就附帶了URL資訊。
@post與@get定義類似。
def post(path):
'''
Define decorator @post('/path')
'''
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kw):
return func(*args, **kw)
wrapper.__method__ = 'POST'
wrapper.__route__ = path
return wrapper
return decorator
定義RequestHandler
URL處理函式不一定是一個coroutine,因此我們用RequestHandler()來封裝一個URL處理函式。
RequestHandler是一個類,建立的時候定義了_call_()方法,因此可以將其例項視為函式。
RequestHandler目的就是從URL函式中分析其需要接收的引數,從request中獲取必要的引數,呼叫URL函式,然後把結果轉換為web.Response物件,這樣,就完全符合aiohttp框架的要求
#運用inspect模組,建立幾個函式用以獲取URL處理函式與request引數之間的關係
def get_required_kw_args(fn): #收集沒有預設值的命名關鍵字引數
args = []
params = inspect.signature(fn).parameters #inspect模組是用來分析模組,函式
for name, param in params.items():
if str(param.kind) == 'KEYWORD_ONLY' and param.default == inspect.Parameter.empty:
args.append(name)
return tuple(args)
def get_named_kw_args(fn): #獲取命名關鍵字引數
args = []
params = inspect.signature(fn).parameters
for name,param in params.items():
if str(param.kind) == 'KEYWORD_ONLY':
args.append(name)
return tuple(args)
def has_named_kw_arg(fn): #判斷有沒有命名關鍵字引數
params = inspect.signature(fn).parameters
for name,param in params.items():
if str(param.kind) == 'KEYWORD_ONLY':
return True
def has_var_kw_arg(fn): #判斷有沒有關鍵字引數
params = inspect.signature(fn).parameters
for name,param in params.items():
if str(param.kind) == 'VAR_KEYWORD':
return True
def has_request_arg(fn): #判斷是否含有名叫'request'引數,且該引數是否為最後一個引數
params = inspect.signature(fn).parameters
sig = inspect.signature(fn)
found = False
for name,param in params.items():
if name == 'request':
found = True
continue #跳出當前迴圈,進入下一個迴圈
if found and (str(param.kind) != 'VAR_POSITIONAL' and str(param.kind) != 'KEYWORD_ONLY' and str(param.kind != 'VAR_KEYWORD')):
raise ValueError('request parameter must be the last named parameter in function: %s%s'%(fn.__name__,str(sig)))
return found
#定義RequestHandler,正式向request引數獲取URL處理函式所需的引數
class RequestHandler(object):
def __init__(self,app,fn):#接受app引數
self._app = app
self._fn = fn
self._required_kw_args = get_required_kw_args(fn)
self._named_kw_args = get_named_kw_args(fn)
self._has_named_kw_arg = has_named_kw_arg(fn)
self._has_var_kw_arg = has_var_kw_arg(fn)
self._has_request_arg = has_request_arg(fn)
async def __call__(self,request): #__call__這裡要構造協程
kw = None
if self._has_named_kw_arg or self._has_var_kw_arg:
if request.method == 'POST': #判斷客戶端發來的方法是否為POST
if not request.content_type: #查詢有沒提交資料的格式(EncType)
return web.HTTPBadRequest(text='Missing Content_Type.')#這裡被廖大坑了,要有text
ct = request.content_type.lower() #小寫
if ct.startswith('application/json'): #startswith
params = await request.json() #Read request body decoded as json.
if not isinstance(params,dict):
return web.HTTPBadRequest(text='JSON body must be object.')
kw = params
elif ct.startswith('application/x-www-form-urlencoded') or ct.startswith('multipart/form-data'):
params = await request.post() # reads POST parameters from request body.If method is not POST, PUT, PATCH, TRACE or DELETE or content_type is not empty or application/x-www-form-urlencoded or multipart/form-data returns empty multidict.
kw = dict(**params)
else:
return web.HTTPBadRequest(text='Unsupported Content_Tpye: %s'%(request.content_type))
if request.method == 'GET':
qs = request.query_string #The query string in the URL
if qs:
kw = dict()
for k,v in parse.parse_qs(qs,True).items(): #Parse a query string given as a string argument.Data are returned as a dictionary. The dictionary keys are the unique query variable names and the values are lists of values for each name.
kw[k] = v[0]
if kw is None:
kw = dict(**request.match_info)
else:
if not self._has_var_kw_arg and self._named_kw_args: #當函式引數沒有關鍵字引數時,移去request除命名關鍵字引數所有的引數資訊
copy = dict()
for name in self._named_kw_args:
if name in kw:
copy[name] = kw[name]
kw = copy
for k,v in request.match_info.items(): #檢查命名關鍵引數
if k in kw:
logging.warning('Duplicate arg name in named arg and kw args: %s' % k)
kw[k] = v
if self._has_request_arg:
kw['request'] = request
if self._required_kw_args: #假如命名關鍵字引數(沒有附加預設值),request沒有提供相應的數值,報錯
for name in self._required_kw_args:
if name not in kw:
return web.HTTPBadRequest(text='Missing argument: %s'%(name))
logging.info('call with args: %s' % str(kw))
try:
r = await self._fn(**kw)
return r
except APIError as e: #APIError另外建立
return dict(error=e.error, data=e.data, message=e.message)
在上述RequestHandler程式碼可以看出最後呼叫URL函式時,URL函式可能會返回一個名叫APIError的錯誤,它的作用是用來返回諸如賬號登入資訊的錯誤,這會在day10編寫使用者註冊API裡面講到。
add_route()與add_static()函式
由於我們現在要建立的的Web框架基於aiohttp框架,所以需要再編寫一個add_route函式,用來註冊一個URL處理函式,主要用來驗證函式是否有包含URL的響應方法與路徑資訊,以及將函式變為協程。
程式碼如下:
def add_route(app, fn):
method = getattr(fn, '__method__', None)
path = getattr(fn, '__route__', None)
if path is None or method is None:
raise ValueError('@get or @post not defined in %s.' % str(fn))
if not asyncio.iscoroutinefunction(fn) and not inspect.isgeneratorfunction(fn):
fn = asyncio.coroutine(fn)
logging.info('add route %s %s => %s(%s)' % (method, path, fn.__name__, ', '.join(inspect.signature(fn).parameters.keys())))
app.router.add_route(method, path, RequestHandler(app, fn))
通常add_route()註冊會呼叫很多次,而為了框架使用者更加方便,可以編寫了一個可以批量註冊的函式,預期效果是:只需向這個函式提供要批量註冊函式的檔案路徑,新編寫的函式就會篩選,註冊檔案內所有符合註冊條件的函式。
# 自動把handler模組的所有符合條件的函式註冊了:
def add_routes(app, module_name):
n = module_name.rfind('.')
if n == (-1):
mod = __import__(module_name, globals(), locals())
else:
name = module_name[n+1:]
mod = getattr(__import__(module_name[:n], globals(), locals(), [name]), name)
for attr in dir(mod):
if attr.startswith('_'):
continue
fn = getattr(mod, attr)
if callable(fn):
method = getattr(fn, '__method__', None)
path = getattr(fn, '__route__', None)
if method and path:
add_route(app, fn)
然後新增靜態資料夾的路徑:
def add_static(app):
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'static')
app.router.add_static('/static/', path)
logging.info('add static %s => %s' % ('/static/', path))
之後,要在主程式app.py的初始化函式init()中新增middleware、jinja2模板和自注冊的支援
新增程式碼如下:
from coroweb import add_routes, add_static
from aiohttp import web
from jinja2 import Environment, FileSystemLoader
app = web.Application(loop=loop, middlewares=[
logger_factory, response_factory
])
init_jinja2(app, filters=dict(datetime=datetime_filter))
add_routes(app, 'handlers')
add_static(app)
其中jinja2模板的初始化也需要我們在app.py中實現:
def init_jinja2(app, **kw):
logging.info('init jinja2...')
options = dict(
autoescape = kw.get('autoescape', True),
block_start_string = kw.get('block_start_string', '{%'),
block_end_string = kw.get('block_end_string', '%}'),
variable_start_string = kw.get('variable_start_string', '{{'),
variable_end_string = kw.get('variable_end_string', '}}'),
auto_reload = kw.get('auto_reload', True)
)
path = kw.get('path', None)
if path is None:
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'templates')
logging.info('set jinja2 template path: %s' % path)
env = Environment(loader=FileSystemLoader(path), **options)
filters = kw.get('filters', None)
if filters is not None:
for name, f in filters.items():
env.filters[name] = f
app['__templating__'] = env
其引數中用到的datetime_filter()函式實質是一個攔截器,具體作用在day8中會提及
先給出程式碼:
def datetime_filter(t):
delta = int(time.time() - t)
if delta < 60:
return u'1分鐘前'
if delta < 3600:
return u'%s分鐘前' % (delta // 60)
if delta < 86400:
return u'%s小時前' % (delta // 3600)
if delta < 604800:
return u'%s天前' % (delta // 86400)
dt = datetime.fromtimestamp(t)
return u'%s年%s月%s日' % (dt.year, dt.month, dt.day)
middleware
上面的RequestHandler對於URL做了一系列的處理,但是aiohttp框架最終需要的是返回web.Response物件,實現這一步,這裡引入aiohttp框架的web.Application()中的middleware引數。
簡介:middleware是一種攔截器,一個URL在被某個函式處理前,可以經過一系列的middleware的處理。一個middleware可以改變URL的輸入、輸出,甚至可以決定不繼續處理而直接返回。middleware的用處就在於把通用的功能從每個URL處理函式中拿出來,集中放到一個地方。
當建立web.appliction的時候,可以設定middleware引數,而middleware的設定是通過建立一些middleware factory(協程函式)。這些middleware factory接受一個app例項,一個handler兩個引數,並返回一個新的handler。
一個記錄URL日誌的logger可以簡單定義如下:
async def logger_factory(app, handler):
async def logger(request):
logging.info('Request: %s %s' % (request.method, request.path))
# await asyncio.sleep(0.3)
return (await handler(request))
return logger
response這個middleware把返回值轉換為web.Response物件再返回,以保證滿足aiohttp的要求:
async def response_factory(app, handler):
async def response(request):
logging.info('Response handler...')
r = await handler(request)
if isinstance(r, web.StreamResponse):
return r
if isinstance(r, bytes):
resp = web.Response(body=r)
resp.content_type = 'application/octet-stream'
return resp
if isinstance(r, str):
if r.startswith('redirect:'):
return web.HTTPFound(r[9:])
resp = web.Response(body=r.encode('utf-8'))
resp.content_type = 'text/html;charset=utf-8'
return resp
if isinstance(r, dict):
template = r.get('__template__')
if template is None:
resp = web.Response(body=json.dumps(r, ensure_ascii=False, default=lambda o: o.__dict__).encode('utf-8'))
resp.content_type = 'application/json;charset=utf-8'
return resp
else:
resp = web.Response(body=app['__templating__'].get_template(template).render(**r).encode('utf-8'))
resp.content_type = 'text/html;charset=utf-8'
return resp
if isinstance(r, int) and r >= 100 and r < 600:
return web.Response(r)
if isinstance(r, tuple) and len(r) == 2:
t, m = r
if isinstance(t, int) and t >= 100 and t < 600:
return web.Response(t, str(m))
# default:
resp = web.Response(body=str(r).encode('utf-8'))
resp.content_type = 'text/plain;charset=utf-8'
return resp
return response
在廖雪峰老師提供的原始碼中,還有一個叫做data_factory的函式:
async def data_factory(app, handler):
async def parse_data(request):
if request.method == 'POST':
if request.content_type.startswith('application/json'):
request.__data__ = await request.json()
logging.info('request json: %s' % str(request.__data__))
elif request.content_type.startswith('application/x-www-form-urlencoded'):
request.__data__ = await request.post()
logging.info('request form: %s' % str(request.__data__))
return (await handler(request))
return parse_data
不知道老師為什麼在教程中沒有提及。
以目前水平也難以猜測其作用。
小結
框架這一塊比較難以理解,我覺得甚至難於ORM,以至於部落格理了一遍之後還是似懂非懂。今天的梳理參考了一位同學的筆記,對我幫助很大,在此表示感謝。原博傳送門