from collections import namedtuple from typing import List, Any from utils.list_utils import map_dict_keys, flattern, unique import csv import io ExportKey = namedtuple("ExportKey", ["order", "name"]) class ExportItem: def __init__(self, init=None): if init is not None: self.__dict__.update(init) def __getitem__(self, key): return self.__dict__[key] def __delitem__(self, key): del self.__dict__[key] def __contains__(self, key): return key in self.__dict__ def __len__(self): return len(self.__dict__) def __repr__(self): return repr(self.__dict__) def add_item(self, key: str, order: int, value: Any): if value is not None: self.__dict__[ExportKey(order, key)] = value def to_dict(self): return map_dict_keys(self.__dict__, lambda key: key.name) def keys(self): return self.__dict__.keys() class CSVExporter: def __init__(self, items: List[ExportItem] = []): self.export_items = [] self.export_items.extend(items) def add_export_item(self, item: ExportItem): self.export_items.append(item) def export(self): with io.StringIO() as f: all_keys = flattern([list(item.keys()) for item in self.export_items]) campaigns_to_export = [item.to_dict() for item in self.export_items] headers = list(map(lambda key: key.name, sorted(unique(all_keys), key=lambda x: (x.order, x.name)))) writer = csv.DictWriter(f, fieldnames=headers) writer.writeheader() [writer.writerow(campaign_dict_repr) for campaign_dict_repr in campaigns_to_export] f.seek(0) doc = f.read() f.close() return doc