connection

ConnectionInterface

Bases: ExportInterface['ElizabethService']

Elizabeth 连接接口

Source code in src/graia/ariadne/connection/__init__.py

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: ConnectionStatus property

获取连接状态

__init__(service, account=None)

初始化连接接口

Parameters:

Name	Type	Description	Default
service	ElizabethService	连接服务	required
account	int	对应账号	None

Source code in src/graia/ariadne/connection/__init__.py

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(callback)

添加事件回调

Parameters:

Name	Type	Description	Default
callback	Callable[[MiraiEvent], Awaitable[Any]]	回调函数	required

Source code in src/graia/ariadne/connection/__init__.py

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(account)

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

Parameters:

Name	Type	Description	Default
account	int	账号	required

Returns:

Name	Type	Description
ConnectionInterface	Self	新实例

Source code in src/graia/ariadne/connection/__init__.py

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

    Args:
        account (int): 账号

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

call(command, method, params, *, account=None, in_session=True) async

发起一个调用

Parameters:

Name	Type	Description	Default
command	str	调用命令	required
method	CallMethod	调用方法	required
params	dict	调用参数	required
account	Optional[int]	账号. Defaults to None.	None
in_session	bool	是否在会话中. Defaults to True.	True

Returns:

Name	Type	Description
Any	Any	调用结果

Source code in src/graia/ariadne/connection/__init__.py

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

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(command, method, params=None, *, in_session=True) async

调用下层 API

Parameters:

Name	Type	Description	Default
command	str	命令	required
method	CallMethod	调用类型	required
params	dict	调用参数	None

Source code in src/graia/ariadne/connection/__init__.py

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: ConnectionStatus, LaunchableStatus

连接状态

Source code in src/graia/ariadne/connection/__init__.py

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}",
                ]
            )
        )