directory.archive

Module Contents

Classes

_Sentinel	Generic enumeration.
FieldParser	Parses records read by the directory archive reader.
DirectoryArchiveReader	Reading part of DirectoryArchive.
DirectoryArchiveWriter	Writing part of DirectoryArchive.
DirectoryArchive	Offers the ability to read/write a directory and its entries to a
DirectoryZipArchive	Offers the same interface as the DirectoryArchive, additionally

Attributes

UnknownFieldType
UNKNOWN_FIELD

directory.archive.UnknownFieldType: typing_extensions.TypeAlias = 'Literal[_Sentinel.UNKNOWN_FIELD]'

class directory.archive._Sentinel

Bases: enum.Enum

Generic enumeration.

Derive from this class to define new enumerations.

UNKNOWN_FIELD

directory.archive.UNKNOWN_FIELD

exception directory.archive.DirectoryFileNotFound(file_id: str, entry_name: str, filename: str)

Bases: FileNotFoundError

File not found.

class directory.archive.FieldParser(directory: onegov.directory.models.Directory, archive_path: pathlib.Path)

Parses records read by the directory archive reader.

get_field(key: str) → ParsedField | None

CSV Files header parsing is inconsistent with the the internal id ( field.id) of the field. The headers are lovercased, so that the first will not yield the field, the second will also not success because characters like ( are not replaced by underscores.

parse_fileinput(key: str, value: str, field: onegov.form.parser.core.FileinputField) → onegov.core.utils.Bunch | None

parse_multiplefileinput(key: str, value: str, field: onegov.form.parser.core.MultipleFileinputField) → tuple[onegov.core.utils.Bunch, Ellipsis]

parse_generic(key: str, value: str, field: onegov.form.parser.core.ParsedField) → object

parse_item(key: str, value: str) → tuple[str, Any | None] | UnknownFieldType

parse(record: SupportsItems[str, str]) → dict[str, Any | None]

class directory.archive.DirectoryArchiveReader

Reading part of DirectoryArchive.

path: pathlib.Path

read(target: onegov.directory.models.Directory | None = None, skip_existing: bool = True, limit: int = 0, apply_metadata: bool = True, after_import: Callable[[DirectoryEntry], Any] | None = None) → onegov.directory.models.Directory

Reads the archive resulting in a dictionary and entries.

Parameters:

• target – Uses the given directory as a target for the read. Otherwise, a new directory is created in memory (default).

• skip_existing – Excludes already existing entries from being added to the directory. Only applies if target is not None.

• limit – Limits the number of records which are imported. If the limit is reached, the read process silently ignores all extra items.

• apply_metadata – True if the metadata found in the archive should be applied to the directory.

• after_import – Called with the newly added entry, right after it has been added.

apply_metadata(directory: onegov.directory.models.Directory, metadata: dict[str, Any]) → onegov.directory.models.Directory

Applies the metadata to the given directory and returns it.

read_metadata() → dict[str, Any]

Returns the metadata as a dictionary.

read_data() → Sequence[dict[str, Any]]

Returns the entries as a sequence of dictionaries.

read_data_from_json() → list[dict[str, Any]]

read_data_from_csv() → tuple[dict[str, Any], Ellipsis]

read_data_from_xlsx() → tuple[dict[str, Any], Ellipsis]

class directory.archive.DirectoryArchiveWriter

Writing part of DirectoryArchive.

path: pathlib.Path

format: Literal[onegov.core.custom.json, csv, xlsx]

transform: FieldValueTransform

write(directory: onegov.directory.models.Directory, *args: Any, entry_filter: DirectoryEntryFilter | None = None, query: Query[DirectoryEntry] | None = None, **kwargs: Any) → None

Writes the given directory.

write_directory_metadata(directory: onegov.directory.models.Directory) → None

Writes the metadata.

write_directory_entries(directory: onegov.directory.models.Directory, entry_filter: DirectoryEntryFilter | None = None, query: Query[DirectoryEntry] | None = None) → None

Writes the directory entries. Allows filtering with custom entry_filter function as well as passing a query object

write_paths(session: sqlalchemy.orm.Session, paths: dict[str, str], fid_to_entry: dict[str, str] | None = None) → None

Writes the given files to the archive path.

Parameters:

• session – The database session in use.

• paths – A dictionary with each key being a file id and each value being a path where this file id should be written to.

• fid_to_entry – A dictionary with the mapping of the file id to the entry name

write_json(path: pathlib.Path, data: onegov.core.types.JSON_ro) → None

write_xlsx(path: pathlib.Path, data: Iterable[dict[str, Any]]) → None

write_csv(path: pathlib.Path, data: Iterable[dict[str, Any]]) → None

class directory.archive.DirectoryArchive(path: _typeshed.StrPath, format: Literal[onegov.core.custom.json, csv, xlsx] = 'json', transform: FieldValueTransform | None = None)

Bases: DirectoryArchiveReader, DirectoryArchiveWriter

Offers the ability to read/write a directory and its entries to a folder.

Usage:

archive = DirectoryArchive('/tmp/directory')
archive.write()

archive = DirectoryArchive('/tmp/directory')
archive.read()

The archive content is as follows:

• metadata.json (contains the directory data)

• data.json/data.csv/data.xlsx (contains the directory entries)

• ./<field_id>/<entry_id>.<ext> (files referenced by the directory entries)

The directory entries are stored as json, csv or xlsx. Json is preferred.

class directory.archive.DirectoryZipArchive(path: _typeshed.StrPath, *args: Any, **kwargs: Any)

Offers the same interface as the DirectoryArchive, additionally zipping the folder on write and extracting the zip on read.

format: Literal[zip] = 'zip'

classmethod from_buffer(buffer: SupportsReadAndSeek) → typing_extensions.Self

Creates a zip archive instance from a file object in memory.

write(directory: onegov.directory.models.Directory, *args: Any, **kwargs: Any) → None

read(*args: Any, **kwargs: Any) → onegov.directory.models.Directory

compress() → None

extract() → None