Source code for diggrtoolbox.zipaccess.zip_access

#!/usr/bin/env python3
"""
Zip Access is a small tool providing access to zipped
json files.
"""

import zipfile
import json

FILE_EXT = ".json"


[docs]class ZipAccess: """ Baseclass for the ZipSingleAccess and ZipMultiAccess classes """ def __init__(self, filename, file_ext=FILE_EXT): if zipfile.is_zipfile(filename): self.filename = filename self.z = zipfile.ZipFile(self.filename) else: raise zipfile.BadZipFile self.file_ext = file_ext
[docs] def json(self, content_filename=None): """ Opens the zipfile and returns the first zipped JSON file as python object """ if not content_filename: content_filename = self.z.filelist[0].filename with self.z.open(content_filename) as content_file: content = content_file.read() return json.loads(content.decode("utf-8"))
[docs]class ZipSingleAccess(ZipAccess): """ This class is meant to provide access to a single JSON-file in a zipfile. """
[docs] def json(self): """ Opens the zipfile and returns the zipped JSON file as python object """ if len(self.z.filelist) > 1: raise ValueError("Not more than one file per zipfile allowed.") return super().json()
[docs]class ZipMultiAccess(ZipAccess): """ This class is meant to provide access to a Zip file containing one base json file and a folder with other json files extending the first ZipMultiAccess provides a __getitem__ method to allow more easy access to the contents. """
[docs] def get(self, file_id): """ Returns a specific object, which is not the base object. :param file_id: Identifier of the object to be returned. :type file_id: str """ if not isinstance(file_id, str): file_id = str(file_id) for filename in self.z.namelist(): if "/" in filename: if file_id == filename.split("/")[1][:-len(self.file_ext)]: return self.json(filename) else: raise ValueError("Item {} not found.".format(file_id))
def __getitem__(self, file_id): """ Wrapper around get, to allow for more pythonic object access. :param file_id: Identifier of the object to be returned. :type file_id: str """ return self.get(file_id)
[docs]class ZipListAccess(ZipAccess): """ Class to read a Zipfile. """ def __iter__(self): """ Iterates through zip archive. """ for filename in self.z.namelist(): with self.z.open(filename) as f: data = f.read().decode("utf-8") yield json.loads(data)
[docs] def read_archive(self): """ Reads archive zipfile and returns contents as list of dicts. """ object_list = [] for data in self: object_list.append(data) return object_list
def __getitem__(self, filename): """ Access specific file in zip archive. """ return self.json(content_filename=filename)