aiocsv

异步CSV读取和写入。

安装

pip install aiocsv. 需要 Python 3.8+。

此模块包含用C编写的扩展。可能不存在适用于您配置的预构建二进制文件。您可能需要C编译器和Python头文件来安装aiocsv。

使用方法

AsyncReader & AsyncDictReader接受任何具有read(size: int)协程的对象，该协程应返回一个字符串。

AsyncWriter & AsyncDictWriter接受任何具有write(b: str)协程的对象。

读取是通过自定义CSV解析器实现的，其行为应与CPython解析器完全相同。

写入是通过同步的csv.writer和csv.DictWriter对象实现的 - 序列化器将数据写入StringIO，然后该缓冲区被重写到底层异步文件。

示例

使用aiofiles的示例使用方法。

import asyncio
import csv

import aiofiles
from aiocsv import AsyncReader, AsyncDictReader, AsyncWriter, AsyncDictWriter

async def main():
    # simple reading
    async with aiofiles.open("some_file.csv", mode="r", encoding="utf-8", newline="") as afp:
        async for row in AsyncReader(afp):
            print(row)  # row is a list

    # dict reading, tab-separated
    async with aiofiles.open("some_other_file.tsv", mode="r", encoding="utf-8", newline="") as afp:
        async for row in AsyncDictReader(afp, delimiter="\t"):
            print(row)  # row is a dict

    # simple writing, "unix"-dialect
    async with aiofiles.open("new_file.csv", mode="w", encoding="utf-8", newline="") as afp:
        writer = AsyncWriter(afp, dialect="unix")
        await writer.writerow(["name", "age"])
        await writer.writerows([
            ["John", 26], ["Sasha", 42], ["Hana", 37]
        ])

    # dict writing, all quoted, "NULL" for missing fields
    async with aiofiles.open("new_file2.csv", mode="w", encoding="utf-8", newline="") as afp:
        writer = AsyncDictWriter(afp, ["name", "age"], restval="NULL", quoting=csv.QUOTE_ALL)
        await writer.writeheader()
        await writer.writerow({"name": "John", "age": 26})
        await writer.writerows([
            {"name": "Sasha", "age": 42},
            {"name": "Hana"}
        ])

asyncio.run(main())

与csv的区别

aiocsv力求成为Python内置的csv模块的替代品。但是，存在3个明显的区别

• 读取器接受具有异步read方法的对象，而不是文件行上的AsyncIterable。
• AsyncDictReader.fieldnames可以是None - 使用await AsyncDictReader.get_fieldnames()代替。
• csv.field_size_limit 的更改不会被现有的 Reader 实例所识别。字段大小限制在 Reader 实例化时缓存，以避免在输入的每个字符上调用昂贵的函数。

其他一些细微的差异包括

• AsyncReader.line_num、AsyncDictReader.line_num 和 AsyncDictReader.dialect 不可设置，
• AsyncDictReader.reader 是 AsyncReader 类型，
• AsyncDictWriter.writer 是 AsyncWriter 类型，
• AsyncDictWriter 提供了一个额外的只读 dialect 属性。

参考

aiocsv.AsyncReader

AsyncReader(
    asyncfile: aiocsv.protocols.WithAsyncRead,
    dialect: str | csv.Dialect | Type[csv.Dialect] = "excel",
    **csv_dialect_kwargs: Unpack[aiocsv.protocols.CsvDialectKwargs],
)

一个对象，它遍历给定异步 CSV 文件中的记录。额外的关键字参数被视为方言参数。

遍历此对象返回解析后的 CSV 行 (List[str])。

方法:

• __aiter__(self) -> self
• async __anext__(self) -> List[str]

只读属性:

• dialect：解析时使用的 csv.Dialect
• line_num：从源文件中读取的行数。这与最近解析的记录的最后一行的 1 为基索引的行号相吻合。

aiocsv.AsyncDictReader

AsyncDictReader(
    asyncfile: aiocsv.protocols.WithAsyncRead,
    fieldnames: Optional[Sequence[str]] = None,
    restkey: Optional[str] = None,
    restval: Optional[str] = None,
    dialect: str | csv.Dialect | Type[csv.Dialect] = "excel",
    **csv_dialect_kwargs: Unpack[aiocsv.protocols.CsvDialectKwargs],
)

一个对象，它遍历给定异步 CSV 文件中的记录。所有参数与 csv.DictReader 完全相同。

遍历此对象返回解析后的 CSV 行 (Dict[str, str])。

方法:

• __aiter__(self) -> self
• async __anext__(self) -> Dict[str, str]
• async get_fieldnames(self) -> List[str]

属性:

• fieldnames：在将行转换为字典时使用的字段名
  ⚠️ 与 csv.DictReader 不同，此属性无法读取缺失的字段名 - 无法在属性获取器的标题行上 await。 请使用 await reader.get_fieldnames()。

  reader = csv.DictReader(some_file)
  reader.fieldnames  # ["cells", "from", "the", "header"]
  
  areader = aiofiles.AsyncDictReader(same_file_but_async)
  areader.fieldnames   # ⚠️ None
  await areader.get_fieldnames()  # ["cells", "from", "the", "header"]
  

• restkey：如果一行比标题多单元格，所有剩余的单元格都将存储在此键下返回的字典中。默认为 None。
• restval：如果一行比标题少单元格，则缺失的键将使用此值。默认为 None。
• reader：底层的 aiofiles.AsyncReader 实例

只读属性:

• dialect：链接到 self.reader.dialect - 当前 csv.Dialect
• line_num：从源文件中读取的行数。这与最近解析的记录的最后一行的 1 为基索引的行号相吻合。

aiocsv.AsyncWriter

AsyncWriter(
    asyncfile: aiocsv.protocols.WithAsyncWrite,
    dialect: str | csv.Dialect | Type[csv.Dialect] = "excel",
    **csv_dialect_kwargs: Unpack[aiocsv.protocols.CsvDialectKwargs],
)

一个对象，它将 CSV 行写入给定的异步文件。在此对象中，“行”是值的序列。

额外的关键字参数传递到底层的 csv.writer 实例。

方法:

• async writerow(self, row: Iterable[Any]) -> None：将一行写入指定的文件。
• async writerows(self, rows: Iterable[Iterable[Any]]) -> None：将多行写入指定的文件。

只读属性:

• dialect：链接到底层的 csv.writer 的 dialect 属性

aiocsv.AsyncDictWriter

AsyncDictWriter(
    asyncfile: aiocsv.protocols.WithAsyncWrite,
    fieldnames: Sequence[str],
    restval: Any = "",
    extrasaction: Literal["raise", "ignore"] = "raise",
    dialect: str | csv.Dialect | Type[csv.Dialect] = "excel",
    **csv_dialect_kwargs: Unpack[aiocsv.protocols.CsvDialectKwargs],
)

一个对象，它将 CSV 行写入给定的异步文件。在此对象中，“行”是从字段名到值的映射。

额外的关键字参数传递到底层的 csv.DictWriter 实例。

方法:

• async writeheader(self) -> None：将标题行写入指定的文件。
• async writerow(self, row: Mapping[str, Any]) -> None：将一行写入指定的文件。
• async writerows(self, rows: Iterable[Mapping[str, Any]]) -> None：将多行写入指定的文件。

属性:

• fieldnames：序列，用于标识在将行写入底层文件时值的顺序
• restval：当行中的键不在字段名中时使用的占位符值，默认为 ""
• extrasaction：当一行中存在不在字段名中的键时采取的操作，默认为 "raise"，它会在额外键上引发 ValueError，也可以设置为 "ignore" 来忽略任何额外键
• writer：链接到底层的 AsyncWriter

只读属性:

• dialect：链接到底层的 csv.reader 的 dialect 属性

aiocsv.protocols.WithAsyncRead

一个 typing.Protocol，描述了一个可以读取的异步文件。

aiocsv.protocols.WithAsyncWrite

描述可异步写入的文件的 typing.Protocol。

aiocsv.protocols.CsvDialectArg

dialect 参数的类型，用于 csv 模块。

aiocsv.protocols.CsvDialectKwargs

csv 模块在读取器/写入器实例化期间用于覆盖方言设置的键值对参数。

开发

欢迎贡献，但请在之前提交问题。 aiocsv 被设计为内置 csv 的替代品，任何后者没有的功能都将被拒绝。

从源代码构建

要创建一个 wheel（以及源代码压缩包），运行 python -m build。

对于本地开发，请使用 虚拟环境。 pip install --editable . 将构建 C 扩展并将其提供给当前的 venv。这是运行测试所必需的。然而，由于 Python 打包混乱，这将强制进行优化构建，而不包含调试符号。如果您需要调试 aiocsv 的 C 部分，并使用例如调试符号构建库，唯一合理的做法是运行 python setup.py build --debug，然后手动将共享对象/DLL 从 build/lib*/aiocsv 复制到 aiocsv。

测试

该项目使用 pytest 和 pytest-asyncio 进行测试。在以上述方式安装库后运行 pytest。

代码风格检查与其他工具

该库使用 black 和 isort 进行格式化，并在严格模式下使用 pyright 进行类型检查。

对于库的 C 部分，请使用 clang-format 进行格式化和 clang-tidy 进行代码风格检查，然而这些工具尚未集成到 CI 中。

安装所需的工具

pip install -r requirements.dev.txt 将安装上述所有开发工具，但这可能不是必需的，具体取决于您的设置。例如，如果您使用带有 Python 扩展的 VS Code，pyright 已经捆绑在内，不需要再次安装。

推荐的 VS Code 设置

使用 Python、Pylance（应与 Python 扩展自动安装），以及 black 和 isort Python 扩展。

您需要从 requirements.dev.txt 安装所有开发依赖项，除了 pyright。推荐的 .vscode/settings.json

{
    "C_Cpp.codeAnalysis.clangTidy.enabled": true,
    "python.testing.pytestArgs": [
        "."
    ],
    "python.testing.unittestEnabled": false,
    "python.testing.pytestEnabled": true,
    "[python]": {
        "editor.formatOnSave": true,
        "editor.codeActionsOnSave": {
            "source.organizeImports": "always"
        }
    },
    "[c]": {
        "editor.formatOnSave": true
    }
}

对于库的 C 部分，C/C++ 扩展 就足够了。请确保您的系统已安装 Python 标头。通常需要安装单独的包，例如 python3-dev，有关此信息，请咨询您的系统仓库。需要在 .vscode/c_cpp_properties.json 中手动包含 Python 标头在 includePath 下。在我的特定系统中，此配置文件如下所示

{
    "configurations": [
        {
            "name": "Linux",
            "includePath": [
                "${workspaceFolder}/**",
                "/usr/include/python3.11"
            ],
            "defines": [],
            "compilerPath": "/usr/bin/clang",
            "cStandard": "c17",
            "cppStandard": "c++17",
            "intelliSenseMode": "linux-clang-x64"
        }
    ],
    "version": 4
}