(data, headers, title=None, fmt='', filename=None, append=False, **kwargs)
| 367 | |
| 368 | |
| 369 | def dump_report_data(data, headers, title=None, fmt='', filename=None, append=False, **kwargs): |
| 370 | # type: (List[List], Sequence[str], Optional[str], Optional[str], Optional[str], bool, ...) -> Optional[str] |
| 371 | # kwargs: |
| 372 | # row_number: boolean - Add row number. table only |
| 373 | # column_width: int - Truncate long columns. table only |
| 374 | # no_header: boolean - Do not print header |
| 375 | # group_by: int - Sort and Group by columnNo |
| 376 | # sort_by: int - Sort by columnNo |
| 377 | # sort_desc: bool - Descending Sort |
| 378 | # right_align: Sequence[int] - Force right align |
| 379 | # footer: str - Footer text |
| 380 | # PDF format requires reportlab library to be installed |
| 381 | |
| 382 | sort_by = kwargs.get('sort_by') |
| 383 | group_by = kwargs.get('group_by') |
| 384 | if group_by is not None: |
| 385 | group_by = int(group_by) |
| 386 | sort_by = group_by |
| 387 | |
| 388 | if isinstance(sort_by, int): |
| 389 | key_fn = detect_column_type((x[sort_by] for x in data if 0 <= sort_by < len(x))) |
| 390 | if callable(key_fn): |
| 391 | reverse = kwargs.get('sort_desc') is True |
| 392 | data.sort(key=lambda r: key_fn(r[sort_by] if 0 <= sort_by < len(r) else None), reverse=reverse) |
| 393 | |
| 394 | if fmt == 'csv': |
| 395 | if filename: |
| 396 | _, ext = os.path.splitext(filename) |
| 397 | if not ext: |
| 398 | filename += '.csv' |
| 399 | logging.info('Report path: %s', os.path.abspath(filename)) |
| 400 | with open(filename, 'a' if append else 'w', newline='', encoding='utf-8') if filename else io.StringIO() as fd: |
| 401 | csv_writer = csv.writer(fd) |
| 402 | if title: |
| 403 | csv_writer.writerow([]) |
| 404 | csv_writer.writerow([title]) |
| 405 | csv_writer.writerow([]) |
| 406 | elif append: |
| 407 | csv_writer.writerow([]) |
| 408 | |
| 409 | starting_column = 0 |
| 410 | if headers: |
| 411 | if headers[0] == '#': |
| 412 | starting_column = 1 |
| 413 | csv_writer.writerow(headers[starting_column:]) |
| 414 | for row in data: |
| 415 | for i in range(len(row)): |
| 416 | if isinstance(row[i], list): |
| 417 | row[i] = '\n'.join(row[i]) |
| 418 | csv_writer.writerow(row[starting_column:]) |
| 419 | if isinstance(fd, io.StringIO): |
| 420 | report = fd.getvalue() |
| 421 | if append: |
| 422 | logging.info(report) |
| 423 | else: |
| 424 | return report |
| 425 | elif fmt == 'json': |
| 426 | data_list = [] |
no test coverage detected