vcorelib.io
index
/home/vkottler/src/vkottler/workspace/vcorelib/vcorelib/io/__init__.py

A module exposing data-file encoders and decoders.

 
Package Contents
       
abc
arbiter (package)
archive (package)
cache
decode
encode
fifo
file_writer
mapping
types

 
Classes
       
abc.ABC(builtins.object)
vcorelib.io.abc.Serializable
vcorelib.io.abc.FileEntity
builtins.object
vcorelib.io.fifo.ByteFifo
vcorelib.io.file_writer.IndentedFileWriter
builtins.tuple(builtins.object)
vcorelib.io.types.LoadResult
enum.Enum(builtins.object)
vcorelib.io.types.FileExtension
vcorelib.io.arbiter.context.DataArbiterContext(vcorelib.io.arbiter.directory.DataArbiterDirectories)
vcorelib.io.arbiter.DataArbiter

 
class ByteFifo(builtins.object)
    ByteFifo() -> None
 
A simple fifo for bytes.
 
  Methods defined here:
__init__(self) -> None
Initialize this instance.
ingest(self, data: bytes) -> None
Append new data to the end.
pop(self, size: int) -> Optional[bytes]
Attempt to read some number of bytes from the front.

Data descriptors defined here:
__dict__
dictionary for instance variables
__weakref__
list of weak references to the object

 
class DataArbiter(vcorelib.io.arbiter.context.DataArbiterContext)
    DataArbiter(logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = <Logger vcorelib.io.arbiter.base (WARNING)>, encoding: str = 'utf-8') -> None
 
A class aggregating all data-arbiter capabilities.
 
 
Method resolution order:
DataArbiter
vcorelib.io.arbiter.context.DataArbiterContext
vcorelib.io.arbiter.directory.DataArbiterDirectories
vcorelib.io.arbiter.base.DataArbiterBase
builtins.object

Methods inherited from vcorelib.io.arbiter.context.DataArbiterContext:
object_directory_context(self, pathlike: Union[pathlib.Path, str, NoneType], decode_kwargs: Dict[str, Any] = None, encode_kwargs: Dict[str, Any] = None) -> Iterator[Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]]]
Provide a loaded directory as a context.
object_file_context(self, pathlike: Union[pathlib.Path, str, NoneType], decode_kwargs: Dict[str, Any] = None, encode_kwargs: Dict[str, Any] = None) -> Iterator[Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]]]
Provide data loaded from a file as a context so that it's written back
when the context ends.

Methods inherited from vcorelib.io.arbiter.directory.DataArbiterDirectories:
decode_directory(self, pathlike: Union[pathlib.Path, str, NoneType], logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, require_success: bool = False, path_filter: Callable[[pathlib.Path], bool] = None, recurse: bool = False, **kwargs) -> vcorelib.io.types.LoadResult
Attempt to decode data files in a directory. Assigns data loaded from
each file to a key, returns whether or not any files failed to load
and the cumulative time that each file-load took.
encode_directory(self, pathlike: Union[pathlib.Path, str, NoneType], data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]], ext: str = 'json', logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, **kwargs) -> Tuple[bool, int]
Encode data to a directory where every key becomes a file with the
provided extension. The encoding scheme is implied by the extension.

Methods inherited from vcorelib.io.arbiter.base.DataArbiterBase:
__init__(self, logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = <Logger vcorelib.io.arbiter.base (WARNING)>, encoding: str = 'utf-8') -> None
Initialize a new data arbiter.
decode(self, pathlike: Union[pathlib.Path, str, NoneType], logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, require_success: bool = False, includes_key: str = None, preprocessor: Callable[[Union[TextIO, _io.StringIO]], Union[TextIO, _io.StringIO]] = None, maxsplit: int = 1, expect_overwrite: bool = False, strategy: vcorelib.dict.MergeStrategy = <MergeStrategy.RECURSIVE: 1>, files_loaded: List[pathlib.Path] = None, **kwargs) -> vcorelib.io.types.LoadResult
Attempt to load data from a file.
decode_stream(self, ext: str, stream: Union[TextIO, _io.StringIO], logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, **kwargs) -> vcorelib.io.types.LoadResult
Attempt to load data from a text stream.
encode(self, pathlike: Union[pathlib.Path, str, NoneType], data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]], logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, maxsplit: int = 1, **kwargs) -> Tuple[bool, int]
Encode data to a file on disk.
encode_stream(self, ext: str, stream: Union[TextIO, _io.StringIO], data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]], logger: Union[logging.Logger, logging.LoggerAdapter[Any]] = None, **kwargs) -> Tuple[bool, int]
Serialize data to an output stream.

Data descriptors inherited from vcorelib.io.arbiter.base.DataArbiterBase:
__dict__
dictionary for instance variables
__weakref__
list of weak references to the object

 
class FileEntity(Serializable)
    FileEntity(location: Union[pathlib.Path, str, NoneType]) -&gt; None
 
A class for working with objects that are backed by a disk location.
 
 
Method resolution order:
FileEntity
Serializable
abc.ABC
builtins.object

Methods defined here:
__hash__(self) -> int
Use this instance's disk location for hashing.
__init__(self, location: Union[pathlib.Path, str, NoneType]) -> None
Initialize this file-backed entity.
default_location(self) -> Union[pathlib.Path, str, NoneType]
Get a default serialization destination for this instance.

Data and other attributes defined here:
__abstractmethods__ = frozenset({'to_stream'})

Methods inherited from Serializable:
to_file(self, path: Union[pathlib.Path, str, NoneType] = None, encoding: str = 'utf-8', **kwargs) -> None
Write this object to a file.
to_stream(self, stream: <class 'TextIO'>, **kwargs) -> None
Write this object to a text stream.

Data descriptors inherited from Serializable:
__dict__
dictionary for instance variables
__weakref__
list of weak references to the object

 
class FileExtension(enum.Enum)
    FileExtension(*values)
 
A mapping of expected encoding type to file extensions.
 
 
Method resolution order:
FileExtension
enum.Enum
builtins.object

Data and other attributes defined here:
INI = <FileExtension.INI: ['ini', 'cfg']>
JINJA = <FileExtension.JINJA: ['j2', 'jinja', 'j2_template', 'j2_macro']>
JSON = <FileExtension.JSON: ['json']>
TAR = <FileExtension.TAR: ['tar.gz', 'tgz', 'tar', 'tar.bz2', 'tar.lzma', 'tar.xz']>
TOML = <FileExtension.TOML: ['toml']>
UNKNOWN = <FileExtension.UNKNOWN: ['unknown']>
YAML = <FileExtension.YAML: ['yaml', 'yml', 'eyaml']>
ZIP = <FileExtension.ZIP: ['zip']>

Data descriptors inherited from enum.Enum:
name
The name of the Enum member.
value
The value of the Enum member.

Methods inherited from enum.EnumType:
__contains__(value) from enum.EnumType
Return True if `value` is in `cls`.
 
`value` is in `cls` if:
1) `value` is a member of `cls`, or
2) `value` is the value of one of the `cls`'s members.
__getitem__(name) from enum.EnumType
Return the member matching `name`.
__iter__() from enum.EnumType
Return members in definition order.
__len__() from enum.EnumType
Return the number of members (no aliases)

Readonly properties inherited from enum.EnumType:
__members__
Returns a mapping of member name->value.
 
This mapping lists all enum members, including aliases. Note that this
is a read-only view of the internal mapping.

 
class IndentedFileWriter(builtins.object)
    IndentedFileWriter(stream: &lt;class 'TextIO'&gt;, space: str = ' ', per_indent: int = 1, prefix: str = '', suffix: str = '', linesep: str = '\n') -&gt; None
 
A class for writing lines to a file and tracking indentation.
 
  Methods defined here:
__init__(self, stream: <class 'TextIO'>, space: str = ' ', per_indent: int = 1, prefix: str = '', suffix: str = '', linesep: str = '\n') -> None
Initialize this instance.
c_comment(self, data: str) -> int
A helper for writing C-style comments.
cpp_comment(self, data: str) -> int
A helper for writing C++-style comments.
dedent(self, amount: int = 1) -> None
Decrease the current indent depth (if not zero).
empty(self, count: int = 1) -> int
Add some number of empty lines.
ends(self, prefix: str = '', suffix: str = '') -> Iterator[NoneType]
Adds a temporary prefix and suffix to lines.
indent(self, amount: int = 1) -> None
Increase the current indent depth.
indented(self, amount: int = 1) -> Iterator[NoneType]
Increase the current indent depth and decrease upon exit.
javadoc(self, opener: str = '/**', closer: str = ' */', prefix: str = ' * ') -> Iterator[NoneType]
A helper for writing javadoc-style comments.
join(self, *lines: str, joiner=',') -> None
Join lines with some joiner (appended to end), except after the last
line.
padding(self, count: int = 1) -> Iterator[NoneType]
Add padding lines as a managed context.
prefix(self, prefix: str) -> Iterator[NoneType]
Set a new line prefix as a managed context.
scope(self, opener: str = '{', closer: str = '}', prefix: str = '', suffix: str = '', indent: int = 1) -> Iterator[NoneType]
A helper for common programming syntax scoping.
suffix(self, suffix: str) -> Iterator[NoneType]
Set a new line suffix as a managed context.
trailing_comment_lines(self, style: vcorelib.io.file_writer.CommentStyle = <CommentStyle.C: 1>, pad: str = ' ', min_pad: int = 1) -> Iterator[List[Tuple[str, Optional[str]]]]
Align indentations for trailing comments.
write(self, data: str) -> int
method taking the str data for a new line of text to write
to the file: first writes the indent (some number of
e.g. space characters), then writes the str data (function parameter),
then writes a newline character (os.linesep).

Static methods defined here:
from_path(path: pathlib.Path, space: str = ' ', per_indent: int = 1, **kwargs) -> Iterator[ForwardRef('IndentedFileWriter')]
Create an instance from a path as a managed context.
string(space: str = ' ', per_indent: int = 1) -> Iterator[ForwardRef('IndentedFileWriter')]
Create an instance for a string.
temporary(space: str = ' ', per_indent: int = 1, **kwargs) -> Iterator[ForwardRef('IndentedFileWriter')]
Create an instance from a temporary file as a managed context.

Data descriptors defined here:
__dict__
dictionary for instance variables
__weakref__
list of weak references to the object

 
class LoadResult(builtins.tuple)
    LoadResult(data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]], success: bool, time_ns: int = -1)
 
An encapsulation of the result of loading raw data, the data collected and
whether or not it succeeded.
 
 
Method resolution order:
LoadResult
builtins.tuple
builtins.object

Methods defined here:
__bool__(self) -> bool
A simple wrapper.
__eq__(self, other: object) -> bool
Don't compare timing when checking equivalence.
__getnewargs__(self)
Return self as a plain tuple.  Used by copy and pickle.
__repr__(self)
Return a nicely formatted representation string
_asdict(self)
Return a new dict which maps field names to their values.
_replace(self, /, **kwds)
Return a new LoadResult object replacing specified fields with new values
merge(self, other: 'LoadResult', is_left: bool = False, **kwargs) -> 'LoadResult'
Merge two load results.
require_success(self, path: Union[pathlib.Path, str]) -> None
Raise a canonical exception if this result is a failure.

Class methods defined here:
_make(iterable) from builtins.type
Make a new LoadResult object from a sequence or iterable

Static methods defined here:
__new__(_cls, data: Dict[str, Union[str, int, float, bool, NoneType, Dict[str, Union[str, int, float, bool, NoneType]], List[Union[str, int, float, bool, NoneType]]]], success: bool, time_ns: int = -1)
Create new instance of LoadResult(data, success, time_ns)

Data descriptors defined here:
data
Alias for field number 0
success
Alias for field number 1
time_ns
Alias for field number 2

Data and other attributes defined here:
__annotations__ = {'data': typing.Dict[str, typing.Union[str, int, float, b...[typing.Union[str, int, float, bool, NoneType]]]], 'success': <class 'bool'>, 'time_ns': <class 'int'>}
__match_args__ = ('data', 'success', 'time_ns')
__orig_bases__ = (<function NamedTuple>,)
_field_defaults = {'time_ns': -1}
_fields = ('data', 'success', 'time_ns')

Methods inherited from builtins.tuple:
__add__(self, value, /)
Return self+value.
__contains__(self, key, /)
Return bool(key in self).
__ge__(self, value, /)
Return self>=value.
__getattribute__(self, name, /)
Return getattr(self, name).
__getitem__(self, key, /)
Return self[key].
__gt__(self, value, /)
Return self>value.
__hash__(self, /)
Return hash(self).
__iter__(self, /)
Implement iter(self).
__le__(self, value, /)
Return self<=value.
__len__(self, /)
Return len(self).
__lt__(self, value, /)
Return self<value.
__mul__(self, value, /)
Return self*value.
__ne__(self, value, /)
Return self!=value.
__rmul__(self, value, /)
Return value*self.
count(self, value, /)
Return number of occurrences of value.
index(self, value, start=0, stop=9223372036854775807, /)
Return first index of value.
 
Raises ValueError if the value is not present.

Class methods inherited from builtins.tuple:
__class_getitem__(...) from builtins.type
See PEP 585

 
class Serializable(abc.ABC)
    An interface definition for serializable classes.
 
 
Method resolution order:
Serializable
abc.ABC
builtins.object

Methods defined here:
default_location(self) -> Union[pathlib.Path, str, NoneType]
Get a default serialization destination for this instance.
to_file(self, path: Union[pathlib.Path, str, NoneType] = None, encoding: str = 'utf-8', **kwargs) -> None
Write this object to a file.
to_stream(self, stream: <class 'TextIO'>, **kwargs) -> None
Write this object to a text stream.

Data descriptors defined here:
__dict__
dictionary for instance variables
__weakref__
list of weak references to the object

Data and other attributes defined here:
__abstractmethods__ = frozenset({'to_stream'})

 
Data
        ARBITER = <vcorelib.io.arbiter.DataArbiter object>
DEFAULT_INCLUDES_KEY = 'includes'
DataDecoder = typing.Callable[[typing.Union[typing.TextIO, _io...pter[typing.Any]]], vcorelib.io.types.LoadResult]
DataEncoder = typing.Callable[[typing.Dict[str, typing.Union[s...Logger, logging.LoggerAdapter[typing.Any]]], int]
DataStream = typing.Union[typing.TextIO, _io.StringIO]
EncodeResult = typing.Tuple[bool, int]
JsonArray = typing.List[typing.Union[str, int, float, bool, ...[typing.Union[str, int, float, bool, NoneType]]]]
JsonObject = typing.Dict[str, typing.Union[str, int, float, b...[typing.Union[str, int, float, bool, NoneType]]]]
JsonPrimitive = typing.Union[str, int, float, bool, NoneType]
JsonValue = typing.Union[str, int, float, bool, NoneType, ty...t[typing.Union[str, int, float, bool, NoneType]]]
StreamProcessor = typing.Callable[[typing.Union[typing.TextIO, _io...gIO]], typing.Union[typing.TextIO, _io.StringIO]]
__all__ = ['DataArbiter', 'ARBITER', 'JsonPrimitive', 'JsonValue', 'JsonArray', 'JsonObject', 'FileExtension', 'LoadResult', 'EncodeResult', 'DataStream', 'StreamProcessor', 'DataDecoder', 'DataEncoder', 'Serializable', 'FileEntity', 'DEFAULT_INCLUDES_KEY', 'IndentedFileWriter', 'ByteFifo']