client

async def close(self):
    """关闭客户端连接."""
    await self._session.close()

async def execute(self, request: BaseRequest[ResultT]) -> ResultT:
    """执行单个请求描述符并解析响应结果.

    Args:
        request: 请求描述符实例.
    """
    match request:
        case CgiRequest():
            if request.require_login:
                cred = request.credential or self._context.credential
                if not cred or not cred.musicid or not cred.musickey:
                    from .exceptions import CredentialInvalidError

                    raise CredentialInvalidError("请求需要登录, 未提供有效的登录凭证")

            req_item = request._build_args()

            url, payload, params, headers = await self._context.build_api_kwargs(
                data=[req_item],
                comm=request.comm,
                credential=request.credential,
                platform=request.platform,
                override_comm=request.override_comm,
                sign=request.sign,
            )

            try:
                resp = await self._session.post(
                    url,
                    json=payload,
                    params=params,
                    headers=headers,
                    proxies=self.proxies,
                    hooks=self.hooks,
                    cert=self.cert,
                    verify=self.verify,
                )
                await self._session.gather(resp)
            except RequestException as exc:
                raise NetworkError(str(exc)) from exc

            raw_data = self._unwrap_cgi_batch(resp, expected_count=1)[0]

            return request._parse_response(raw_data)

        case HttpRequest():
            request = cast("HttpRequest", request)
            kwargs = await self._context.prepare_http_kwargs(
                credential=request.credential,
                **request._build_args(),
            )

            try:
                resp = await self._session.request(
                    request.method,
                    request.url,
                    **kwargs,
                    proxies=self.proxies,
                    hooks=self.hooks,
                    cert=self.cert,
                    verify=self.verify,
                )
                await self._session.gather(resp)
            except RequestException as exc:
                raise NetworkError(str(exc)) from exc

            return request._parse_response(resp)
        case _:
            raise TypeError(f"不支持的请求类型: {type(request)}")

async def gather(
    self,
    requests: list[BaseRequest[Any]],
    *,
    batch_size: int = 20,
    return_exceptions: bool = False,
) -> list[Any]:
    """并发执行多个请求描述符并按输入顺序返回解析结果.

    CGI 请求会按可合并条件自动分组, 同一分组内的请求按 `batch_size`
    批量合并为一次 CGI 多参数调用 (req_0, req_1, ...), 以减少网络往返;
    不同分组之间并发执行. HTTP 请求不参与合并, 直接并发执行.

    Args:
        requests: 待执行的请求描述符列表.
        batch_size: 单个 CGI 批量调用 (多参数合并) 包含的最大请求数; 仅对
            CGI 请求生效, 不影响 HTTP 请求.
        return_exceptions: 是否捕捉异常并作为结果返回而不抛出. 为 True 时,
            请求构造、网络传输、响应解析等所有异常都会被写入对应位置的结果;
            为 False 时, 任一请求的异常会以异常组形式抛出.

    Returns:
        与 `requests` 顺序一致的解析结果列表. 当 `return_exceptions` 为
        True 时, 失败位置的结果为对应的异常对象.

    Raises:
        ValueError: 当 `batch_size` 小于等于 0 时抛出.
        ExceptionGroup: 当 `return_exceptions` 为 False 且任一请求执行
            期间发生异常时, 其余并发请求会被取消, 失败异常会以异常组的
            形式抛出 (anyio 将异常包装为 `ExceptionGroup`, 它是
            `BaseExceptionGroup` 的子类; 即使只有一个请求失败也会被包装
            成异常组; 多个请求同时各自抛出异常时, 异常组可能包含多个
            异常).
        ApiDataError: 当内部依赖的结果未能完整回填时抛出 (一般不应发生).
    """
    if batch_size <= 0:
        raise ValueError("batch_size 必须大于 0")
    if not requests:
        return []

    results: list[Any] = [MISSING] * len(requests)
    all_task: defaultdict[str, list[tuple[int, BaseRequest[Any]]]] = defaultdict(list)
    for idx, req in enumerate(requests):
        all_task[req._protocol].append((idx, req))

    async def _gather_cgi(tasks: list[tuple[int, CgiRequest]]):
        batch_responses = []
        grouped_indices: defaultdict[Any, list[tuple[int, CgiRequest[Any]]]] = defaultdict(list)
        for orig_idx, req in tasks:
            if req.require_login:
                cred = req.credential or self._context.credential
                if not cred or not cred.musicid or not cred.musickey:
                    from .exceptions import CredentialInvalidError

                    exc = CredentialInvalidError("请求需要登录, 未提供有效的登录凭证")
                    if return_exceptions:
                        results[orig_idx] = exc
                        continue
                    raise exc
            grouped_indices[req._group_key].append((orig_idx, req))

        for group in grouped_indices.values():
            base_req = group[0][1]
            for start in range(0, len(group), batch_size):
                chunk = group[start : start + batch_size]
                chunk_orig_indices = [item[0] for item in chunk]

                url, payload, params, headers = await self._context.build_api_kwargs(
                    data=[r[1]._build_args() for r in chunk],
                    comm=base_req.comm,
                    credential=base_req.credential,
                    platform=base_req.platform,
                    override_comm=base_req.override_comm,
                    sign=base_req.sign,
                )

                try:
                    resp = await self._session.post(
                        url,
                        json=payload,
                        params=params,
                        headers=headers,
                        proxies=self.proxies,
                        hooks=self.hooks,
                        cert=self.cert,
                        verify=self.verify,
                    )
                except RequestException as exc:
                    error = NetworkError(str(exc))
                    if return_exceptions:
                        for req_index in chunk_orig_indices:
                            results[req_index] = error
                        continue
                    raise error from exc
                batch_responses.append((chunk_orig_indices, resp))

        if not batch_responses:
            return

        try:
            await self._session.gather(*(resp for _, resp in batch_responses))
        except RequestException as exc:
            error = NetworkError(str(exc))
            if return_exceptions:
                for batch_indices, _ in batch_responses:
                    for req_index in batch_indices:
                        results[req_index] = error
                return
            raise error from exc

        for batch_indices, response in batch_responses:
            try:
                data = self._unwrap_cgi_batch(response, len(batch_indices))
            except Exception as exc:
                if return_exceptions:
                    for req_index in batch_indices:
                        results[req_index] = exc
                    continue
                raise

            for batch_index, req_index in enumerate(batch_indices):
                request = cast("CgiRequest", requests[req_index])
                try:
                    results[req_index] = request._parse_response(data[batch_index])
                except Exception as exc:
                    if return_exceptions:
                        results[req_index] = exc
                    else:
                        raise

    async def _gather_http(tasks: list[tuple[int, HttpRequest]]):
        http_responses = []
        for orig_idx, req in tasks:
            kwargs = await self._context.prepare_http_kwargs(
                credential=req.credential,
                **req._build_args(),
            )

            try:
                resp = await self._session.request(
                    req.method,
                    req.url,
                    **kwargs,
                    proxies=self.proxies,
                    hooks=self.hooks,
                    cert=self.cert,
                    verify=self.verify,
                )
            except RequestException as exc:
                error = NetworkError(str(exc))
                if return_exceptions:
                    results[orig_idx] = error
                    continue
                raise error from exc
            http_responses.append((orig_idx, req, resp))

        if not http_responses:
            return

        try:
            await self._session.gather(*(resp for _, _, resp in http_responses))
        except RequestException as exc:
            error = NetworkError(str(exc))
            if return_exceptions:
                for orig_idx, _, _ in http_responses:
                    results[orig_idx] = error
                return
            raise error from exc

        for orig_idx, req, resp in http_responses:
            try:
                results[orig_idx] = req._parse_response(resp)
            except Exception as exc:  # noqa: PERF203
                if return_exceptions:
                    results[orig_idx] = exc
                else:
                    raise

    async with anyio.create_task_group() as tg:
        for protocol, tasks in all_task.items():
            if protocol == CgiRequest._protocol:
                tasks = cast("list[tuple[int, CgiRequest]]", tasks)
                tg.start_soon(_gather_cgi, tasks)
            elif protocol == HttpRequest._protocol:
                tasks = cast("list[tuple[int, HttpRequest]]", tasks)
                tg.start_soon(_gather_http, tasks)

    missing = [i for i, res in enumerate(results) if res is MISSING]
    if missing:
        raise ApiDataError(f"缺少以下索引结果: {missing}")

    return results