casanova

如果您经常使用Python处理CSV文件，您很快就会注意到，虽然使用起来更舒适，但csv.DictReader的速度仍然远慢于csv.reader

# To read a 1.5G CSV file:
csv.reader: 24s
csv.DictReader: 84s
casanova.reader: 25s

因此，casanova试图在保持舒适的用户界面同时，仍然保持与csv.reader相同的性能，并且仍然能够考虑标题（甚至重复的标题，这是csv.DictReader无法做到的）等。

如果您需要以下功能，casanova非常适合您：

• 流式传输大型CSV文件而不会耗尽内存
• 通过输出类似文件来丰富相同的CSV文件，同时添加、过滤和编辑单元格。
• 如果您的进程退出，有机会恢复上述丰富操作
• 即使您的输出顺序与输入不同，也可以继续操作

casanova 还打包了能够以反向方式读取 csv 文件（无需将整个文件加载到内存中，且在常规 O(n) 时间内完成）的异国工具，因此您可以，例如，检索文件末尾的有用信息以重新启动某些中断的过程。

最后，casanova 可以作为一个命令行工具，能够评估给定 CSV 文件的每一行的 python 表达式（如果需要，可以并行化），以产生典型结果，如基于其他列添加列等。

命令行工具的文档可以在这里找到。

对于不需要 python 评估的更通用任务，我们建议使用性能非常出色的 xsv 工具，或者我们自己的 分支 工具。

安装

您可以使用以下命令使用 pip 安装 casanova

pip install casanova

如果您想能够从网络将 CSV 文件喂给 casanova 读取器 & 富化器，您还需要安装至少 urllib3，以及可选的 certifi（如果您想要安全的 SSL）。请注意，许多 python 包，包括流行的 requests 库，已经依赖于这两个库，所以您可能已经安装了它们。

# Installing them explicitly
pip install urllib3 certifi

# Installing casanova with those implicitly
pip install casanova[http]

用法

• 读取器
• 反向读取器
• 标题
• 写入器
• 富化器
• 索引富化器
• 批量富化器
• 恢复器
  • 行数恢复器
  • 索引恢复器
  • 批量恢复器
  • 最后一行恢复器
  • 最后一行比较恢复器
• 计数
• 最后一行
• 设置默认值
• xsv 选择迷你 DSL

读取器

简单的 CSV 读取器，产生列表行，但提供有关潜在标题及其位置的一些信息。

import casanova

with open('./people.csv') as f:

    # Creating a reader
    reader = casanova.reader(f)

    # Getting header information
    reader.fieldnames
    >>> ['name', 'surname']

    reader.headers
    >>> Headers(name=0, surname=1)

    name_pos = reader.headers.name
    name_pos = reader.headers['name']

    'name' in reader.headers
    >>> True

    # Iterating over the rows
    for row in reader:
        name = row[name_pos] # it's better to cache your pos outside the loop
        name = row[reader.headers.name] # this works, but is slower

    # Interested in a single column?
    for name in reader.cells('name'):
        print(name)

    # Need also the current row when iterating on cells?
    for row, name in reader.cells('name', with_rows=True):
        print(row, name, surname)

    # Want to iterate over records
    # NOTE: this has a performance cost
    for name, surname in reader.records('name', 'surname'):
        print(name, surname)

    for record in reader.records(['name', 'age']):
        print(record[0])

    for record in reader.records({'name': 'name', 'age': 1}):
        print(record['age'])

    # No headers? No problem.
    reader = casanova.reader(f, no_headers=True)

# Note that you can also create a reader from a path
with casanova.reader('./people.csv') as reader:
  ...

# And if you need exotic encodings
with casanova.reader('./people.csv', encoding='latin1') as reader:
  ...

# The reader will also handle gzipped files out of the box
with casanova.reader('./people.csv.gz') as reader:
  ...

# If you have `urllib3` installed, casanova is also able to stream
# remote CSV file out of the box
with casanova.reader('https://mydomain.fr/some-file.csv') as reader:
    ...

# The reader will also accept iterables of rows
rows = [['name', 'surname'], ['John', 'Moran']]
reader = casanova.reader(rows)

# And you can of course use the typical dialect-related kwargs
reader = casanova.reader('./french-semicolons.csv', delimiter=';')

# Readers can also be closed if you want to avoid context managers
reader.close()

参数

• 输入文件 str 或 Path 或文件或 list[str] 的迭代器：提供给读取器的输入文件。可以是读取器为您打开的路径，文件句柄，甚至是一个任意列表行的迭代器。
• 无标题 bool，可选 [False]：如果 输入文件 没有标题，则设置为 True。
• 编码 str，可选 [utf-8]：如果 输入文件 是路径，则使用的编码。
• 方言 str 或 csv.Dialect，可选：读取器要使用的 CSV 方言。有关更多信息，请参阅 python 标准的 csv 模块文档。
• 引号字符 str，可选：CSV 解析器使用的引号字符。
• 分隔符 str，可选：CSV 解析器使用的分隔符字符。
• 预缓冲字节 int，可选：在尝试提前获取总行数之前，预缓冲输入文件中的字节数。
• 总数 int，可选：如果提前知道文件中的总行数，则预期的总行数。如果给定，则读取器不会预先缓冲数据，即使设置了 预缓冲字节。
• 多路复用 casanova.Multiplexer，可选：要使用的多路复用器。有关更多信息，请参阅此。
• 在读取时删除空字节 bool，可选 [False]：在 python 3.11 之前，当尝试读取包含空字节的 CSV 文件时，csv 模块将引发错误。如果设置为 True，则读取器将在解析行时实时删除空字节。
• 反向 bool，可选 [False]：是否以反向方式读取文件（当然不包括标题）。

属性

• 总数 int，可选：如果通过预缓冲或通过 总数 关键字参数知道，则文件中的总行数。
• 标题 casanova.Headers，可选，可选：如果 无标题=False，则 CSV 文件标题。
• empty bool：给定的文件是否为空。
• fieldnames list[str]，可选：表示 CSV 文件标题的列表，当 no_headers=False 时使用。
• row_len int：每行期望的项目数。

方法

• rows：返回一个遍历读取器行的迭代器。与直接遍历读取器相同。
• cells：接受列名或其位置，并返回给定列值的迭代器。如果需要迭代 value, row 元组，则可以提供 with_rows=True。
• enumerate：提供安全的枚举行，产生 index, row 元组。接受与内置 enumerate 相同的可选 start 参数。
• enumerate_cells：提供安全的枚举单元格，产生 index, cell 或如果给定 with_rows=True，则产生 index, row, cell。接受与内置 enumerate 相同的可选 start 参数。
• wrap：接受一个列表行并返回一个用于包装的 RowWrapper 对象的方法。
• close：在未使用专用上下文管理器时手动清理读取器资源。通常仅在读取器被给定路径而不是已打开的文件句柄时才有用。

多路复用

有时，CSV 文件中的一列可能包含多个值，这些值由一个任意的分隔符字符（如 |）分隔。

在这种情况下，可能希望通过使读取器为单元格中包含的每个值发出一行副本来“多路复用”文件。

为此，casanova 公开了一个特殊的 Multiplexer 对象，您可以将其提供给任何读取器，如下所示：

import casanova

# Most simple case: a column named "colors", separated by "|"
reader = casanova.reader(
    input_file,
    multiplex=casanova.Multiplexer('colors')
)

# Customizing the separator:
reader = casanova.reader(
    input_file,
    multiplex=casanova.Multiplexer('colors', separator='$')
)

# Renaming the column on the fly:
reader = casanova.reader(
    input_file,
    multiplex=casanova.Multiplexer('colors', new_column='color')
)

反向读取器

反向 CSV 读取器听起来可能很傻，但在某些情况下可能很有用。特别是当您需要在不读取整个文件的情况下以恒定时间读取输出文件的最后一行时。

它主要用于 casanova resumers，您不太可能需要自己使用它们。

import casanova

# people.csv looks like this
# name,surname
# John,Doe,
# Mary,Albert
# Quentin,Gold

with open('./people.csv', 'rb') as f:
    reader = casanova.reverse_reader(f)

    reader.fieldnames
    >>> ['name', 'surname']

    next(reader)
    >>> ['Quentin', 'Gold']

标题

表示 CSV 文件标题的类。用于查找某些列的行位置并进行复杂选择很有用。

import casanova

# Headers can be instantiated thusly
headers = casanova.headers(['name', 'surname', 'age'])

# But you will usually use a reader or an enricher's one:
headers = casanova.reader(input_file).headers

# Accessing a column through attributes
headers.surname
>>> 1

# Accessing a column by indexing:
headers['surname']
>>> 1

# Getting a column
headers.get('surname')
>>> 1
headers.get('not-found')
>>> None

# Getting a duplicated column name
casanova.headers(['surname', 'name', 'name'])['name', 1]
>>> 2

casanova.headers(['surname', 'name', 'name']).get('name', index=1)
>>> 2

# Asking if a column exists:
'name' in headers:
>>> True

# Retrieving fieldnames:
headers.fieldnames
>>> ['name', 'surname', 'age']

# Iterating over headers
for col in headers:
    print(col)

# Couting columns:
len(headers)
>>> 3

# Retrieving the nth header:
headers.nth(1)
>>> 'surname'

# Wraping a row
headers.wrap(['John', 'Matthews', '45'])
>>> RowWrapper(name='John', surname='Matthews', age='45')

# Selecting some columns (by name and/or index)):
headers.select(['name', 2])
>>> [0, 2]

# Selecting using xsv mini DSL:
headers.select('name,age')
>>> [0, 2]
headers.select('!name')
>>> [1, 2]

有关 xsv mini DSL 的更多信息，请参阅文档中的此部分：这里。

写入器

casanova 还导出 CSV 写入器。它可以在需要时自动写入标题，并且能够恢复一些任务。

import casanova

with open('output.csv') as f:
    writer = casanova.writer(f, fieldnames=['name', 'surname'])

    writer.writerow(['John', 'Davis'])

    # If you want to write headers yourself:
    writer = casanova.writer(f, fieldnames=['name', 'surname'], write_header=False)

    writer.writeheader()

参数

• output_file file 或 casanova.Resumer：目标文件。
• fieldnames Iterable[str]，可选：列名。
• strip_null_bytes_on_write bool，可选 [False]：在写入行时是否删除空字节。注意，在 python 3.10 中，有一个错误会阻止 csv.writer 在尝试写入包含空字节的行时引发错误。
• dialect csv.Dialect 或 str，可选：用于写入 CSV 的方言。
• delimiter str，可选：CSV 分隔符。
• quotechar str，可选：CSV 引用字符。
• quoting csv.QUOTE_*，可选：CSV 引用策略。
• escapechar str，可选：CSV 转义字符。
• lineterminator str，可选：CSV 行终止符。
• write_header bool，可选 [True]：是否在需要时自动写入标题（考虑恢复）。

属性

• headers casanova.Headers，可选，可选：如果提供了字段名，则为 CSV 文件标题。
• fieldnames list[str]，可选：提供的字段名。

恢复

casanova.writer 可以通过 LastCellResumer 进行恢复。

富化器

casanova enrichers 是读取器和写入器的智能组合。

它可以用来转换给定的CSV文件。这意味着您可以在运行时动态转换其值，轻松选择要保留的列，并添加新列。

请注意， enrichers 继承自 casanova.reader 和 casanova.writer，因此保留了它们的属性和方法。

import casanova

with open('./people.csv') as input_file, \
     open('./enriched-people.csv', 'w') as output_file:

    enricher = casanova.enricher(input_file, output_file)

    # The enricher inherits from casanova.reader
    enricher.fieldnames
    >>> ['name', 'surname']

    # You can iterate over its rows
    name_pos = enricher.headers.name

    for row in enricher:

        # Editing a cell, so that everyone is called John now
        row[name_pos] = 'John'
        enricher.writerow(row)

    # Want to add columns?
    enricher = casanova.enricher(f, of, add=['age', 'hair'])

    for row in enricher:
        enricher.writerow(row, ['34', 'blond'])

    # Want to keep only some columns from input?
    enricher = casanova.enricher(f, of, add=['age'], select=['surname'])

    for row in enricher:
        enricher.writerow(row, ['45'])

    # Want to select columns to keep using xsv mini dsl?
    enricher = casanova.enricher(f, of, select='!1-4')

    # You can of course still use #.cells etc.
    for row, name in enricher.cells('name', with_rows=True):
        print(row, name)

参数

• input_file 文件或str：读取的文件对象或打开的路径。
• output_file 文件或Resumer：写入的文件对象。
• 无标题 bool，可选 [False]：如果 输入文件 没有标题，则设置为 True。
• 编码 str，可选 [utf-8]：如果 输入文件 是路径，则使用的编码。
• add Iterable[str|int]，可选：要添加到输出的列名。
• select Iterable[str|int]|str，可选：从输入中保留的列的选择。可以是列名和/或列位置的集合，或使用 xsv mini DSL 编写的选择字符串。
• 方言 str 或 csv.Dialect，可选：读取器要使用的 CSV 方言。有关更多信息，请参阅 python 标准的 csv 模块文档。
• 引号字符 str，可选：CSV 解析器使用的引号字符。
• 分隔符 str，可选：CSV 解析器使用的分隔符字符。
• 预缓冲字节 int，可选：在尝试提前获取总行数之前，预缓冲输入文件中的字节数。
• 总数 int，可选：如果提前知道文件中的总行数，则预期的总行数。如果给定，则读取器不会预先缓冲数据，即使设置了 预缓冲字节。
• 多路复用 casanova.Multiplexer，可选：要使用的多路复用器。有关更多信息，请参阅此。
• 反向 bool，可选 [False]：是否以反向方式读取文件（当然不包括标题）。
• 在读取时删除空字节 bool，可选 [False]：在 python 3.11 之前，当尝试读取包含空字节的 CSV 文件时，csv 模块将引发错误。如果设置为 True，则读取器将在解析行时实时删除空字节。
• strip_null_bytes_on_write bool，可选 [False]：在写入行时是否删除空字节。注意，在 python 3.10 中，有一个错误会阻止 csv.writer 在尝试写入包含空字节的行时引发错误。
• writer_dialect csv.Dialect 或 str，可选：用于写入CSV的方言。
• writer_delimiter str，可选：写入者的CSV分隔符。
• writer_quotechar str，可选：写入者的CSV引号字符。
• writer_quoting csv.QUOTE_*，可选：写入者的CSV引号策略。
• writer_escapechar str，可选：写入者的CSV转义字符。
• writer_lineterminator str，可选：写入者的CSV行终止符。
• write_header bool，可选 [True]：是否在需要时自动写入标题（考虑恢复）。

属性

• 总数 int，可选：如果通过预缓冲或通过 总数 关键字参数知道，则文件中的总行数。
• 标题 casanova.Headers，可选，可选：如果 无标题=False，则 CSV 文件标题。
• empty bool：给定的文件是否为空。
• fieldnames list[str]，可选：表示 CSV 文件标题的列表，当 no_headers=False 时使用。
• row_len int：每行期望的项目数。
• output_headers casanova.Headers，可选，可选：如果 no_headers=False，则输出CSV头。
• output_fieldnames list[str]，可选：表示如果 no_headers=False，则输出CSV头的列表。
• added_count int：添加到输出的列数。

恢复

casanova.enricher 能够通过 RowCountResumer 或 LastCellComparisonResumer 来恢复。

索引富化器

有时，您可能希望同时处理多个输入行。这可能意味着您将按任意顺序发出行，这与输入顺序不同。

当然，这是可以的，但如果您仍然希望能够有效地恢复中断的进程（使用 `IndexedResumer），则您的输出将需要特定的添加才能使其工作，具体来说，需要包含原始输入中输出行索引的列。

casanova.indexed_enricher 通过提供定制的 writerow 方法，并始终提供行索引来简化这一点。

请注意，只有当输入中的一行在输出中产生一行时，这种恢复才可行。

import casanova

with open('./people.csv') as f, \
     open('./enriched-people.csv', 'w') as of:

    enricher = casanova.indexed_enricher(f, of, add=['age', 'hair'])

    for index, row in enricher:
        enricher.writerow(index, row, ['67', 'blond'])

    for index, value in enricher.cells('name'):
        ...

    for index, row, value in enricher.cells('name', with_rows=True):
        ...

参数

casanova.enricher 的一切加上

• index_column str，可选 [index]：自动添加的索引列的名称。

恢复

casanova.indexed_enricher 能够通过 IndexedResumer 来恢复。

批量富化器

有时，您可能想要处理CSV文件，并按行分页API调用。这意味着您的输入文件的每一行都应该产生多行新内容，这些内容将在每次API调用返回时批量写入。

有时，分页可能相当长（例如，收集非常受欢迎的账户的Twitter关注者），在将所有结果累积到单个行原子性地刷新到文件之前，这不是一个好的主意，因为如果出现问题，您将丢失大量工作。

但如果您仍然希望能够在中断时恢复处理，则需要将某些内容添加到输出中。具体来说，需要包含用于恢复API调用的可选“游标”数据的列以及表示我们已完成当前输入行处理的“结束”符号。

import casanova

with open('./twitter-users.csv') as input_file, \
     casanova.BatchResumer('./output.csv') as output_file:

    enricher = casanova.batch_resumer(input_file, output_file)

    for row in enricher:
        for results, next_cursor in paginate_api_calls(row):

            # NOTE: if we reached the end, next_cursor is None
            enricher.writebatch(row, results, next_cursor)

参数

casanova.enricher 的一切加上

• cursor_column str，可选 [cursor]：要添加的游标列的名称。
• end_symbol str，可选 [end]：明确的（从游标）结束符号，用于标记输入行处理的结束。

恢复

casanova.batch_enricher 能够通过 BatchResumer 来恢复。

恢复器

通过便捷的 Resumer 类，casanova 允许其增强器和作者在进程中断后恢复操作。

这些类必须用作包装器以打开输出文件，并可以评估恢复对您是否有用。

所有恢复器都像文件句柄一样工作，可以用 with 关键字用作上下文管理器，如果需要，可以使用 close 方法手动关闭。

最后，了解恢复器应该与复用完美协同工作。

行数恢复器

RowCountResumer 通过计算输出中的行数并从输入中跳过这么多行来工作。

它只能在 1-to-1 场景中工作，其中您仅对每行输入输出单行。

它在 O(2n) => O(n) 时间内工作，并在 O(1) 内存中工作，其中 n 是已处理的行数。

它仅由 casanova.enricher 支持。

import casanova

with open('input.csv') as input_file, \
     casanova.RowCountResumer('output.csv') as resumer:

    # Want to know if we can resume?
    resumer.can_resume()

    # Want to know how many rows were already done?
    resumer.already_done_count()

    # Giving the resumer to an enricher as if it was the output file
    enricher = casanova.enricher(input_file, resumer)

索引恢复器

casanova 导出一个索引恢复器，允许行并发处理并以不同的顺序输出。

在这种情况下，仅计数行是不够的，我们需要更聪明。

一种做法是利用索引增强器添加的索引列，在读取输出时计算已处理的行集。然后我们可以简单地跳过那些索引在此集合中的输入行。

问题在于这消耗了高达 O(n) 的内存，这在某些用例中是不可行的。

为了确保在消耗极小内存的情况下仍然可以这样做，casanova 使用了一种名为“连续范围集”的异构数据结构。

这意味着我们可以以 O(n + log(h) * n)) => O(log(h) * n) 的时间恢复操作，并在 O(log(h)) 的内存中工作，其中 n 是已处理的行数，而 h 是这些行排序索引中最大空洞的大小。请注意，由于输出大部分是排序的（尽管不是在局部层面），大多数情况下 h << n。

您可以在这篇博客文章中了解更多关于此数据结构的信息。

请注意，此恢复器只能在 1-to-1 场景中工作，其中您仅对每行输入输出单行。

它仅由 casanova.indexed_enricher 支持。

import casanova

with open('input.csv') as input_file, \
     casanova.IndexedResumer('output.csv') as resumer:

    # Want to know if we can resume?
    resumer.can_resume()

    # Want to know how many rows were already done?
    resumer.already_done_count()

    # Giving the resumer to an enricher as if it was the output file
    enricher = casanova.indexed_enricher(input_file, resumer)

# If you want to use casanova ContiguousRangeSet for whatever reason
from casanova import ContiguousRangeSet

批量恢复器

待办...

最后一行恢复器

有时您可能在执行分页操作时编写输出 CSV 文件。所述操作可能已中止，您可能希望从中断的地方继续。

因此，LastCellResumer 允许您通过使用 casanova.reverse_reader 读取输出的最后一行并提取您需要恢复的值以恒定时间和内存来恢复写入 CSV 文件。

它仅由 casanova.writer 支持。

import casanova

with casanova.LastCellResumer('output.csv', value_column='user_id') as resumer:

    # Giving the resumer to a writer as if it was the output file
    writer = casanova.writer(resumer)

    # Extracting last relevant value if any, so we can properly resume
    last_value = resumer.get_state()

最后一行比较恢复器

在某些场景中，如果您知道输出中某些列的最后一个值，则可以恢复增强器的操作。

幸运的是，使用 casanova.reverse_reader，可以在恒定时间内读取 CSV 文件的最后一行。

因此，LastCellComparisonResumer 允许您以 O(n) 的时间和 O(1) 的内存恢复增强器的操作，其中 n 是在重新定位输入时必须快速跳过的已完成的行数。

请注意，它仅在增强器每行输入输出一行且考虑的列值在整个输入文件中是唯一的情况下才工作。

它仅由 casanova.enricher 支持。

import casanova

with open('input.csv') as input_file, \
     casanova.LastCellComparisonResumer('output.csv', value_colum='user_id') as resumer:

    # Giving the resumer to an enricher as if it was the output file
    enricher = casanova.enricher(input_file, resumer)

计数

casanova 提供了一个辅助函数，可以快速统计 CSV 文件中的行数。

import casanova

count = casanova.count('./people.csv')

# You can also stop reading the file if you go beyond a number of rows
count = casanova.count('./people.csv', max_rows=100)
>>> None # if the file has more than 100 rows
>>> 34   # else the actual count

# Any additional kwarg will be passed to the underlying reader as-is
count = casanova.count('./people.csv', delimiter=';')

最后一行

casanova 使用一个 reverse_reader 提供了一个辅助函数，用于从 CSV 文件的给定列中仅读取最后一列的值。

import casanova

last_cell = casanova.last_cell('./people.csv', column='name')
>>> 'Quentin'

# Will return None if the file is empty
last_cell = casanova.last_cell('./empty.csv', column='name')
>>> None

# Any additional kwarg will be passed to the underlying reader as-is
last_cell = casanova.last_cell('./people.csv', column='name', delimiter=';')

设置默认值

casanova.set_defaults 允许您编辑 casanova 的全局默认设置。

import casanova

casanova.set_defaults(strip_null_bytes_on_read=True)

# As a context manager:
with casanova.temporary_defaults(strip_null_bytes_on_read=True):
    ...

参数

• strip_null_bytes_on_read bool, 可选 [False]: 读取时，读取器和富化器是否应该删除空字节？
• strip_null_bytes_on_write bool, 可选 [False]: 写入时，写入器和富化器是否应该删除空字节？
• prebuffer_bytes int, 可选：读取器和富化器的默认预缓冲字节。

xsv 选择迷你 DSL

xsv，一个用Rust编写的处理csv文件的命令行工具，使用一个巧妙的迷你DSL让用户指定列选择。

casanova实现了这个迷你DSL的Python版本，可以通过headers.select方法和富化器的select kwargs使用。

以下是它的要点（直接从xsv文档中复制）

Select one column by name:
    * name

Select one column by index (1-based):
    * 2

Select the first and fourth columns:
    * 1,4

Select the first 4 columns (by index and by name):
    * 1-4
    * Header1-Header4

Ignore the first 2 columns (by range and by omission):
    * 3-
    * '!1-2'

Select the third column named 'Foo':
    * 'Foo[2]'

Re-order and duplicate columns arbitrarily:
    * 3-1,Header3-Header1,Header1,Foo[2],Header1

Quote column names that conflict with selector syntax:
    * '"Date - Opening","Date - Actual Closing"'