aiohttp文件翻譯-server(一)
web server 快速入門
執行一個簡單的web server
為了實現web server, 首先需要實現request handler
一個 request handler 必須是一個coroutine (協程), 它接受一個Request例項作為其唯一引數 ,並返回一個Response 例項,如下程式碼中的hello
from aiohttp import web async def hello(request): return web.Response(text="Hello, world")
接下來,建立一個Application例項並在特定的HTTP方法和路徑上註冊請求處理程式
app = web.Application() app.add_routes([web.get('/', hello)])
然後,通過run_app() 呼叫執行應用程式
web.run_app(app)
這樣,我們就可以通過http://127.0.0.1:8080 檢視結果
或者如果喜歡使用路徑裝飾器的方式,則建立路由表並註冊web處理程式,程式碼實現方式為:
routes = web.RouteTableDef() @routes.get('/') async def hello(request): return web.Response(text="Hello, world") app = web.Application() app.add_routes(routes) web.run_app(app)
這兩種風格做的是同樣的工作,完全看你個人喜好,看你是喜歡Django的urls.py風格的還是Flask的裝飾器風格
Handler
請求處理程式必須是一個協程,它接受一個Request例項作為其唯一引數,並返回一個StreamResponse派生(例如,響應)例項
async def handler(request): return web.Response()
Handlers 通過使用get() 和post() 等方法將程式在特定路由(HTTP方法和路徑對) 上使用Application.add_routes()註冊它們來處理請求
app.add_routes([web.get('/', handler), web.post('/post', post_handler), web.put('/put', put_handler)])
當然你也可以使用裝飾器方式
routes = web.RouteTableDef() @routes.get('/') async def get_handler(request): ... @routes.post('/post') async def post_handler(request): ... @routes.put('/put') async def put_handler(request): ... app.add_routes(routes)
route()或RouteTableDef.route()也支援萬用字元HTTP方法,允許處理程式在具有任何HTTP方法的路徑上提供傳入請求
app.add_routes[web.route('*', '/path', all_handler)]
預設情況下,使用GET方法新增的端點將接受HEAD請求並返回與GET請求相同的響應頭。您還可以拒絕路由上的HEAD請求:
web.get('/', handler, allow_head=False)
這裡不會在HEAD請求上呼叫處理程式,伺服器將響應405:Method Not Allowed。
Resources and Routes
內部路由由Application.router(UrlDispatcher例項)提供。
路由器是資源列表
資源是路由表中的條目,對應於請求的URL。
資源至少有一條路由
Route對應於通過呼叫web處理程式來處理HTTP方法。
這個庫實現合併相同路徑的路由,為所有HTTP方法新增唯一資源。
考慮兩個例子:
app.add_routes([web.get('/path1', get_1), web.post('/path1', post_1), web.get('/path2', get_2), web.post('/path2', post_2)]
app.add_routes([web.get('/path1', get_1), web.get('/path2', get_2), web.post('/path2', post_2), web.post('/path1', post_1)]
Variable Resources
資源也可能具有可變路徑。例如,路徑為“/ a / {name} / c”的資源會匹配所有傳入請求,其路徑為“/ a / b / c”,“/ a / 1 / c”和“/ a /”
變數部分以{identifier}形式指定,其中識別符號稍後可以在請求處理程式中使用,以訪問該部分的匹配值。這是通過在Request.match_info對映中查詢識別符號來完成的:
@routes.get('/{name}') async def variable_handler(request): return web.Response( text="Hello, {}".format(request.match_info['name']))
預設情況下,每個部分都匹配正則表示式[^ {} /] +。你還可以使用{identifier:regex}格式指定自定義正則表示式:
web.get(r'/{name:\d+}', handler)
Reverse URL Constructing using Named Resources
routes也可以給出一個名字:
@routes.get('/root', name='root') async def handler(request): ...
然後可以使用它來訪問和構建該資源的URL(例如在請求處理程式中):
url == request.app.router['root'].url_for().with_query({"a": "b", "c": "d"}) assert url == URL('/root?a=b&c=d')
一個更有趣的例子是為可變資源構建URL:
app.router.add_resource(r'/{user}/info', name='user-info')
在這種情況下,你還可以傳遞路線的各個部分:
url = request.app.router['user-info'].url_for(user='john_doe') url_with_qs = url.with_query("a=b") assert url_with_qs == '/john_doe/info?a=b'
Organizing Handlers in Classes
class Handler: def __init__(self): pass async def handle_intro(self, request): return web.Response(text="Hello, world") async def handle_greeting(self, request): name = request.match_info.get('name', "Anonymous") txt = "Hello, {}".format(name) return web.Response(text=txt) handler = Handler() app.add_routes([web.get('/intro', handler.handle_intro), web.get('/greet/{name}', handler.handle_greeting)]
Class Based Views
aiohttp.web支援基於類的檢視。你可以從View派生並定義處理http請求的方法:
class MyView(web.View): async def get(self): return await get_resp(self.request) async def post(self): return await post_resp(self.request)
處理程式應該是coroutines只接受self並返回響應物件作為常規Web處理程式。View.request屬性可以檢索請求物件。實現檢視後(上面的例子中的MyView)應該在應用程式的路由器中註冊:
web.view('/path/to', MyView)
或者
@routes.view('/path/to') class MyView(web.View): ...
Resource Views
可以使用UrlDispatcher.resources()方法檢視路由器中的所有註冊資源
for resource in app.router.resources(): print(resource)
可以使用UrlDispatcher.named_resources()方法檢視使用名稱註冊的資源的子集:
for name, resource in app.router.named_resources().items(): print(name, resource)
Alternative ways for registering routes
上面顯示的程式碼示例使用命令式樣式新增新路由:它們呼叫app.router.add_get(...)等。有兩種選擇:路由表和路由裝飾器。路由表看起來像Django方式:
async def handle_get(request): ... async def handle_post(request): ... app.router.add_routes([web.get('/get', handle_get), web.post('/post', handle_post),
該片段呼叫add_routes()來註冊由aiohttp.web.get()或aiohttp.web.post()函式建立的路由定義列表(aiohttp.web.RouteDef例項)。
路由裝飾器和Flask更像
routes = web.RouteTableDef() @routes.get('/get') async def handle_get(request): ... @routes.post('/post') async def handle_post(request): ... app.router.add_routes(routes)
當然你可以在類視圖裡使用裝飾器的方式
routes = web.RouteTableDef() @routes.view("/view") class MyView(web.View): async def get(self): ... async def post(self): ... app.router.add_routes(routes)
該示例首先建立一個aiohttp.web.RouteTableDef容器。容器是一個類似列表的物件,帶有額外的裝飾器aiohttp.web.RouteTableDef.get(),aiohttp.web.RouteTableDef.post()等,用於註冊新路由。填充容器後,add_routes()用於將已註冊的路由定義新增到應用程式的路由器中。
JSON Response
返回JSON資料是一種常見的情況,aiohttp.web提供了返回JSON的快捷方式 - aiohttp.web.json_response():
async def handler(request): data = {'some': 'data'} return web.json_response(data)
這個快捷方法返回aiohttp.web.Response例項,因此你可以在從處理程式返回cookie之前設定cookie。
User Sessions
通常,您需要一個容器來跨請求儲存使用者資料。該概念通常稱為會話。aiohttp.web沒有內建的會話概念,但是,有一個第三方庫aiohttp_session,它增加了會話支援:
import asyncio import time import base64 from cryptography import fernet from aiohttp import web from aiohttp_session import setup, get_session, session_middleware from aiohttp_session.cookie_storage import EncryptedCookieStorage async def handler(request): session = await get_session(request) last_visit = session['last_visit'] if 'last_visit' in session else None text = 'Last visited: {}'.format(last_visit) return web.Response(text=text) async def make_app(): app = web.Application() # secret_key must be 32 url-safe base64-encoded bytes fernet_key = fernet.Fernet.generate_key() secret_key = base64.urlsafe_b64decode(fernet_key) setup(app, EncryptedCookieStorage(secret_key)) app.add_routes([web.get('/', handler)]) return app web.run_app(make_app())
HTTP Forms
開箱即用支援HTTP表單。
如果form的方法是“GET”(<form method =“get”>),請使用Request.query獲取表單資料。要使用“POST”方法訪問表單資料,請使用Request.post()或Request.multipart()。
Request.post()接受'application / x-www-form-urlencoded'和'multipart / form-data'表單的資料編碼(例如<form enctype =“multipart / form-data”>)。它將檔案資料儲存在臨時目錄中。如果在引發ValueError異常後指定了client_max_size。
為了提高效率,請使用Request.multipart(),它對於上傳大檔案(檔案上傳)特別有效。
通過以下表格提交的值:
<form action="/login" method="post" accept-charset="utf-8" enctype="application/x-www-form-urlencoded"> <label for="login">Login</label> <input id="login" name="login" type="text" value="" autofocus/> <label for="password">Password</label> <input id="password" name="password" type="password" value=""/> <input type="submit" value="login"/> </form>
async def do_login(request): data = await request.post() login = data['login'] password = data['password']
File Uploads
aiohttp.web內建支援處理從瀏覽器上傳的檔案。首先,確保HTML <form>元素的enctype屬性設定為enctype =“multipart / form-data”。
例如,這是一個接受MP3檔案的表單:
<form action="/store/mp3" method="post" accept-charset="utf-8" enctype="multipart/form-data"> <label for="mp3">Mp3</label> <input id="mp3" name="mp3" type="file" value=""/> <input type="submit" value="submit"/> </form>
然後,在請求處理程式中,你可以將檔案輸入欄位作為FileField例項進行訪問。FileField只是檔案的容器及其一些元資料:
async def store_mp3_handler(request): # WARNING: don't do that if you plan to receive large files! data = await request.post() mp3 = data['mp3'] # .filename contains the name of the file in string format. filename = mp3.filename # .file contains the actual file data that needs to be stored somewhere. mp3_file = data['mp3'].file content = mp3_file.read() return web.Response(body=content, headers=MultiDict( {'CONTENT-DISPOSITION': mp3_file}))
您可能已經注意到上面示例中的一個重要警告。一般問題是Request.post()讀取記憶體中的整個有效負載,導致可能的OOM錯誤。為避免這種情況,對於分段上傳,您應該使用Request.multipart()來返回多部分閱讀器:
async def store_mp3_handler(request): reader = await request.multipart() # /!\ Don't forget to validate your inputs /!\ # reader.next() will `yield` the fields of your form field = await reader.next() assert field.name == 'name' name = await field.read(decode=True) field = await reader.next() assert field.name == 'mp3' filename = field.filename # You cannot rely on Content-Length if transfer is chunked. size = 0 with open(os.path.join('/spool/yarrr-media/mp3/', filename), 'wb') as f: while True: chunk = await field.read_chunk()# 8192 bytes by default. if not chunk: break size += len(chunk) f.write(chunk) return web.Response(text='{} sized of {} successfully stored' ''.format(filename, size))
Socket/">WebSockets
aiohttp.web支援WebSockets開箱即用。要設定WebSocket,請在請求處理程式中建立WebSocketResponse,然後使用它與對等方進行通訊:
async def websocket_handler(request): ws = web.WebSocketResponse() await ws.prepare(request) async for msg in ws: if msg.type == aiohttp.WSMsgType.TEXT: if msg.data == 'close': await ws.close() else: await ws.send_str(msg.data + '/answer') elif msg.type == aiohttp.WSMsgType.ERROR: print('ws connection closed with exception %s' % ws.exception()) print('websocket connection closed') return ws
處理程式應註冊為HTTP GET處理器
app.add_routes([web.get('/ws', websocket_handler)])
Redirects
將使用者重定向到另一個端點 - 使用絕對URL,相對URL或檢視名稱(來自路由器的引數)引發HTTPFound:
raise web.HTTPFound('/redirect')
以下示例顯示重定向到路徑中名為'login'的檢視:
async def handler(request): location = request.app.router['login'].url_for() raise web.HTTPFound(location=location) router.add_get('/handler', handler) router.add_get('/login', login_handler, name='login')
登入驗證示例
@aiohttp_jinja2.template('login.html') async def login(request): if request.method == 'POST': form = await request.post() error = validate_login(form) if error: return {'error': error} else: # login form is valid location = request.app.router['index'].url_for() raise web.HTTPFound(location=location) return {} app.router.add_get('/', index, name='index') app.router.add_get('/login', login, name='login') app.router.add_post('/login', login, name='login')