使用aiohttp构建异步API包装器时的None类型请求参数问题

问题描述 #

使用aiohttp构建异步API包装器时，可选参数默认值设为None会导致请求报错——aiohttp不允许None类型的请求参数（而requests会自动忽略这类参数）。现有代码如下：

import asyncio
import aiohttp

class AsyncWrapper:
    def __init__(self, api_key, useragent="API-Wrapper/0.2"):
        self.url = f"https://example.api/v1"
        self._api_key = api_key
        self._useragent = useragent
        self._headers = {"X-API-Key": self._api_key, "accept": "application/json", "User-Agent": self._useragent}
        self._session = aiohttp.ClientSession()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

    async def close(self):
        await self._session.close()

    async def get_endpoint1(self, arg1, arg2):
        endpoint = "/pair"

        async with self._session.get(self.url + endpoint, params={"arg1": arg1, "arg2": arg2}, headers=self._headers) as response:
            return await response.json()
    
    async def get_endpoint2(self, arg1, arg2, optionalarg1=None, optionalarg2=None):
        endpoint = "/token"

        async with self._session.get(self.url + endpoint, params={"arg1": arg1, "arg2": arg2, "optionalarg1": optionalarg1, "optionalarg2": optionalarg2}, headers=self._headers) as response:
            return await response.json()

调用get_endpoint2时会抛出错误：

TypeError: Invalid variable type: value should be str, int or float, got None of type <class 'NoneType'>

解决方案 #

1. 字典推导式过滤None值 #

直接在构建params字典时过滤掉值为None的键值对，这是最直接的实现方式：

async def get_endpoint2(self, arg1, arg2, optionalarg1=None, optionalarg2=None):
    endpoint = "/token"
    params = {
        "arg1": arg1,
        "arg2": arg2,
        "optionalarg1": optionalarg1,
        "optionalarg2": optionalarg2
    }
    # 过滤None值参数
    filtered_params = {k: v for k, v in params.items() if v is not None}
    
    async with self._session.get(self.url + endpoint, params=filtered_params, headers=self._headers) as response:
        return await response.json()

2. 封装通用参数过滤函数 #

如果多个接口都需要处理可选参数，封装工具函数可以避免重复代码：

def filter_none_params(params_dict):
    return {k: v for k, v in params_dict.items() if v is not None}

# 在类中调用工具函数
async def get_endpoint2(self, arg1, arg2, optionalarg1=None, optionalarg2=None):
    endpoint = "/token"
    params = filter_none_params({
        "arg1": arg1,
        "arg2": arg2,
        "optionalarg1": optionalarg1,
        "optionalarg2": optionalarg2
    })
    
    async with self._session.get(self.url + endpoint, params=params, headers=self._headers) as response:
        return await response.json()

3. 关键字参数动态构建params #

利用Python**语法，将可选参数作为关键字参数传入，动态构建不含None的字典：

async def get_endpoint2(self, arg1, arg2, **kwargs):
    endpoint = "/token"
    params = {"arg1": arg1, "arg2": arg2}
    # 仅保留kwargs中值不为None的项
    params.update({k: v for k, v in kwargs.items() if v is not None})
    
    async with self._session.get(self.url + endpoint, params=params, headers=self._headers) as response:
        return await response.json()

# 调用方式不变：await wrapper.get_endpoint2("val1", "val2", optionalarg1="val3")

现有代码优化建议 #

1. 安全拼接URL：直接用+拼接URL可能出现格式错误（如原url末尾带/时生成//pair），建议用urllib.parse.urljoin：

from urllib.parse import urljoin

# 替换接口内的URL拼接逻辑
full_url = urljoin(self.url, endpoint)

2. 延迟创建ClientSession：当前在__init__中创建session，若实例化后长时间未使用可能失效，建议在__aenter__中创建：

def __init__(self, api_key, useragent="API-Wrapper/0.2"):
    self.url = "https://example.api/v1"
    self._api_key = api_key
    self._useragent = useragent
    self._headers = {"X-API-Key": self._api_key, "accept": "application/json", "User-Agent": self._useragent}
    self._session = None

async def __aenter__(self):
    self._session = aiohttp.ClientSession()
    return self

3. 添加响应状态检查：请求成功后先校验状态码，避免解析错误响应内容：

async with self._session.get(full_url, params=filtered_params, headers=self._headers) as response:
    response.raise_for_status()  # 抛出4xx/5xx类HTTP错误
    return await response.json()

4. 添加类型提示：提升代码可读性与可维护性：

from typing import Optional, Dict, Any

class AsyncWrapper:
    def __init__(self, api_key: str, useragent: str = "API-Wrapper/0.2") -> None:
        # ...
    async def get_endpoint2(self, arg1: str, arg2: str, optionalarg1: Optional[str] = None, optionalarg2: Optional[int] = None) -> Dict[str, Any]:
        # ...

内容的提问来源于stack exchange，提问作者alb2001