Source code for swdata.reports

import csv
import io
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterator

import petl as etl


@dataclass
class ReportViewer:
    """Data viewer for reports saved on disc"""

    path: str

    def __getitem__(self, item):
        if isinstance(item, int):
            item = slice(item, item + 1)
        return list(etl.fromcsv(self.path, delimiter=',')[item])

    def distinct(self, *columns):
        return list(
            etl.fromcsv(self.path, delimiter=',').cut(columns).distinct(key=columns, count='count')
        )


@dataclass
class Report(ReportViewer):
    """Report metadata DTO"""

    when_saved: datetime


def save_csv(path: str, data: Iterator):
    """
    Utility function to save data to a CSV file.
    """
    print(f'---> Saving: {Path(path)}')
    with io.open(path, 'w', newline='\n') as csvfile:
        writer = csv.writer(csvfile)
        for row in data:
            writer.writerow(row)


def save_report(directory: str, data: Iterator, timestamp_fmt='%Y-%m-%d-%H%M%S') -> Report:
    """
    Utility function create a report file with timestamp in the name
    """
    now = datetime.now()
    timestamp = now.strftime(timestamp_fmt)
    path = Path(f'{directory}/sw-people-{timestamp}.csv').absolute()
    if not path.parent.exists():
        raise ValueError(f'Directory does not exist: {path.parent}')
    save_csv(path=path, data=data)
    return Report(when_saved=now, path=path)