Skip to content

pygx.io

Pluggable IO with unified interfaces.

io

Pluggable IO with unified interfaces.

pygx.io defines abstract FileSystem and File interfaces, plus a Sequence abstraction for append-only record streams. Module-level helpers (open, readfile, writefile, mkdirs, open_sequence, ...) dispatch each path to the registered FileSystem whose prefix matches it, so the same call works on local paths, in-memory paths (/mem/...) and remote URIs. Implementations include:

  • An in-memory backend (MemoryFile, MemoryFileSystem, MemorySequence) useful for tests and ephemeral fixtures.
  • A fsspec bridge (FsspecFile, FsspecFileSystem) that delegates to any filesystem registered with fsspec (local, GCS, S3, etc.).
  • A line-based sequence type (LineSequence, LineSequenceIO) used by JSONL serialization in pg.open_jsonl.

External code can register additional FileSystem implementations and have the rest of PyGX (save/load/open_jsonl) pick them up transparently.

File

Interface for a file.

read abstractmethod 

read(size: int | None = None) -> str | bytes

Reads the content of the file.

Source code in pygx/io/_file_system.py 
40
41
42
@abc.abstractmethod
def read(self, size: int | None = None) -> str | bytes:
    """Reads the content of the file."""

readline abstractmethod 

readline() -> str | bytes

Reads the next line.

Source code in pygx/io/_file_system.py 
44
45
46
@abc.abstractmethod
def readline(self) -> str | bytes:
    """Reads the next line."""

write abstractmethod 

write(content: str | bytes) -> None

Writes content to the file.

Source code in pygx/io/_file_system.py 
48
49
50
@abc.abstractmethod
def write(self, content: str | bytes) -> None:
    """Writes content to the file."""

seek abstractmethod 

seek(offset: int, whence: Literal[0, 1, 2] = 0) -> int

Changes the current position of the file.

Parameters:

Name Type Description Default
offset int

Offset from the position to a reference point.

 required
whence Literal[0, 1, 2]

The reference point, with 0 meaning the beginning of the file, 1 meaning the current position, or 2 meaning the end of the file.

 0

Returns:

Type Description
int

The position from the beginning of the file.
Source code in pygx/io/_file_system.py 
52
53
54
55
56
57
58
59
60
61
62
63
@abc.abstractmethod
def seek(self, offset: int, whence: Literal[0, 1, 2] = 0) -> int:
    """Changes the current position of the file.

    Args:
      offset: Offset from the position to a reference point.
      whence: The reference point, with 0 meaning the beginning of the file,
        1 meaning the current position, or 2 meaning the end of the file.

    Returns:
      The position from the beginning of the file.
    """

tell abstractmethod 

tell() -> int

Returns the current position of the file.

Source code in pygx/io/_file_system.py 
65
66
67
@abc.abstractmethod
def tell(self) -> int:
    """Returns the current position of the file."""

flush abstractmethod 

flush() -> None

Flushes the written content to the storage.

Source code in pygx/io/_file_system.py 
69
70
71
@abc.abstractmethod
def flush(self) -> None:
    """Flushes the written content to the storage."""

close abstractmethod 

close() -> None

Closes the file.

Source code in pygx/io/_file_system.py 
73
74
75
@abc.abstractmethod
def close(self) -> None:
    """Closes the file."""

FileSystem

Interface for a file system.

open abstractmethod 

open(path: str | PathLike[str], mode: str = 'r', **kwargs) -> File

Opens a file with a path.

Source code in pygx/io/_file_system.py 
88
89
90
91
92
@abc.abstractmethod
def open(
    self, path: str | os.PathLike[str], mode: str = 'r', **kwargs
) -> File:
    """Opens a file with a path."""

chmod abstractmethod 

chmod(path: str | PathLike[str], mode: int) -> None

Changes the permission of a file.

Source code in pygx/io/_file_system.py 
94
95
96
@abc.abstractmethod
def chmod(self, path: str | os.PathLike[str], mode: int) -> None:
    """Changes the permission of a file."""

exists abstractmethod 

exists(path: str | PathLike[str]) -> bool

Returns True if a path exists.

Source code in pygx/io/_file_system.py 
 98
 99
100
@abc.abstractmethod
def exists(self, path: str | os.PathLike[str]) -> bool:
    """Returns True if a path exists."""

glob abstractmethod 

glob(pattern: str | PathLike[str]) -> list[str]

Returns the paths matching a glob pattern.

Source code in pygx/io/_file_system.py 
102
103
104
@abc.abstractmethod
def glob(self, pattern: str | os.PathLike[str]) -> list[str]:
    """Returns the paths matching a glob pattern."""

listdir abstractmethod 

listdir(path: str | PathLike[str]) -> list[str]

Lists all files or sub-directories under a directory.

Source code in pygx/io/_file_system.py 
106
107
108
@abc.abstractmethod
def listdir(self, path: str | os.PathLike[str]) -> list[str]:
    """Lists all files or sub-directories under a directory."""

isdir abstractmethod 

isdir(path: str | PathLike[str]) -> bool

Returns True if a path is a directory.

Source code in pygx/io/_file_system.py 
110
111
112
@abc.abstractmethod
def isdir(self, path: str | os.PathLike[str]) -> bool:
    """Returns True if a path is a directory."""

getctime abstractmethod 

getctime(path: str | PathLike[str]) -> float

Returns the creation time of a file.

Source code in pygx/io/_file_system.py 
114
115
116
@abc.abstractmethod
def getctime(self, path: str | os.PathLike[str]) -> float:
    """Returns the creation time of a file."""

getmtime abstractmethod 

getmtime(path: str | PathLike[str]) -> float

Returns the last modification time of a file.

Source code in pygx/io/_file_system.py 
118
119
120
@abc.abstractmethod
def getmtime(self, path: str | os.PathLike[str]) -> float:
    """Returns the last modification time of a file."""

mkdir abstractmethod 

mkdir(path: str | PathLike[str], mode: int = 511) -> None

Makes a directory based on a path.

Source code in pygx/io/_file_system.py 
122
123
124
@abc.abstractmethod
def mkdir(self, path: str | os.PathLike[str], mode: int = 0o777) -> None:
    """Makes a directory based on a path."""

mkdirs abstractmethod 

mkdirs(
    path: str | PathLike[str], mode: int = 511, exist_ok: bool = True
) -> None

Makes a directory chain based on a path.

Source code in pygx/io/_file_system.py 
126
127
128
129
130
131
132
133
@abc.abstractmethod
def mkdirs(
    self,
    path: str | os.PathLike[str],
    mode: int = 0o777,
    exist_ok: bool = True,
) -> None:
    """Makes a directory chain based on a path."""

rm abstractmethod 

rm(path: str | PathLike[str]) -> None

Removes a file based on a path.

Source code in pygx/io/_file_system.py 
135
136
137
@abc.abstractmethod
def rm(self, path: str | os.PathLike[str]) -> None:
    """Removes a file based on a path."""

rename abstractmethod 

rename(oldpath: str | PathLike[str], newpath: str | PathLike[str]) -> None

Renames a file based on a path.

Source code in pygx/io/_file_system.py 
139
140
141
142
143
144
145
@abc.abstractmethod
def rename(
    self,
    oldpath: str | os.PathLike[str],
    newpath: str | os.PathLike[str],
) -> None:
    """Renames a file based on a path."""

rmdir abstractmethod 

rmdir(path: str | PathLike[str]) -> None

Removes a directory based on a path.

Source code in pygx/io/_file_system.py 
147
148
149
@abc.abstractmethod
def rmdir(self, path: str | os.PathLike[str]) -> None:
    """Removes a directory based on a path."""

rmdirs abstractmethod 

rmdirs(path: str | PathLike[str]) -> None

Removes a directory chain based on a path.

Source code in pygx/io/_file_system.py 
151
152
153
@abc.abstractmethod
def rmdirs(self, path: str | os.PathLike[str]) -> None:
    """Removes a directory chain based on a path."""

copy abstractmethod 

copy(oldpath: str | PathLike[str], newpath: str | PathLike[str]) -> None

Copies a file to a new path.

Source code in pygx/io/_file_system.py 
155
156
157
158
159
160
161
@abc.abstractmethod
def copy(
    self,
    oldpath: str | os.PathLike[str],
    newpath: str | os.PathLike[str],
) -> None:
    """Copies a file to a new path."""

StdFile 

StdFile(file_object, path: str | PathLike[str])

Bases: File

The standard file.

Source code in pygx/io/_file_system.py 
180
181
182
183
def __init__(self, file_object, path: str | os.PathLike[str]) -> None:
    super().__init__()
    self._file_object = file_object
    self._path = path

StdFileSystem

Bases: FileSystem

The standard file system.

MemoryFile 

MemoryFile(buffer: IOBase)

Bases: File

Memory file.

Source code in pygx/io/_file_system.py 
284
285
286
287
288
289
290
def __init__(self, buffer: io.IOBase):
    super().__init__()
    self._buffer = buffer
    self._pos = 0
    now = datetime.datetime.now().timestamp()
    self.ctime = now
    self.mtime = now

MemoryFileSystem 

MemoryFileSystem(prefix: str = '/mem/')

Bases: FileSystem

The in-memory file system.

Source code in pygx/io/_file_system.py 
318
319
320
321
def __init__(self, prefix: str = '/mem/'):
    super().__init__()
    self._root = {}
    self._prefix = prefix

getctime 

getctime(path: str | PathLike[str]) -> float

Returns the creation time of a file.

Source code in pygx/io/_file_system.py 
385
386
387
388
389
390
391
392
def getctime(self, path: str | os.PathLike[str]) -> float:
    """Returns the creation time of a file."""
    f = self._locate(path)
    if isinstance(f, dict):
        raise IsADirectoryError(path)
    if f is None:
        raise FileNotFoundError(path)
    return f.ctime

getmtime 

getmtime(path: str | PathLike[str]) -> float

Returns the last modification time of a file.

Source code in pygx/io/_file_system.py 
394
395
396
397
398
399
400
401
def getmtime(self, path: str | os.PathLike[str]) -> float:
    """Returns the last modification time of a file."""
    f = self._locate(path)
    if isinstance(f, dict):
        raise IsADirectoryError(path)
    if f is None:
        raise FileNotFoundError(path)
    return f.mtime

FsspecFile 

FsspecFile(f)

Bases: File

File object based on fsspec.

Source code in pygx/io/_file_system.py 
640
641
def __init__(self, f):
    self._f = f

FsspecFileSystem

Bases: FileSystem

File system based on fsspec.

Sequence 

Sequence(
    perms: int | None = None,
    serializer: Callable[[Any], bytes | str] | None = None,
    deserializer: Callable[[bytes | str], Any] | None = None,
)

Interface for a sequence of records.

Source code in pygx/io/_sequence.py 
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    perms: int | None = None,
    serializer: Callable[[Any], bytes | str] | None = None,
    deserializer: Callable[[bytes | str], Any] | None = None,
):
    self._perms = perms
    self._serializer = serializer
    self._deserializer = deserializer

add 

add(record: Any) -> None

Appends a record to the sequence.

The record is first converted with the serializer given at open time, when one was provided; the result must be a str or bytes.

Source code in pygx/io/_sequence.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def add(self, record: Any) -> None:
    """Appends a record to the sequence.

    The record is first converted with the `serializer` given at open
    time, when one was provided; the result must be a str or bytes.
    """
    if self._serializer:
        record = self._serializer(record)
    if not isinstance(record, (str, bytes)):
        raise ValueError(
            f'Cannot write record with type {type(record)}. '
            'Did you forget to pass a serializer?'
        )
    self._add(record)

close abstractmethod 

close() -> None

Closes the sequence.

Source code in pygx/io/_sequence.py 
79
80
81
@abc.abstractmethod
def close(self) -> None:
    """Closes the sequence."""

flush abstractmethod 

flush() -> None

Flushes written records to the storage.

Source code in pygx/io/_sequence.py 
83
84
85
@abc.abstractmethod
def flush(self) -> None:
    """Flushes written records to the storage."""

SequenceIO

Interface for a record IO system.

open abstractmethod 

open(
    path: str | PathLike[str],
    mode: str,
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
    **kwargs
) -> Sequence

Opens a sequence for reading or writing.

Source code in pygx/io/_sequence.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
@abc.abstractmethod
def open(
    self,
    path: str | os.PathLike[str],
    mode: str,
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
    **kwargs,
) -> Sequence:
    """Opens a sequence for reading or writing."""

MemorySequence 

MemorySequence(
    path: str | PathLike[str],
    mode: str,
    records: list[str | bytes],
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None
)

Bases: Sequence

An in-memory sequence.

Source code in pygx/io/_sequence.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def __init__(
    self,
    path: str | os.PathLike[str],
    mode: str,
    records: list[str | bytes],
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
):
    super().__init__(perms, serializer, deserializer)
    self._path = path
    self._mode = mode
    self._records = records
    self._closed = False

MemorySequenceIO 

MemorySequenceIO()

Bases: SequenceIO

Memory-based record IO.

Source code in pygx/io/_sequence.py 
249
250
251
def __init__(self):
    super().__init__()
    self._root = collections.defaultdict(list)

open 

open(
    path: str | PathLike[str],
    mode: str,
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
    **kwargs
) -> Sequence

Opens a reader for a sequence.

Source code in pygx/io/_sequence.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def open(
    self,
    path: str | os.PathLike[str],
    mode: str,
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
    **kwargs,
) -> Sequence:
    """Opens a reader for a sequence."""
    del kwargs
    if 'w' in mode:
        self._root[path] = []
    return MemorySequence(
        path,
        mode,
        self._root[path],
        perms=perms,
        serializer=serializer,
        deserializer=deserializer,
    )

LineSequence 

LineSequence(
    path: str | PathLike[str],
    mode: str,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
)

Bases: Sequence

A sequence of records stored as newline-delimited lines in a file.

Source code in pygx/io/_sequence.py 
283
284
285
286
287
288
289
290
291
292
293
294
def __init__(
    self,
    path: str | os.PathLike[str],
    mode: str,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
) -> None:
    super().__init__(perms, serializer, deserializer)
    self._path = path
    self._mode = mode
    self._file = file_system.open(path, mode)

LineSequenceIO

Bases: SequenceIO

Line-based record IO.

open 

open(
    path: str | PathLike[str],
    mode: str,
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
    **kwargs
) -> Sequence

Opens a reader for a sequence.

Source code in pygx/io/_sequence.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
def open(
    self,
    path: str | os.PathLike[str],
    mode: str,
    *,
    perms: int | None,
    serializer: Callable[[Any], bytes | str] | None,
    deserializer: Callable[[bytes | str], Any] | None,
    **kwargs,
) -> Sequence:
    """Opens a reader for a sequence."""
    del kwargs
    return LineSequence(path, mode, perms, serializer, deserializer)

add_file_system 

add_file_system(prefix: str, fs: FileSystem) -> None

Registers a file system for paths that start with a prefix.

Every pygx.io helper (open, readfile, mkdirs, ...) dispatches to the registered file system with the longest prefix matching the path.

Source code in pygx/io/_file_system.py 
594
595
596
597
598
599
600
def add_file_system(prefix: str, fs: FileSystem) -> None:
    """Registers a file system for paths that start with a prefix.

    Every `pygx.io` helper (`open`, `readfile`, `mkdirs`, ...) dispatches to
    the registered file system with the longest prefix matching the path.
    """
    _fs.add(prefix, fs)

open 

open(path: str | PathLike[str], mode: str = 'r', **kwargs) -> File

Opens a file with a path.

Source code in pygx/io/_file_system.py 
897
898
899
def open(path: str | os.PathLike[str], mode: str = 'r', **kwargs) -> File:  # pylint:disable=redefined-builtin
    """Opens a file with a path."""
    return _fs.get(path).open(path, mode, **kwargs)

chmod 

chmod(path: str | PathLike[str], mode: int) -> None

Changes the permission of a file.

Source code in pygx/io/_file_system.py 
902
903
904
def chmod(path: str | os.PathLike[str], mode: int) -> None:
    """Changes the permission of a file."""
    _fs.get(path).chmod(path, mode)

readfile 

readfile(
    path: str | PathLike[str],
    mode: str = "r",
    nonexist_ok: bool = False,
    **kwargs
) -> bytes | str | None

Reads content from a file.

Returns None instead of raising FileNotFoundError when nonexist_ok is True and the path does not exist.

Source code in pygx/io/_file_system.py 
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
def readfile(
    path: str | os.PathLike[str],
    mode: str = 'r',
    nonexist_ok: bool = False,
    **kwargs,
) -> bytes | str | None:
    """Reads content from a file.

    Returns None instead of raising ``FileNotFoundError`` when `nonexist_ok`
    is True and the path does not exist.
    """
    try:
        with _fs.get(path).open(path, mode=mode, **kwargs) as f:
            return f.read()
    except FileNotFoundError as e:
        if nonexist_ok:
            return None
        raise e

writefile 

writefile(
    path: str | PathLike[str],
    content: str | bytes,
    *,
    mode: str = "w",
    perms: int | None = 436,
    **kwargs
) -> None

Writes content to a file.

After writing, the file's permission bits are set to perms (0o664 — world-readable — by default) unless perms is None.

Source code in pygx/io/_file_system.py 
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
def writefile(
    path: str | os.PathLike[str],
    content: str | bytes,
    *,
    mode: str = 'w',
    perms: int | None = 0o664,  # Default to world-readable.
    **kwargs,
) -> None:
    """Writes content to a file.

    After writing, the file's permission bits are set to `perms`
    (``0o664`` — world-readable — by default) unless `perms` is None.
    """
    with _fs.get(path).open(path, mode=mode, **kwargs) as f:
        f.write(content)
    if perms is not None:
        chmod(path, perms)

rename 

rename(oldpath: str | PathLike[str], newpath: str | PathLike[str]) -> None

Renames a file.

Source code in pygx/io/_file_system.py 
946
947
948
949
950
951
def rename(
    oldpath: str | os.PathLike[str],
    newpath: str | os.PathLike[str],
) -> None:
    """Renames a file."""
    _fs.get(oldpath).rename(oldpath, newpath)

rm 

rm(path: str | PathLike[str]) -> None

Removes a file.

Source code in pygx/io/_file_system.py 
954
955
956
def rm(path: str | os.PathLike[str]) -> None:
    """Removes a file."""
    _fs.get(path).rm(path)

copy 

copy(oldpath: str | PathLike[str], newpath: str | PathLike[str]) -> None

Copies a file.

Source code in pygx/io/_file_system.py 
959
960
961
962
963
964
def copy(
    oldpath: str | os.PathLike[str],
    newpath: str | os.PathLike[str],
) -> None:
    """Copies a file."""
    _fs.get(oldpath).copy(oldpath, newpath)

path_exists 

path_exists(path: str | PathLike[str]) -> bool

Returns True if path exists.

Source code in pygx/io/_file_system.py 
967
968
969
def path_exists(path: str | os.PathLike[str]) -> bool:
    """Returns True if path exists."""
    return _fs.get(path).exists(path)

glob 

glob(pattern: str | PathLike[str]) -> list[str]

Returns the paths matching a glob pattern.

Source code in pygx/io/_file_system.py 
972
973
974
def glob(pattern: str | os.PathLike[str]) -> list[str]:
    """Returns the paths matching a glob pattern."""
    return _fs.get(pattern).glob(pattern)

listdir 

listdir(path: str | PathLike[str], fullpath: bool = False) -> list[str]

Lists all files or sub-directories under a dir.

Source code in pygx/io/_file_system.py 
977
978
979
980
981
982
def listdir(path: str | os.PathLike[str], fullpath: bool = False) -> list[str]:  # pylint: disable=redefined-builtin
    """Lists all files or sub-directories under a dir."""
    entries = _fs.get(path).listdir(path)
    if fullpath:
        return [os.path.join(path, entry) for entry in entries]
    return entries

isdir 

isdir(path: str | PathLike[str]) -> bool

Returns True if path is a directory.

Source code in pygx/io/_file_system.py 
985
986
987
def isdir(path: str | os.PathLike[str]) -> bool:
    """Returns True if path is a directory."""
    return _fs.get(path).isdir(path)

getctime 

getctime(path: str | PathLike[str]) -> float

Returns the creation time of a file.

Source code in pygx/io/_file_system.py 
990
991
992
def getctime(path: str | os.PathLike[str]) -> float:
    """Returns the creation time of a file."""
    return _fs.get(path).getctime(path)

getmtime 

getmtime(path: str | PathLike[str]) -> float

Returns the last modification time of a file.

Source code in pygx/io/_file_system.py 
995
996
997
def getmtime(path: str | os.PathLike[str]) -> float:
    """Returns the last modification time of a file."""
    return _fs.get(path).getmtime(path)

mkdir 

mkdir(path: str | PathLike[str], mode: int = 511) -> None

Makes a directory.

Source code in pygx/io/_file_system.py 
1000
1001
1002
def mkdir(path: str | os.PathLike[str], mode: int = 0o777) -> None:
    """Makes a directory."""
    _fs.get(path).mkdir(path, mode=mode)

mkdirs 

mkdirs(
    path: str | PathLike[str], mode: int = 511, exist_ok: bool = True
) -> None

Makes a directory chain.

Source code in pygx/io/_file_system.py 
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
def mkdirs(
    path: str | os.PathLike[str],
    mode: int = 0o777,
    exist_ok: bool = True,
) -> None:
    """Makes a directory chain."""
    # An empty path names the current directory, which always exists;
    # `os.makedirs('')` would raise. This lets callers pass
    # `os.path.dirname(p)` through unguarded.
    if not os.fspath(path):
        return
    _fs.get(path).mkdirs(path, mode=mode, exist_ok=exist_ok)

rmdir 

rmdir(path: str | PathLike[str]) -> None

Removes a directory.

Source code in pygx/io/_file_system.py 
1019
1020
1021
def rmdir(path: str | os.PathLike[str]) -> None:
    """Removes a directory."""
    _fs.get(path).rmdir(path)

rmdirs 

rmdirs(path: str | PathLike[str]) -> None

Removes a directory chain until a parent directory is not empty.

Source code in pygx/io/_file_system.py 
1024
1025
1026
def rmdirs(path: str | os.PathLike[str]) -> None:
    """Removes a directory chain until a parent directory is not empty."""
    _fs.get(path).rmdirs(path)

add_sequence_io 

add_sequence_io(extension: str, sequence_io: SequenceIO) -> None

Registers a record IO system for a file extension.

open_sequence picks the IO system by the path's extension (ignoring shard suffixes like @10 or -00001-of-00010); paths with an unregistered extension use line-based records (LineSequenceIO).

Source code in pygx/io/_sequence.py 
142
143
144
145
146
147
148
149
def add_sequence_io(extension: str, sequence_io: SequenceIO) -> None:
    """Registers a record IO system for a file extension.

    `open_sequence` picks the IO system by the path's extension (ignoring
    shard suffixes like ``@10`` or ``-00001-of-00010``); paths with an
    unregistered extension use line-based records (`LineSequenceIO`).
    """
    _registry.add(extension, sequence_io)

open_sequence 

open_sequence(
    path: str | PathLike[str],
    mode: str = "r",
    *,
    perms: int | None = 436,
    serializer: Callable[[Any], bytes | str] | None = None,
    deserializer: Callable[[bytes | str], Any] | None = None,
    make_dirs_if_not_exist: bool = True
) -> Sequence

Opens a sequence of records for reading or writing.

Parameters:

Name Type Description Default
path str | PathLike[str]

The path to the sequence. Its file extension selects the record IO system (see add_sequence_io); unregistered extensions store one record per line (LineSequenceIO).

 required
mode str

The IO mode: 'r' (read), 'w' (write) or 'a' (append).

 'r'
perms int | None

(Optional) The file permission bits to apply when writing.

 436
serializer Callable[[Any], bytes | str] | None

(Optional) A serializer function for converting a structured object to a string or bytes.

 None
deserializer Callable[[bytes | str], Any] | None

(Optional) A deserializer function for converting a string or bytes to a structured object.

 None
make_dirs_if_not_exist bool

(Optional) Whether to create the directories if they do not exist. Applicable when opening in write or append mode.

 True

Returns:

Type Description
Sequence

A sequence for reading or writing.
Source code in pygx/io/_sequence.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def open_sequence(
    path: str | os.PathLike[str],
    mode: str = 'r',
    *,
    perms: int | None = 0o664,  # Default to world-readable.
    serializer: Callable[[Any], bytes | str] | None = None,
    deserializer: Callable[[bytes | str], Any] | None = None,
    make_dirs_if_not_exist: bool = True,
) -> Sequence:
    """Opens a sequence of records for reading or writing.

    Args:
      path: The path to the sequence. Its file extension selects the record
        IO system (see `add_sequence_io`); unregistered extensions store one
        record per line (`LineSequenceIO`).
      mode: The IO mode: 'r' (read), 'w' (write) or 'a' (append).
      perms: (Optional) The file permission bits to apply when writing.
      serializer: (Optional) A serializer function for converting a structured
        object to a string or bytes.
      deserializer: (Optional) A deserializer function for converting a string or
        bytes to a structured object.
      make_dirs_if_not_exist: (Optional) Whether to create the directories
        if they do not exist. Applicable when opening in write or append mode.

    Returns:
      A sequence for reading or writing.
    """
    if 'w' in mode or 'a' in mode:
        parent_dir = os.path.dirname(path)
        if make_dirs_if_not_exist:
            file_system.mkdirs(parent_dir, exist_ok=True)
    return _registry.get(path).open(
        path,
        mode,
        perms=perms,
        serializer=serializer,
        deserializer=deserializer,
    )

options: show_root_heading: false show_root_toc_entry: false members_order: source heading_level: 2