aio asyncio框架的Web服务器
项目描述
为aio asyncio框架提供的Web服务器
构建状态
安装
需要python >= 3.4
使用以下命令安装
pip install aio.web.server
快速入门 - Hello world Web服务器
创建一个问候语的Web服务器
将以下内容保存到文件“hello.conf”中
[aio]
modules = aio.web.server
[server/my_server]
factory = aio.web.server.factory
port = 8080
[web/my_server/my_route]
match = /
route = my_example.handler
并将以下内容保存到名为my_example.py的文件中
import aiohttp
import aio.web.server
@aio.web.server.route
def handler(request, config):
return aiohttp.web.Response(body=b"Hello, web world")
使用aio run命令运行
aio run -c hello.conf
Web服务器配置
Web服务器定义以下格式,其中SERVER_NAME对应于server/SERVER_NAME
[web/SERVER_NAME]
例如,你可能会有以下内容
[server/my_server]
factory = aio.web.server.factory
port = 8080
[web/my_server]
modules = ${aio:modules}
some.web.module
路由配置
路由定义在以下格式的部分中
[web/SERVER_NAME/ROUTE_NAME]
因此,以下示例服务器配置中定义了路径/的路由
[aio]
modules = aio.web.server
[server/my_server]
factory = aio.web.server.factory
port = 8080
[web/my_server/my_route]
match = /
route = my.route.handler
aio.web.server使用方法
配置
要设置Web服务器,我们需要
将“aio.web.server”添加到aio:modules初始化Web服务器
添加一个“server/SERVERNAME”部分以创建HTTP服务器
添加一个“web/SERVERNAME/ROUTENAME”以创建路由
让我们创建一个基本的Web服务器配置
>>> web_server_config = """ ... [aio] ... log_level = ERROR ... modules = aio.web.server ... ... [server/server_name] ... factory = aio.web.server.factory ... port = 7070 ... ... [web/server_name/route_name] ... match = / ... route = aio.web.server.tests._example_handler ... """
现在让我们创建一个路由并使其可导入
>>> import aiohttp >>> import aio.web.server
>>> @aio.web.server.route ... def route_handler(route): ... return aiohttp.web.Response(body=b"Hello, web world")
>>> import aio.web.server.tests >>> aio.web.server.tests._example_handler = route_handler
让我们设置一个测试以运行服务器并请求网页
>>> from aio.app.runner import runner >>> import aio.testing
>>> @aio.testing.run_forever(sleep=1) ... def run_web_server(config, request_page="https://127.0.0.1:7070"): ... runner(['run'], config_string=config) ... ... def call_web_server(): ... result = yield from ( ... yield from aiohttp.request( ... "GET", request_page)).read() ... ... print(result.decode()) ... ... return call_web_server
并运行测试
>>> run_web_server(web_server_config) Hello, web world
我们可以通过名称访问aiohttp Web应用程序
>>> import aio.web.server >>> web_app = aio.web.server.apps['server_name'] >>> web_app <Application>
>>> web_app['name'] 'server_name'
并且我们可以访问Web应用程序的jinja环境
>>> import aiohttp_jinja2 >>> jinja_env = aiohttp_jinja2.get_env(web_app) >>> jinja_env <jinja2.environment.Environment object ...>
我们还没有注册任何模板
>>> jinja_env.list_templates() []
让我们清除Web应用程序,这将也会调用aio.app.clear()
>>> aio.web.server.clear() >>> aio.web.server.apps {}
>>> print(aio.app.config, aio.app.signals) None None
Web应用程序模块
默认情况下,为aio:modules中列出的任何模块注册了模板资源
>>> config = """ ... [aio] ... modules = aio.web.server ... aio.web.server.tests ... ... [server/server_name] ... factory = aio.web.server.factory ... port = 7070 ... """
让我们创建一个测试来运行服务器并打印已安装的 Jinja 模板列表
>>> @aio.testing.run_forever(sleep=1) ... def run_server_print_templates(config_string): ... runner(['run'], config_string=config_string) ... ... def print_templates(): ... web_app = aio.web.server.apps['server_name'] ... print( ... [x for x in ... aiohttp_jinja2.get_env( ... web_app).list_templates(extensions=["html"])]) ... aio.web.server.clear() ... ... return print_templates
aio.web.server.tests 模块包含 2 个 HTML 模板
>>> run_server_print_templates(config) ['fragments/test_fragment.html', 'test_template.html']
我们可以在 aio/web:modules 选项中设置所有 Web 应用的模块
这将覆盖 aio:modules 中的设置
>>> config = """ ... [aio] ... modules = aio.web.server ... aio.web.server.tests ... ... [aio/web] ... modules = aio.web.server ... ... [server/server_name] ... factory = aio.web.server.factory ... port = 7070 ... """
>>> run_server_print_templates(config) []
或者,您可以在 web/SERVER_NAME:modules 选项中设置模块。
这将覆盖 aio/web:modules 和 aio:modules 中的设置
>>> config = """ ... [aio] ... modules = aio.web.server ... aio.web.server.tests ... ... [aio/web] ... modules = aio.web.server ... ... [web/server_name] ... modules = aio.web.server.tests ... ... [server/server_name] ... factory = aio.web.server.factory ... port = 7070 ... """
>>> run_server_print_templates(config) ['fragments/test_fragment.html', 'test_template.html']
路由
>>> config_template = """ ... [aio] ... modules = aio.web.server ... aio.web.server.tests ... log_level: ERROR ... ... [server/server_name] ... factory: aio.web.server.factory ... port: 7070 ... ... [web/server_name/route_name] ... match = / ... route = aio.web.server.tests._example_route_handler ... """
路由函数必须使用 aio.server.route 装饰,并接收一个 aio.server.Route 对象
路由对象有一个包含路由配置的 request 属性和 config 属性
>>> @aio.web.server.route("test_template.html") ... def route_handler(route): ... return { ... 'message': 'Hello, world at %s from match(%s) handled by: %s' % ( ... route.request.path, route.config['match'], route.config['route'])}
>>> aio.web.server.tests._example_route_handler = route_handler
>>> run_web_server(config_template) <html> <body> Hello, world at / from match(/) handled by: aio.web.server.tests._example_route_handler </body> </html>
>>> aio.web.server.clear()
静态目录
web/SERVER_NAME 部分接受 static_url 和 static_dir 选项来托管静态文件
>>> config_static = """ ... [aio] ... log_level: ERROR ... modules = aio.web.server ... ... [server/test] ... factory: aio.web.server.factory ... port: 7070 ... ... [web/test] ... static_url: /static ... static_dir: %s ... """
>>> import os >>> import tempfile
让我们创建一个临时目录并向其中添加一个 CSS 文件
>>> with tempfile.TemporaryDirectory() as tmp: ... with open(os.path.join(tmp, "test.css"), 'w') as cssfile: ... result = cssfile.write("body {background: black}") ... ... run_web_server( ... config_static % tmp, ... request_page="https://127.0.0.1:7070/static/test.css") body {background: black}
>>> aio.web.server.clear()
模板过滤器
您可以通过将它们添加到 aio/web:filters 选项来配置 Jinja 过滤器
>>> config = """ ... [aio] ... log_level: ERROR ... modules = aio.web.server ... ... [server/server_name] ... factory: aio.web.server.factory ... port: 7070 ... ... [aio/web] ... filters = example_filter aio.web.server.tests._example_filter ... """
过滤器 不在 协程中调用
>>> def filter(value, *la): ... return value
>>> aio.web.server.tests._example_filter = filter
>>> @aio.testing.run_forever(sleep=1) ... def run_server_check_filter(config_string): ... runner(['run'], config_string=config_string) ... ... def check_filter(): ... web_app = aio.web.server.apps['server_name'] ... env = aiohttp_jinja2.get_env(web_app) ... ... if "example_filter" in env.filters.keys(): ... print("example_filter is in the jinja environment!") ... ... return check_filter
>>> run_server_check_filter(config) example_filter is in the jinja environment!
>>> aio.web.server.clear()
您还可以将过滤器添加到 web/server_name 部分,这将覆盖 aio/web 中的设置
>>> config = """ ... [aio] ... log_level: ERROR ... modules = aio.web.server ... ... [server/server_name] ... factory: aio.web.server.factory ... port: 7070 ... ... [aio/web] ... filters = example_filter aio.web.server.tests._example_filter ... ... [web/server_name] ... filters = example_filter_2 aio.web.server.tests._example_filter ... """
>>> @aio.testing.run_forever(sleep=1) ... def run_server_check_filter(config_string): ... runner(['run'], config_string=config_string) ... ... def check_filter(): ... web_app = aio.web.server.apps['server_name'] ... env = aiohttp_jinja2.get_env(web_app) ... ... if "example_filter" not in env.filters.keys(): ... print("example_filter is not in the jinja environment!") ... ... if "example_filter_2" in env.filters.keys(): ... print("example_filter_2 is in the jinja environment!") ... ... return check_filter
>>> run_server_check_filter(config) example_filter is not in the jinja environment! example_filter_2 is in the jinja environment!
>>> aio.web.server.clear()
项目详情
aio.web.server-0.1.3.tar.gz 的哈希值
算法 | 哈希摘要 | |
---|---|---|
SHA256 | c25421bcd2ac2328df7c450d476308bbbfc27fa11af32247f825265df9362686 |
|
MD5 | a31839c4f98a71fe27e76223b17c585b |
|
BLAKE2b-256 | 1ad4de3725065cd3b483052d41f680b074a05d68bb4b4baa964b8a985aa216c2 |