pages/eng_12811_force_system_to_process_workday_files_with_invalid_columns.org

ID: d788300f-c740-4ad2-ad58-f3e39e278488
ROAM_ALIASES: ENG-12811

ENG-12811 - Force system to process Workday files with invalid columns

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

Sample code

ID: 62a0399f-79be-4dbd-8e45-5bf172c50449

#+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'}