跳转至

runner

OneBot Connect 连接 Runner。

ClientRunner

Bases: Runner

客户端运行器。

Attributes:

Name Type Description
task_manager TaskManager

任务管理器
Source code in src/pylibob/runner.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class ClientRunner(Runner):
    """客户端运行器。

    Attributes:
        task_manager (TaskManager): 任务管理器
    """

    def __init__(self, task_manager: TaskManager) -> None:
        """初始化客户端运行器。

        Args:
            task_manager (TaskManager): 任务管理器
        """
        self.should_exit: asyncio.Event = asyncio.Event()
        self.force_exit: bool = False
        self.task_manager = task_manager
        self._lifespan_manager = LifespanManager()

    @property
    def lifespan_manager(self) -> LifespanManager:
        return self._lifespan_manager

    async def _loop(self):
        await self.should_exit.wait()

    def _handle_exit(self, sig, frame):
        self.should_exit.set()

    async def _shutdown(self):
        await self.lifespan_manager.shutdown()
        self.task_manager.cancel_all()

    def on_startup(self, func: L_FUNC):
        self.lifespan_manager.on_startup(func)

    def on_shutdown(self, func: L_FUNC):
        self.lifespan_manager.on_shutdown(func)

    async def run(self):
        for sig in HANDLED_SIGNALS:
            logger.debug(f"注册信号 {sig} 捕获器")
            signal.signal(sig, self._handle_exit)
        logger.info("启动 ClientRunner")
        await self.lifespan_manager.startup()
        await self._loop()
        await self._shutdown()

__init__(task_manager)

初始化客户端运行器。

Parameters:

Name Type Description Default
task_manager TaskManager

任务管理器

 required
Source code in src/pylibob/runner.py 
101
102
103
104
105
106
107
108
109
110
def __init__(self, task_manager: TaskManager) -> None:
    """初始化客户端运行器。

    Args:
        task_manager (TaskManager): 任务管理器
    """
    self.should_exit: asyncio.Event = asyncio.Event()
    self.force_exit: bool = False
    self.task_manager = task_manager
    self._lifespan_manager = LifespanManager()

Runner

Bases: ABC

抽象运行器基类。

Attributes:

Name Type Description
lifespan_manager LifespanManager

异步生命周期管理器
Source code in src/pylibob/runner.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class Runner(abc.ABC):
    """抽象运行器基类。

    Attributes:
        lifespan_manager (LifespanManager): 异步生命周期管理器
    """

    @abc.abstractmethod
    async def run(self) -> None:
        """启动运行器。"""
        raise NotImplementedError

    @abc.abstractmethod
    def on_startup(self, func: L_FUNC) -> None:
        """添加 startup 生命周期函数。"""
        raise NotImplementedError

    @abc.abstractmethod
    def on_shutdown(self, func: L_FUNC) -> None:
        """添加 shutdown 生命周期。"""
        raise NotImplementedError

    @abc.abstractproperty
    def lifespan_manager(self) -> LifespanManager:
        """异步生命周期管理器。"""
        raise NotImplementedError

lifespan_manager()

异步生命周期管理器。

Source code in src/pylibob/runner.py 
45
46
47
48
@abc.abstractproperty
def lifespan_manager(self) -> LifespanManager:
    """异步生命周期管理器。"""
    raise NotImplementedError

on_shutdown(func) abstractmethod

添加 shutdown 生命周期。

Source code in src/pylibob/runner.py 
40
41
42
43
@abc.abstractmethod
def on_shutdown(self, func: L_FUNC) -> None:
    """添加 shutdown 生命周期。"""
    raise NotImplementedError

on_startup(func) abstractmethod

添加 startup 生命周期函数。

Source code in src/pylibob/runner.py 
35
36
37
38
@abc.abstractmethod
def on_startup(self, func: L_FUNC) -> None:
    """添加 startup 生命周期函数。"""
    raise NotImplementedError

run() abstractmethod async

启动运行器。

Source code in src/pylibob/runner.py 
30
31
32
33
@abc.abstractmethod
async def run(self) -> None:
    """启动运行器。"""
    raise NotImplementedError

ServerRunner

Bases: Runner

服务器运行器。

Attributes:

Name Type Description
host str

服务器监听 IP
port int

服务器监听端口
uvicorn_params dict[str, Any]

传入到 uvicorn 的其他参数
Source code in src/pylibob/runner.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class ServerRunner(Runner):
    """服务器运行器。

    Attributes:
        host (str): 服务器监听 IP
        port (int): 服务器监听端口
        uvicorn_params (dict[str, Any]): 传入到 uvicorn 的其他参数
    """

    def __init__(self, host: str, port: int, **kwargs: Any) -> None:
        """初始化服务器运行器。

        Args:
            host (str): 服务器监听 IP
            port (int): 服务器监听端口
            **kwargs: 传入到 uvicorn 的其他参数
        """
        self.host = host
        self.port = port
        self.uvicorn_params: dict[str, Any] = kwargs

    async def run(self):
        logger.info("启动 ServerRunner")
        await uvicorn.Server(
            uvicorn.Config(
                asgi_app,
                host=self.host,
                port=self.port,
                **self.uvicorn_params,
            ),
        ).serve()

    def on_startup(self, func: L_FUNC):
        asgi_lifespan_manager.on_startup(func)

    def on_shutdown(self, func: L_FUNC):
        asgi_lifespan_manager.on_shutdown(func)

    @property
    def lifespan_manager(self) -> LifespanManager:
        return asgi_lifespan_manager

__init__(host, port, **kwargs)

初始化服务器运行器。

Parameters:

Name Type Description Default
host str

服务器监听 IP

 required
port int

服务器监听端口

 required
**kwargs Any

传入到 uvicorn 的其他参数

 {}
Source code in src/pylibob/runner.py 
60
61
62
63
64
65
66
67
68
69
70
def __init__(self, host: str, port: int, **kwargs: Any) -> None:
    """初始化服务器运行器。

    Args:
        host (str): 服务器监听 IP
        port (int): 服务器监听端口
        **kwargs: 传入到 uvicorn 的其他参数
    """
    self.host = host
    self.port = port
    self.uvicorn_params: dict[str, Any] = kwargs