跳转至

connection

ConnectionInterface 🔗

Bases: ExportInterface['ElizabethService']

Elizabeth 连接接口

Source code in src/graia/ariadne/connection/__init__.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class ConnectionInterface(ExportInterface["ElizabethService"]):
    """Elizabeth 连接接口"""

    service: ElizabethService
    connection: ConnectionMixin | None

    def __init__(self, service: ElizabethService, account: int | None = None) -> None:
        """初始化连接接口

        Args:
            service (ElizabethService): 连接服务
            account (int, optional): 对应账号
        """
        self.service = service
        self.connection = None
        if account:
            if account not in service.connections:
                raise ValueError(f"Account {account} not found")
            self.connection = service.connections[account]

    def bind(self, account: int) -> Self:
        """绑定账号, 返回一个新实例

        Args:
            account (int): 账号

        Returns:
            ConnectionInterface: 新实例
        """
        return ConnectionInterface(self.service, account)

    async def call(
        self,
        command: str,
        method: CallMethod,
        params: dict,
        *,
        account: int | None = None,
        in_session: bool = True,
    ) -> Any:
        """发起一个调用

        Args:
            command (str): 调用命令
            method (CallMethod): 调用方法
            params (dict): 调用参数
            account (Optional[int], optional): 账号. Defaults to None.
            in_session (bool, optional): 是否在会话中. Defaults to True.

        Returns:
            Any: 调用结果
        """
        if account is None:
            connection = self.connection
        else:
            connection = self.service.connections.get(account)
        if connection is None:
            raise ValueError(f"Unable to find connection to execute {command}")

        return await connection.call(command, method, params, in_session=in_session)

    def add_callback(self, callback: Callable[[MiraiEvent], Awaitable[Any]]) -> None:
        """添加事件回调

        Args:
            callback (Callable[[MiraiEvent], Awaitable[Any]]): 回调函数
        """
        if self.connection is None:
            raise ValueError("Unable to find connection to add callback")
        self.connection.event_callbacks.append(callback)

    @property
    def status(self) -> ConnectionStatus:
        """获取连接状态"""
        if self.connection:
            return self.connection.status
        raise ValueError(f"{self} is not bound to an account")

status property 🔗

status: ConnectionStatus

获取连接状态

__init__ 🔗

__init__(service: ElizabethService, account: int | None = None) -> None

初始化连接接口

Parameters:

Source code in src/graia/ariadne/connection/__init__.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
def __init__(self, service: ElizabethService, account: int | None = None) -> None:
    """初始化连接接口

    Args:
        service (ElizabethService): 连接服务
        account (int, optional): 对应账号
    """
    self.service = service
    self.connection = None
    if account:
        if account not in service.connections:
            raise ValueError(f"Account {account} not found")
        self.connection = service.connections[account]

add_callback 🔗

add_callback(callback: Callable[[MiraiEvent], Awaitable[Any]]) -> None

添加事件回调

Parameters:

Source code in src/graia/ariadne/connection/__init__.py 
185
186
187
188
189
190
191
192
193
def add_callback(self, callback: Callable[[MiraiEvent], Awaitable[Any]]) -> None:
    """添加事件回调

    Args:
        callback (Callable[[MiraiEvent], Awaitable[Any]]): 回调函数
    """
    if self.connection is None:
        raise ValueError("Unable to find connection to add callback")
    self.connection.event_callbacks.append(callback)

bind 🔗

bind(account: int) -> Self

绑定账号, 返回一个新实例

Parameters:

  • account (int) –

    账号

Returns:

  • ConnectionInterface( Self ) –

    新实例

Source code in src/graia/ariadne/connection/__init__.py 
144
145
146
147
148
149
150
151
152
153
def bind(self, account: int) -> Self:
    """绑定账号, 返回一个新实例

    Args:
        account (int): 账号

    Returns:
        ConnectionInterface: 新实例
    """
    return ConnectionInterface(self.service, account)

call async 🔗

call(
    command: str,
    method: CallMethod,
    params: dict,
    *,
    account: int | None = None,
    in_session: bool = True
) -> Any

发起一个调用

Parameters:

  • command (str) –

    调用命令

  • method (CallMethod) –

    调用方法

  • params (dict) –

    调用参数

  • account (Optional[int]) –

    账号. Defaults to None.

  • in_session (bool) –

    是否在会话中. Defaults to True.

Returns:

  • Any( Any ) –

    调用结果

Source code in src/graia/ariadne/connection/__init__.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def call(
    self,
    command: str,
    method: CallMethod,
    params: dict,
    *,
    account: int | None = None,
    in_session: bool = True,
) -> Any:
    """发起一个调用

    Args:
        command (str): 调用命令
        method (CallMethod): 调用方法
        params (dict): 调用参数
        account (Optional[int], optional): 账号. Defaults to None.
        in_session (bool, optional): 是否在会话中. Defaults to True.

    Returns:
        Any: 调用结果
    """
    if account is None:
        connection = self.connection
    else:
        connection = self.service.connections.get(account)
    if connection is None:
        raise ValueError(f"Unable to find connection to execute {command}")

    return await connection.call(command, method, params, in_session=in_session)

ConnectionMixin 🔗

Bases: Launchable, Generic[T_Info]

Source code in src/graia/ariadne/connection/__init__.py 
 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
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class ConnectionMixin(Launchable, Generic[T_Info]):
    status: ConnectionStatus
    info: T_Info
    dependencies: ClassVar[set[str | type[ExportInterface]]]

    fallback: HttpClientConnection | None
    event_callbacks: list[Callable[[MiraiEvent], Awaitable[Any]]]
    _connection_fail: Callable

    @property
    def required(self) -> set[str | type[ExportInterface]]:
        return self.dependencies

    @property
    def stages(self):
        return {}

    def __init__(self, info: T_Info) -> None:
        self.id = ".".join(
            [
                "elizabeth",
                "connection",
                str(info.account),
                camel_to_snake(self.__class__.__qualname__),
            ]
        )
        self.info = info
        self.fallback = None
        self.event_callbacks = []
        self.status = ConnectionStatus()

    async def call(
        self,
        command: str,
        method: CallMethod,
        params: dict | None = None,
        *,
        in_session: bool = True,
    ) -> Any:
        """调用下层 API

        Args:
            command (str): 命令
            method (CallMethod): 调用类型
            params (dict, optional): 调用参数
        """
        if self.fallback:
            return await self.fallback.call(command, method, params, in_session=in_session)
        raise NotImplementedError(
            f"Connection {self} can't perform {command!r}, consider configuring a HttpClientConnection?"
        )

    def __repr__(self) -> str:
        return f"<{self.__class__.__name__} {self.status} with {len(self.event_callbacks)} callbacks>"

call async 🔗

call(
    command: str,
    method: CallMethod,
    params: dict | None = None,
    *,
    in_session: bool = True
) -> Any

调用下层 API

Parameters:

  • command (str) –

    命令

  • method (CallMethod) –

    调用类型

  • params (dict) –

    调用参数

Source code in src/graia/ariadne/connection/__init__.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def call(
    self,
    command: str,
    method: CallMethod,
    params: dict | None = None,
    *,
    in_session: bool = True,
) -> Any:
    """调用下层 API

    Args:
        command (str): 命令
        method (CallMethod): 调用类型
        params (dict, optional): 调用参数
    """
    if self.fallback:
        return await self.fallback.call(command, method, params, in_session=in_session)
    raise NotImplementedError(
        f"Connection {self} can't perform {command!r}, consider configuring a HttpClientConnection?"
    )

ConnectionStatus 🔗

Bases: BaseConnectionStatus, LaunchableStatus

连接状态

Source code in src/graia/ariadne/connection/__init__.py 
20
21
22
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
49
50
51
52
class ConnectionStatus(BaseConnectionStatus, LaunchableStatus):
    """连接状态"""

    alive = Stats[bool]("alive", default=False)

    def __init__(self) -> None:
        self._session_key: str | None = None
        super().__init__()

    @property
    def session_key(self) -> str | None:
        return self._session_key

    @session_key.setter
    def session_key(self, value: str | None) -> None:
        self._session_key = value
        self.connected = value is not None

    @property
    def available(self) -> bool:
        return bool(self.connected and self.session_key and self.alive)

    def __repr__(self) -> str:
        return "<ConnectionStatus {}>".format(
            " ".join(
                [
                    f"connected={self.connected}",
                    f"alive={self.alive}",
                    f"verified={self.session_key is not None}",
                    f"stage={self.stage}",
                ]
            )
        )