Skip to content

turu.core

turu.core.record.record_to_csv(record_filepath, cursor, *, enable=True, **options) 

record_to_csv(record_filepath: Union[str, Path], cursor: turu.core.cursor.GenericCursor, *, enable: Union[str, bool, None] = True, **options: Unpack[CsvRecorderOptions]) -> _GeneratorContextManager[turu.core.cursor.GenericCursor]

record_to_csv(record_filepath: Union[str, Path], cursor: turu.core.async_cursor.GenericAsyncCursor, *, enable: Union[str, bool, None] = True, **options: Unpack[CsvRecorderOptions]) -> _AsyncGeneratorContextManager[turu.core.async_cursor.GenericAsyncCursor]

Records cursor's fetched data to CSV.

Parameters:

Name Type Description Default
record_filepath Union[str, Path]

Path to record file.

 required
cursor Union[GenericCursor, GenericAsyncCursor]

Cursor to record.

 required
enable Union[str, bool, None]

Enable recording.

 True
options Unpack[CsvRecorderOptions]

Options for CSV recorder.

 {}
Source code in turu-core/src/turu/core/record/__init__.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def record_to_csv(  # type: ignore
    record_filepath: Union[str, Path],
    cursor: Union[
        turu.core.cursor.GenericCursor, turu.core.async_cursor.GenericAsyncCursor
    ],
    *,
    enable: Union[str, bool, None] = True,
    **options: Unpack[CsvRecorderOptions],
):
    """Records cursor's fetched data to CSV.

    Parameters:
        record_filepath: Path to record file.
        cursor: Cursor to record.
        enable: Enable recording.
        options: Options for CSV recorder.
    """
    if isinstance(cursor, turu.core.cursor.Cursor):
        return _GeneratorContextManager(
            _record_to_csv, (record_filepath, cursor), dict(enable=enable, **options)
        )

    elif isinstance(cursor, turu.core.async_cursor.AsyncCursor):
        return _AsyncGeneratorContextManager(
            _record_to_csv_async,
            (record_filepath, cursor),
            dict(enable=enable, **options),
        )

    raise NotImplementedError(f"cursor type {type(cursor)} is not supported")

turu.core.record.CsvRecorderOptions

Options for CSV recorder.

Source code in turu-core/src/turu/core/record/csv_recorder.py 
43
44
45
46
47
48
49
50
class CsvRecorderOptions(TypedDict):
    """Options for CSV recorder."""

    header: NotRequired[bool]
    """Whether to write header or not."""

    limit: NotRequired[int]
    """Limit of rows to write."""

header: NotRequired[bool] instance-attribute

Whether to write header or not.

limit: NotRequired[int] instance-attribute

Limit of rows to write.