跳转至

utils.session

请求 Session 管理

ApiConfig

Bases: TypedDict

API 配置

ORJsonSerializer

Bases: BaseSerializer

Transform data to json string with json.dumps and json.loads to retrieve it back.

dumps

dumps(value)

Serialize the received value using json.dumps.

Source code in qqmusic_api/utils/session.py
def dumps(self, value):
    """Serialize the received value using ``json.dumps``."""
    return json.dumps(value, option=json.OPT_NON_STR_KEYS).decode()

loads

loads(value)

Deserialize value using json.loads.

Source code in qqmusic_api/utils/session.py
def loads(self, value):
    """Deserialize value using ``json.loads``."""
    if value is None:
        return None
    return json.loads(value)

Session

Session(*, credential: Credential | None = None, enable_sign: bool = False, enable_cache: bool = True, cache_ttl: int = 120, http2: bool = True, **kwargs)

Bases: AsyncClient

Session 类,用于管理 QQ 音乐的登录态和 API 请求

PARAMETER DESCRIPTION
credential

全局凭证,每个请求都将使用.

TYPE: Credential | None DEFAULT: None

enable_sign

是否启用加密接口

TYPE: bool DEFAULT: False

enable_cache

是否启用请求缓存

TYPE: bool DEFAULT: True

cache_ttl

缓存过期时间

TYPE: int DEFAULT: 120

Source code in qqmusic_api/utils/session.py
def __init__(
    self,
    *,
    credential: Credential | None = None,
    enable_sign: bool = False,
    enable_cache: bool = True,
    cache_ttl: int = 120,
    http2: bool = True,
    **kwargs,
) -> None:
    super().__init__(**kwargs, http2=http2)
    self.credential = credential
    self.headers.update(
        {
            "User-Agent": self.UA_DEFAULT,
            "Referer": self.HOST,
        }
    )
    self.api_config = ApiConfig(
        version="13.2.5.8",
        version_code=13020508,
        enable_sign=enable_sign,
        endpoint="https://u.y.qq.com/cgi-bin/musicu.fcg",
        enc_endpoint="https://u.y.qq.com/cgi-bin/musics.fcg",
    )
    self.enable_cache = enable_cache
    self._cache = Cache(serializer=ORJsonSerializer(), ttl=cache_ttl)
    self.qimei = get_qimei(self.api_config["version"])["q36"]

clear_cache async

clear_cache()

清除API请求缓存

Source code in qqmusic_api/utils/session.py
async def clear_cache(self):
    """清除API请求缓存"""
    if not self.enable_cache:
        return
    await self._cache.clear()

get_session

get_session() -> Session

获取当前上下文的 Session

Source code in qqmusic_api/utils/session.py
def get_session() -> Session:
    """获取当前上下文的 Session"""
    session = _session_context.get()
    if session is None:
        logger.info("创建新的默认Session")
        session = Session()
        _session_context.set(session)
    return session

set_session

set_session(session: Session) -> None

设置当前上下文的 Session

Source code in qqmusic_api/utils/session.py
def set_session(session: Session) -> None:
    """设置当前上下文的 Session"""
    logger.info("设置新的Session到上下文")
    _session_context.set(session)

clear_session

clear_session() -> None

清除当前上下文的 Session

Source code in qqmusic_api/utils/session.py
def clear_session() -> None:
    """清除当前上下文的 Session"""
    logger.info("清除当前上下文的Session")
    try:
        _session_context.set(None)
    except LookupError:
        logger.warning("尝试清除不存在的Session上下文")
        pass