vcorelib.io package

vcorelib.io package#

Submodules#

vcorelib.io.abc module#

A module for abstract base-classes relevant to IO.

class vcorelib.io.abc.FileEntity(location: Path | str | None)[source]#

Bases: Serializable

A class for working with objects that are backed by a disk location.

default_location() → Path | str | None[source]#: Get a default serialization destination for this instance.

class vcorelib.io.abc.Serializable[source]#

Bases: ABC

An interface definition for serializable classes.

default_location() → Path | str | None[source]#: Get a default serialization destination for this instance.

to_file(path: Path | str | None = None, encoding: str = 'utf-8', **kwargs) → None[source]#: Write this object to a file.

abstract to_stream(stream: TextIO, **kwargs) → None[source]#: Write this object to a text stream.

vcorelib.io.cache module#

A module for cache implementations - conforming to package-wide, data-structure constraints and assumptions.

A class implementing a dictionary that can be saved and loaded from disk, with a specified encoding scheme.

clean(path: Path = None, logger: Logger = None, level: int = 10) → None[source]#: Remove cached data from disk.

load(path: Path = None, logger: Logger = None, level: int = 10, **kwargs) → Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]][source]#: Load data from disk.

load_directory(path: Path | str | None, data: Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]], **kwargs) → int[source]#: Load a directory and update data, return time taken to load.

save(path: Path = None, logger: Logger = None, level: int = 10, archive: bool = False, **kwargs) → None[source]#: Save data to disk.

save_directory(path: Path, **kwargs) → int[source]#: Write data in this cache to a directory.

vcorelib.io.decode module#

A module implementing various data-file decoders.

vcorelib.io.decode.decode_ini(data_file: ~typing.TextIO | ~_io.StringIO, logger: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.decode (WARNING)>, **kwargs) → LoadResult[source]#: Load INI data from a text stream.

vcorelib.io.decode.decode_json(data_file: ~typing.TextIO | ~_io.StringIO, logger: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.decode (WARNING)>, **kwargs) → LoadResult[source]#: Load JSON data from a text stream.

vcorelib.io.decode.decode_toml(data_file: ~typing.TextIO | ~_io.StringIO, logger: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.decode (WARNING)>, **kwargs) → LoadResult[source]#: Load TOML data from a text stream.

vcorelib.io.decode.decode_yaml(data_file: ~typing.TextIO | ~_io.StringIO, logger: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.decode (WARNING)>, **kwargs) → LoadResult[source]#: Load YAML data from a text stream.

vcorelib.io.encode module#

A module implementing various data-file encoders.

vcorelib.io.encode.encode_ini(configs: ~typing.Dict[str, str | int | float | bool | None | ~typing.Dict[str, str | int | float | bool | None] | ~typing.List[str | int | float | bool | None]], ostream: ~typing.TextIO | ~_io.StringIO, _: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.encode (WARNING)>, **kwargs) → int[source]#: Write config data as INI to the output stream.

vcorelib.io.encode.encode_json(configs: ~typing.Dict[str, str | int | float | bool | None | ~typing.Dict[str, str | int | float | bool | None] | ~typing.List[str | int | float | bool | None]], ostream: ~typing.TextIO | ~_io.StringIO, _: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.encode (WARNING)>, **kwargs) → int[source]#: Write config data as JSON to the output stream.

vcorelib.io.encode.encode_toml(configs: ~typing.Dict[str, str | int | float | bool | None | ~typing.Dict[str, str | int | float | bool | None] | ~typing.List[str | int | float | bool | None]], ostream: ~typing.TextIO | ~_io.StringIO, _: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.encode (WARNING)>, **kwargs) → int[source]#: Write config data as TOML to the output stream.

vcorelib.io.encode.encode_yaml(configs: ~typing.Dict[str, str | int | float | bool | None | ~typing.Dict[str, str | int | float | bool | None] | ~typing.List[str | int | float | bool | None]], ostream: ~typing.TextIO | ~_io.StringIO, _: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.encode (WARNING)>, sequence: int = 4, offset: int = 2, mapping: int = 2, document_start: bool = True, **kwargs) → int[source]#: Write config data as YAML to the output stream.

vcorelib.io.fifo module#

A module implementing a simple bytes FIFO interface.

class vcorelib.io.fifo.ByteFifo[source]#

Bases: object

A simple fifo for bytes.

ingest(data: bytes) → None[source]#: Append new data to the end.

pop(size: int) → bytes | None[source]#: Attempt to read some number of bytes from the front.

vcorelib.io.file_writer module#

A module implementing an interface for writing to variably indented files.

class vcorelib.io.file_writer.CommentStyle(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

An enumeration describing different comment styles.

C = 1#

CPP = 3#

C_DOXYGEN = 2#

SCRIPT = 4#

wrap(data: str) → str[source]#: Wrap a string in this comment style.

class vcorelib.io.file_writer.IndentedFileWriter(stream: TextIO, space: str = ' ', per_indent: int = 1, prefix: str = '', suffix: str = '', linesep: str = '\n')[source]#

Bases: object

A class for writing lines to a file and tracking indentation.

c_comment(data: str) → int[source]#: A helper for writing C-style comments.

cpp_comment(data: str) → int[source]#: A helper for writing C++-style comments.

dedent(amount: int = 1) → None[source]#: Decrease the current indent depth (if not zero).

empty(count: int = 1) → int[source]#: Add some number of empty lines.

ends(prefix: str = '', suffix: str = '') → Iterator[None][source]#: Adds a temporary prefix and suffix to lines.

static from_path(path: Path, space: str = ' ', per_indent: int = 1, **kwargs) → Iterator[IndentedFileWriter][source]#: Create an instance from a path as a managed context.

indent(amount: int = 1) → None[source]#: Increase the current indent depth.

indented(amount: int = 1) → Iterator[None][source]#: Increase the current indent depth and decrease upon exit.

javadoc(opener: str = '/**', closer: str = ' */', prefix: str = ' * ') → Iterator[None][source]#: A helper for writing javadoc-style comments.

join(*lines: str, joiner=',') → None[source]#: Join lines with some joiner (appended to end), except after the last line.

padding(count: int = 1) → Iterator[None][source]#: Add padding lines as a managed context.

prefix(prefix: str) → Iterator[None][source]#: Set a new line prefix as a managed context.

scope(opener: str = '{', closer: str = '}', prefix: str = '', suffix: str = '', indent: int = 1) → Iterator[None][source]#: A helper for common programming syntax scoping.

static string(space: str = ' ', per_indent: int = 1) → Iterator[IndentedFileWriter][source]#: Create an instance for a string.

suffix(suffix: str) → Iterator[None][source]#: Set a new line suffix as a managed context.

static temporary(space: str = ' ', per_indent: int = 1, **kwargs) → Iterator[IndentedFileWriter][source]#: Create an instance from a temporary file as a managed context.

trailing_comment_lines(style: CommentStyle = CommentStyle.C, pad: str = ' ', min_pad: int = 1) → Iterator[List[Tuple[str, str | None]]][source]#: Align indentations for trailing comments.

write(data: str) → int[source]#: 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).

vcorelib.io.mapping module#

A module mapping file extensions to encoders and decoders.

Bases: NamedTuple

A description of a data type.

decoder: Callable[[TextIO | StringIO, Logger | LoggerAdapter[Any]], LoadResult]#: Alias for field number 0

encoder: Callable[[Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]], TextIO | StringIO, Logger | LoggerAdapter[Any]], int]#: Alias for field number 1

class vcorelib.io.mapping.DataMapping[source]#

Bases: object

Map interfaces that read and write data formats to the file extensions that most-likely indicate the desire for that kind of format.

static from_ext(ext: FileExtension = None) → DataType | None[source]#: Map a file extension to a data type.

static from_ext_str(ext: str) → DataType | None[source]#: Get a data type from a file-extension string.

mapping = {FileExtension.INI: DataType.INI, FileExtension.JSON: DataType.JSON, FileExtension.TOML: DataType.TOML, FileExtension.YAML: DataType.YAML}#

class vcorelib.io.mapping.DataType(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

An aggregation of all known data types.

INI = (<function decode_ini>, <function encode_ini>)#

JSON = (<function decode_json>, <function encode_json>)#

TOML = (<function decode_toml>, <function encode_toml>)#

YAML = (<function decode_yaml>, <function encode_yaml>)#

vcorelib.io.types module#

Common type definitions for data encoding and decoding interfaces.

class vcorelib.io.types.FileExtension(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

A mapping of expected encoding type to file extensions.

INI: List[str] = ['ini', 'cfg']#

JINJA: List[str] = ['j2', 'jinja', 'j2_template', 'j2_macro']#

JSON: List[str] = ['json']#

TAR: List[str] = ['tar.gz', 'tgz', 'tar', 'tar.bz2', 'tar.lzma', 'tar.xz']#

TOML: List[str] = ['toml']#

UNKNOWN: List[str] = ['unknown']#

YAML: List[str] = ['yaml', 'yml', 'eyaml']#

ZIP: List[str] = ['zip']#

apply(path: Path) → Path[source]#: Apply this extension’s suffix to the given path.

static archive_candidates(path: Path | str | None, exists_only: bool = False) → Iterator[Path][source]#: Iterate over all file extensions that could point to an archive file.

candidates(path: Path | str | None, exists_only: bool = False) → Iterator[Path][source]#: For a given path, iterate over candidate paths that have the suffixes for this kind of file extension.

static data_candidates(path: Path | str | None, exists_only: bool = False) → Iterator[Path][source]#: Iterate over all file extensions that could point to a data file.

static from_ext(ext_str: str) → FileExtension | None[source]#: Given a file extension, determine what kind of file it is.

static from_path(path: Path | str | None, maxsplit: int = 1) → FileExtension | None[source]#: Get a known file extension for a path, if it exists.

static has_archive(path: Path | str | None) → Path | None[source]#: Determine if a path has an associated archive file.

is_archive() → bool[source]#: Determine if this extension is a kind of archive file.

is_data() → bool[source]#: Determine if this etension is a kind of data file.

is_template() → bool[source]#: Determine if this extension is a kind of template.

Bases: NamedTuple

An encapsulation of the result of loading raw data, the data collected and whether or not it succeeded.

data: Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]]#: Alias for field number 0

merge(other: LoadResult, is_left: bool = False, **kwargs) → LoadResult[source]#: Merge two load results.

require_success(path: Path | str) → None[source]#: Raise a canonical exception if this result is a failure.

success: bool#: Alias for field number 1

time_ns: int#: Alias for field number 2

Module contents#

A module exposing data-file encoders and decoders.

class vcorelib.io.ByteFifo[source]#

Bases: object

A simple fifo for bytes.

ingest(data: bytes) → None[source]#: Append new data to the end.

pop(size: int) → bytes | None[source]#: Attempt to read some number of bytes from the front.

class vcorelib.io.DataArbiter(logger: ~logging.Logger | ~logging.LoggerAdapter[~typing.Any] = <Logger vcorelib.io.arbiter.base (WARNING)>, encoding: str = 'utf-8')[source]#

Bases: DataArbiterContext

A class aggregating all data-arbiter capabilities.

class vcorelib.io.FileEntity(location: Path | str | None)[source]#

Bases: Serializable

A class for working with objects that are backed by a disk location.

default_location() → Path | str | None[source]#: Get a default serialization destination for this instance.

class vcorelib.io.FileExtension(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

A mapping of expected encoding type to file extensions.

INI: List[str] = ['ini', 'cfg']#

JINJA: List[str] = ['j2', 'jinja', 'j2_template', 'j2_macro']#

JSON: List[str] = ['json']#

TAR: List[str] = ['tar.gz', 'tgz', 'tar', 'tar.bz2', 'tar.lzma', 'tar.xz']#

TOML: List[str] = ['toml']#

UNKNOWN: List[str] = ['unknown']#

YAML: List[str] = ['yaml', 'yml', 'eyaml']#

ZIP: List[str] = ['zip']#

apply(path: Path) → Path[source]#: Apply this extension’s suffix to the given path.

static archive_candidates(path: Path | str | None, exists_only: bool = False) → Iterator[Path][source]#: Iterate over all file extensions that could point to an archive file.

candidates(path: Path | str | None, exists_only: bool = False) → Iterator[Path][source]#: For a given path, iterate over candidate paths that have the suffixes for this kind of file extension.

static data_candidates(path: Path | str | None, exists_only: bool = False) → Iterator[Path][source]#: Iterate over all file extensions that could point to a data file.

static from_ext(ext_str: str) → FileExtension | None[source]#: Given a file extension, determine what kind of file it is.

static from_path(path: Path | str | None, maxsplit: int = 1) → FileExtension | None[source]#: Get a known file extension for a path, if it exists.

static has_archive(path: Path | str | None) → Path | None[source]#: Determine if a path has an associated archive file.

is_archive() → bool[source]#: Determine if this extension is a kind of archive file.

is_data() → bool[source]#: Determine if this etension is a kind of data file.

is_template() → bool[source]#: Determine if this extension is a kind of template.

class vcorelib.io.IndentedFileWriter(stream: TextIO, space: str = ' ', per_indent: int = 1, prefix: str = '', suffix: str = '', linesep: str = '\n')[source]#

Bases: object

A class for writing lines to a file and tracking indentation.

c_comment(data: str) → int[source]#: A helper for writing C-style comments.

cpp_comment(data: str) → int[source]#: A helper for writing C++-style comments.

dedent(amount: int = 1) → None[source]#: Decrease the current indent depth (if not zero).

empty(count: int = 1) → int[source]#: Add some number of empty lines.

ends(prefix: str = '', suffix: str = '') → Iterator[None][source]#: Adds a temporary prefix and suffix to lines.

static from_path(path: Path, space: str = ' ', per_indent: int = 1, **kwargs) → Iterator[IndentedFileWriter][source]#: Create an instance from a path as a managed context.

indent(amount: int = 1) → None[source]#: Increase the current indent depth.

indented(amount: int = 1) → Iterator[None][source]#: Increase the current indent depth and decrease upon exit.

javadoc(opener: str = '/**', closer: str = ' */', prefix: str = ' * ') → Iterator[None][source]#: A helper for writing javadoc-style comments.

join(*lines: str, joiner=',') → None[source]#: Join lines with some joiner (appended to end), except after the last line.

padding(count: int = 1) → Iterator[None][source]#: Add padding lines as a managed context.

prefix(prefix: str) → Iterator[None][source]#: Set a new line prefix as a managed context.

scope(opener: str = '{', closer: str = '}', prefix: str = '', suffix: str = '', indent: int = 1) → Iterator[None][source]#: A helper for common programming syntax scoping.

static string(space: str = ' ', per_indent: int = 1) → Iterator[IndentedFileWriter][source]#: Create an instance for a string.

suffix(suffix: str) → Iterator[None][source]#: Set a new line suffix as a managed context.

static temporary(space: str = ' ', per_indent: int = 1, **kwargs) → Iterator[IndentedFileWriter][source]#: Create an instance from a temporary file as a managed context.

trailing_comment_lines(style: CommentStyle = CommentStyle.C, pad: str = ' ', min_pad: int = 1) → Iterator[List[Tuple[str, str | None]]][source]#: Align indentations for trailing comments.

write(data: str) → int[source]#: 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).

Bases: NamedTuple

An encapsulation of the result of loading raw data, the data collected and whether or not it succeeded.

data: Dict[str, str | int | float | bool | None | Dict[str, str | int | float | bool | None] | List[str | int | float | bool | None]]#: Alias for field number 0

merge(other: LoadResult, is_left: bool = False, **kwargs) → LoadResult[source]#: Merge two load results.

require_success(path: Path | str) → None[source]#: Raise a canonical exception if this result is a failure.

success: bool#: Alias for field number 1

time_ns: int#: Alias for field number 2

class vcorelib.io.Serializable[source]#

Bases: ABC

An interface definition for serializable classes.

default_location() → Path | str | None[source]#: Get a default serialization destination for this instance.

to_file(path: Path | str | None = None, encoding: str = 'utf-8', **kwargs) → None[source]#: Write this object to a file.

abstract to_stream(stream: TextIO, **kwargs) → None[source]#: Write this object to a text stream.

vcorelib.io package

Contents

vcorelib.io package#

Subpackages#

Submodules#

vcorelib.io.abc module#

vcorelib.io.cache module#

vcorelib.io.decode module#

vcorelib.io.encode module#

vcorelib.io.fifo module#

vcorelib.io.file_writer module#

vcorelib.io.mapping module#

vcorelib.io.types module#

Module contents#