Org Web Adapter

Editing pages/eng_12811_force_system_to_process_workday_files_with_invalid_columns.org

:PROPERTIES:
:ID:       d788300f-c740-4ad2-ad58-f3e39e278488
:ROAM_ALIASES: ENG-12811
:END:
#+title: ENG-12811 - Force system to process Workday files with invalid columns

- source :: https://15five-dev.atlassian.net/browse/ENG-12811

* Sample code
:PROPERTIES:
:ID:       62a0399f-79be-4dbd-8e45-5bf172c50449
:END:

#+BEGIN_SRC python :results output
import io
import csv
from typing import BinaryIO

from django.core.files.base import ContentFile

BULK_IMPORT_COLUMNS = [('email', 'Email')]

def convert_to_bytes_io(file: BinaryIO) -> io.BytesIO:
    """
    Read content from file and returns BytesIO object with this content
    """
    data = file.read()
    if not isinstance(data, bytes):
        data = data.encode('utf-8')
    return io.BytesIO(data)

def get_csv_dict_reader(csv_file: io.BytesIO, fieldnames: list = None) -> csv.DictReader:
    """
    Main goal is to transform field names to lower-case
    todo: this code is duplicated at least 2 other places

and also replace windows new-lines to unix ones (see: https://github.com/15five/fifteen5/issues/5940)
    todo: verify this is still required with python 3 + the new way we decode files
    """
    csv_lines = csv_file.read().splitlines()
    if csv_lines:
        csv_lines[0] = csv_lines[0].lower().strip()

reader = csv.DictReader(csv_lines, fieldnames=fieldnames)
    return reader

def filter_columns_in_csv_file(csv_file: BinaryIO, columns_to_keep) -> io.StringIO:
    """Returns a file with specified columns from given CSV file"""
    in_memory_csv_file = io.BytesIO()
    writer = csv.DictWriter(in_memory_csv_file, columns_to_keep)
    reader = get_csv_dict_reader(convert_to_bytes_io(csv_file))
    # We have our field names, burn the header row
    next(reader)

writer.writeheader()
    for row in reader:
        # All columns not in our field names will have a key of None, throw them out or pass if there are none
        try:
            del row[None]
        except KeyError:
            pass
        writer.writerow(row)

in_memory_csv_file.seek(0)
    return in_memory_csv_file

def create_csv_file(content=b'email\nuser@example.com'):
   byte_stream = io.BytesIO()
   byte_stream.write(content)
   byte_stream.seek(0)
   return byte_stream

csv_file = create_csv_file(b'email,pronoun\nuser@example.com,he')
print(filter_columns_in_csv_file(csv_file, [col[0] for col in BULK_IMPORT_COLUMNS]).getvalue())
#+END_SRC

#+begin_src python :results output
import csv

with open('/tmp/in.csv') as fp:
    reader = csv.DictReader(fp, fieldnames=None)
    for row in reader:
        print(row)
#+end_src

#+RESULTS:
: {'email': 'colin@unbl.ink', 'first_name': 'Colin', 'pronoun': 'He'}
: {'email': 'jane@unbl.ink', 'first_name': 'Jane', 'pronoun': 'She'}
: {'email': 'logs@unbl.ink', 'first_name': 'Logs', 'pronoun': 'It'}