本篇继续将routing.py的余下内容进行解读
本篇之后有关反向查找的内容暂且忽略
本篇内容比较水，信息量少，不建议看
routing.py→websocket_session
def websocket_session(func: typing.Callable) -> ASGIApp:
"""
获得协程 `func(session)`, 然后返回一个 ASGI 应用.
"""
# assert asyncio.iscoroutinefunction(func), "WebSocket endpoints must be async"
async def app(scope: Scope, receive: Receive, send: Send) -> None:
session = WebSocket(scope, receive=receive, send=send)
await func(session)
return app
WebSocketRoute类
class WebSocketRoute(BaseRoute):
def __init__(
self, path: str, endpoint: typing.Callable, *, name: str = None
) -> None:
assert path.startswith("/"), "Routed paths must start with '/'"
self.path = path
self.endpoint = endpoint
self.name = get_name(endpoint) if name is None else name
if inspect.isfunction(endpoint) or inspect.ismethod(endpoint):
# Endpoint 是一个函数或方法. 把它看作 `func(websocket)`.
self.app = websocket_session(endpoint)
else:
# Endpoint 是一个雷. 把它看作 ASGI 应用.
self.app = endpoint
self.path_regex, self.path_format, self.param_convertors = compile_path(path)
# 同Route
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
if scope["type"] == "websocket":
match = self.path_regex.match(scope["path"])
if match:
matched_params = match.groupdict()
for key, value in matched_params.items():
matched_params[key] = self.param_convertors[key].convert(value)
path_params = dict(scope.get("path_params", {}))
path_params.update(matched_params)
child_scope = {"endpoint": self.endpoint, "path_params": path_params}
return Match.FULL, child_scope
return Match.NONE, {}
# 同Route
def url_path_for(self, name: str, **path_params: str) -> URLPath:
# 反向查找暂且忽略
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
# websocket的处理无需做方法判断
def __eq__(self, other: typing.Any) -> bool:
return (
isinstance(other, WebSocketRoute)
and self.path == other.path
and self.endpoint == other.endpoint
)
Mount类
其准确用处暂且为止，等后续更新
class Mount(BaseRoute):
"""
例:
Route('/', homepage),
Mount('/users', routes=[
Route('/', users),
Route('/{username}', user)
])
用于实现子路由, 他的app是一个router
"""
def __init__(
self,
path: str,
app: ASGIApp = None,
routes: typing.Sequence[BaseRoute] = None,
name: str = None,
) -> None:
assert path == "" or path.startswith("/"), "Routed paths must start with '/'"
assert (
app is not None or routes is not None
), "Either 'app=...', or 'routes=' must be specified"
self.path = path.rstrip("/")
if app is not None:
self.app = app # type: ASGIApp
else:
self.app = Router(routes=routes)
# app和router必须有一个不为空
self.name = name
self.path_regex, self.path_format, self.param_convertors = compile_path(
self.path + "/{path:path}"
)
@property
def routes(self) -> typing.List[BaseRoute]:
return getattr(self.app, "routes", None)
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
if scope["type"] in ("http", "websocket"):
path = scope["path"]
match = self.path_regex.match(path)
if match:
matched_params = match.groupdict()
for key, value in matched_params.items():
matched_params[key] = self.param_convertors[key].convert(value)
remaining_path = "/" + matched_params.pop("path")
matched_path = path[: -len(remaining_path)]
# 将子路由部分截出来
path_params = dict(scope.get("path_params", {}))
path_params.update(matched_params)
root_path = scope.get("root_path", "")
child_scope = {
"path_params": path_params,
"app_root_path": scope.get("app_root_path", root_path),
"root_path": root_path + matched_path,
"path": remaining_path,
"endpoint": self.app,
}
return Match.FULL, child_scope
return Match.NONE, {}
def url_path_for(self, name: str, **path_params: str) -> URLPath:
#反向查找暂且忽略
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
def __eq__(self, other: typing.Any) -> bool:
return (
isinstance(other, Mount)
and self.path == other.path
and self.app == other.app
)
Host类
class Host(BaseRoute):
def __init__(self, host: str, app: ASGIApp, name: str = None) -> None:
self.host = host
self.app = app
self.name = name
self.host_regex, self.host_format, self.param_convertors = compile_path(host)
@property
def routes(self) -> typing.List[BaseRoute]:
return getattr(self.app, "routes", None)
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
if scope["type"] in ("http", "websocket"):
headers = Headers(scope=scope)
host = headers.get("host", "").split(":")[0]
# 创建Headers实例，从中提取host
match = self.host_regex.match(host)
if match:
matched_params = match.groupdict()
for key, value in matched_params.items():
matched_params[key] = self.param_convertors[key].convert(value)
path_params = dict(scope.get("path_params", {}))
path_params.update(matched_params)
child_scope = {"path_params": path_params, "endpoint": self.app}
return Match.FULL, child_scope
return Match.NONE, {}
def url_path_for(self, name: str, **path_params: str) -> URLPath:
#反向查找暂且忽略
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
def __eq__(self, other: typing.Any) -> bool:
return (
isinstance(other, Host)
and self.host == other.host
and self.app == other.app
)
网友评论