aio.app 0.1.8

构建状态

安装

需要python >= 3.4

使用以下方式安装

pip install aio.app

快速开始 - “Hello World”调度器

将以下内容保存到文件“hello.conf”中

[schedule/EXAMPLE]
every = 2
func = my_example.schedule_handler

并将以下内容保存到名为“my_example.py”的文件中

import asyncio

def schedule_handler(event):
    yield from asyncio.sleep(1)
    print ("Received scheduled: %s" % event.name)

使用aio run命令运行

aio run -c hello.conf

aio config命令

在保存或读取配置选项时，配置文件将从以下位置按顺序搜索：

• aio.conf

• etc/aio.conf

• /etc/aio/aio.conf

要转储系统配置，可以运行

aio config

要转储配置部分，可以使用-g或–get与部分名称

aio config -g aio

aio config --get aio/commands

要获取配置选项，可以使用-g与部分名称和选项

aio config -g aio:log_level

aio config --get listen/example:example-signal

您可以使用-s或–set设置配置选项

包含插值的选项应放在单引号内

多行选项应放在“”内，并用“\n”分隔

aio config --set aio:log_level DEBUG

aio config -s aio/otherapp:log_level '${aio:log_level}'

aio config -s listen/example:example-signal "my.listener\nmy.listener2"

如果标准位置中没有配置文件，aio将尝试在当前工作目录中的“aio.conf”中保存

要获取或设置特定文件中的选项，您可以使用-f标志

aio config -g aio:modules -f custom.conf

aio config -s aio:log_level DEBUG -f custom.conf

当使用-f标志获取配置值时，不使用ExtendedInterpolation，因此您看到的是原始值

aio run命令

您可以如下运行aio应用程序：

aio run

或使用自定义配置文件

aio -c custom.conf run

在启动时，aio run会设置以下内容：

• 配置 - 系统级配置

• 模块 - 模块的初始化和配置

• 日志 - 系统日志策略

• 调度器 - 在设定时间调用的函数

• 服务器 - 监听tcp/udp或其他类型的套接字

• 信号 - 对事件响应的函数

[aio]
foo = eggs
     spam

from aio.app import config

[logger_root]
level=${aio:log_level}
handlers=consoleHandler
qualname=aio

[logger_custom]
level=${aio:log_level}
handlers=consoleHandler
qualname=custom

[loggers]
keys=root,custom

[aio]
modules = aio.web.server
        aio.manhole.server

from aio.app import modules

[schedule/SCHEDULE_NAME]

[schedule/example]
every = 2
func = my.scheduler.example_scheduler

def example_scheduler(event):
    yield from asyncio.sleep(2)
    # do something
    print(event.name)
    pass

[server/SERVER_NAME]

[server/example]
protocol = my.example.MyServerProtocol
port = 8888

class MyServerProtocol(asyncio.Protocol):

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        # do stuff
        self.transport.close()

[server/example]
protocol = my.example.protocol
port = 8888

import asyncio
import aio.app

@aio.app.server.protocol
def server_protocol():
    yield from asyncio.sleep(1)
    # do something

    return MyServerProtocol

[server/example]
factory = my.example.server_factory
port = 8080

@aio.app.server.factory
def server_factory(name, protocol, address, port):
    yield from asyncio.sleep(1)
    # do something

    loop = asyncio.get_event_loop()
    return (
        yield from loop.create_server(
           MyServerProtocol, address, port))

[signal/SIGNAL_NAME]

[listen/example]
example-signal = my.example.listener

def listener(signal):
    yield from asyncio.sleep(2)
    print(signal.data)

yield from app.signals.emit(
    'example-signal', "BOOM!")

[listen/example]
example-signal = my.example.listener
example-signal-2 = my.example.listener2
                my.example.listener

[listen/example-2]
example-signal-3 = my.example.listener2

“aio test”命令

您可以测试 aio:modules 配置选项中设置的模块

[aio]
modules = aio.config
         aio.core
         aio.signals

默认情况下，aio test 命令将测试所有测试模块

aio test

您也可以指定一个或多个模块

aio test aio.app

aio test aio.app aio.core

如果您想为测试指定不同于应用程序模块的模块集合，可以在 aio/testing:modules 中列出它们

[aio/testing]
modules = aio.config
         aio.core

这可以包括应用程序模块

[aio/testing]
modules = ${aio:modules}
         aio.web.page
         aio.web.server

依赖项

aio.app 依赖于以下包

• aio.core

• aio.signals

• aio.config

相关软件

• aio.testing

• aio.http.server

• aio.web.server

• aio.manhole.server

aio.app 使用

您可以使用配置中 [aio/commands] 部分列出的任何命令运行 aio 命令

还有 3 个内置命令 - run、config 和 test

最初，aio.app 没有任何配置、信号、模块或服务器

>>> import aio.app

>>> print(aio.app.signals, aio.app.config, aio.app.modules, aio.app.servers)
None None () {}

让我们以默认配置在测试循环中启动应用程序运行器，并打印出信号和配置对象

>>> import aio.testing
>>> from aio.app.runner import runner

>>> @aio.testing.run_until_complete
... def run_app():
...     runner(['run'])
...
...     print(aio.app.signals)
...     print(aio.app.config)
...     print(aio.app.modules)
...     print(aio.app.servers)

>>> run_app()
<aio.signals.Signals object ...>
<configparser.ConfigParser ...>
(<module 'aio.app' from ...>,)
{}

清除应用程序

我们可以清除应用程序变量。

这将关闭任何当前正在运行的套接字服务器

>>> aio.app.clear()

>>> print(aio.app.signals, aio.app.config, aio.app.modules, aio.app.servers)
None None () {}

添加信号监听器

我们可以在应用程序配置中添加信号监听器

>>> config = """
... [listen/testlistener]
... test-signal = aio.app.tests._example_listener
... """

让我们创建一个测试监听器，并使其可导入

监听器需要使用 aio.app.signal.listener 进行包装，并在协程中调用

>>> import asyncio

>>> @aio.app.signal.listener
... def listener(signal):
...     yield from asyncio.sleep(1)
...     print("Listener received: %s" % signal.data)

>>> aio.app.tests._example_listener = listener

运行测试…

>>> @aio.testing.run_until_complete
... def run_app(message):
...     runner(['run'], config_string=config)
...     yield from aio.app.signals.emit('test-signal', message)
...     aio.app.clear()

>>> run_app('BOOM!')
Listener received: BOOM!

我们还可以以编程方式添加监听器

>>> @aio.testing.run_until_complete
... def run_app(message):
...     runner(['run'])
...
...     aio.app.signals.listen('test-signal-2', aio.app.signal.listener(listener))
...     yield from aio.app.signals.emit('test-signal-2', message)
...     aio.app.clear()

>>> run_app('BOOM AGAIN!')
Listener received: BOOM AGAIN!

添加应用程序模块

当您以默认配置运行应用程序时，列出的唯一模块是 aio.app

>>> @aio.testing.run_until_complete
... def run_app(config_string=None):
...     runner(['run'], config_string=config_string)
...     print(aio.app.modules)
...     aio.app.clear()

>>> run_app()
(<module 'aio.app' from ...>,)

我们可以让应用程序运行器了解我们想要包含的任何模块，这些模块在运行时导入

>>> config = """
... [aio]
... modules = aio.app
...          aio.core
... """

>>> run_app(config_string=config)
(<module 'aio.app' from ...>, <module 'aio.core' from ...>)

运行计划程序

计划程序的基本配置

>>> config = """
... [schedule/test-scheduler]
... every: 2
... func: aio.app.tests._example_scheduler
... """

让我们创建一个计划程序函数并使其可导入。

计划程序函数被包装在协程中

>>> def scheduler(event):
...      print('HIT: %s' % event.name)

>>> aio.app.tests._example_scheduler = scheduler

我们需要使用 aio.testing.run_forever 来等待计划事件的发生

>>> @aio.testing.run_forever(timeout=5)
... def run_app():
...     runner(['run'], config_string=config)
...
...     return aio.app.clear

运行 5 秒钟的测试我们得到 3 次命中

>>> run_app()
HIT: test-scheduler
HIT: test-scheduler
HIT: test-scheduler

运行服务器

让我们设置和运行一个附加服务器

至少我们应该提供一个协议和一个要监听的端口

>>> config_server_protocol = """
... [server/additiontest]
... protocol: aio.app.tests._example_AdditionServerProtocol
... port: 8888
... """

让我们创建服务器协议并使其可导入

>>> class AdditionServerProtocol(asyncio.Protocol):
...
...     def connection_made(self, transport):
...         self.transport = transport
...
...     def data_received(self, data):
...         nums = [
...            int(x.strip())
...            for x in
...            data.decode("utf-8").split("+")]
...         self.transport.write(str(sum(nums)).encode())
...         self.transport.close()

>>> aio.app.tests._example_AdditionServerProtocol = AdditionServerProtocol

在设置好服务器后，让我们用简单的加法来调用它

>>> @aio.testing.run_forever
... def run_addition_server(config_string, addition):
...     runner(['run'], config_string=config_string)
...
...     def call_addition_server():
...          reader, writer = yield from asyncio.open_connection(
...              '127.0.0.1', 8888)
...          writer.write(addition.encode())
...          yield from writer.drain()
...          result = yield from reader.read()
...          aio.app.clear()
...
...          print(int(result))
...
...     return call_addition_server

>>> run_addition_server(
...     config_server_protocol,
...     '2 + 2 + 3')
7

如果您需要更多控制服务器协议的创建方式，可以指定一个工厂

>>> config_server_factory = """
... [server/additiontest]
... factory = aio.app.tests._example_addition_server_factory
... port: 8888
... """

工厂方法必须使用 aio.app.server.factory 装饰

>>> @aio.app.server.factory
... def addition_server_factory(name, protocol, address, port):
...     loop = asyncio.get_event_loop()
...     return (
...         yield from loop.create_server(
...            AdditionServerProtocol,
...            address, port))

>>> aio.app.tests._example_addition_server_factory = addition_server_factory

>>> run_addition_server(
...     config_server_protocol,
...     '17 + 5 + 1')
23