Background
I receive dataframes from several sources (below I call these sources s1
, s2
and s3
). I first want to check whether the provided data meets certain requirements and if so, the data will be further processed; below I only provide the code for the check whether the provided data is correct.
Approach
For each column in each dataframe I create an own function that returns a Boolean variable depending on the evaluation of a regular expression. The regular expression depends on the data source and the column name.
Input - output
An example could look like this (coming from source s1
):
ID content
x fo_stuff.s1orig_0987 ABBBCCCCBB
y cool_abc.s1orig CCBBAABBTY
z er_something.illegal.foo BABCACCB
In ID
the regex evaluates to True
if an element consists of
(2 letters)_(arbitrary number of letters/numbers/underscores).(s1orig)(optional: underscore and four digits)
So, in this case only the entry in x
is correct.
In content
the regular expression evaluates to True
if the string consists of either only A
, B
, C
or a
, b
, c
(arbitrary many), so x
and z
are correct, y
is not as it contains a T
and a Y
.
If I run the code from below, I then get my expected outcome:
test = ProcessDataTables(df, "s1")
test.get_invalid_entries()
print(test.invalid_entries)
# {'ID': ['y', 'z'], 'content': ['y']}
So, I get a dictionary that has as keys the column headers and as values a list of indices that correspond to invalid data.
Questions
Is there a better way of organizing the code? Now it kind of violates the DRY principle. On the other hand, having individual function for each data source and column is quite flexible if regular expressions have to change/additional columns need to be considered.
Is there are a better way of applying the
is_valid_***
functions to the dataframe columns? Using__get_mapping_dict
works but looks rather strange to pass functions around like this.Is there a straightforward way to have all the
is_valid_***
functions inside the class?
This is the entire code:
import re
import pandas as pd
class ProcessDataTables(object):
def __init__(self, data_table, data_source):
self.data = data_table
if data_source in {'s1', 's2', 's3'}:
self.source = data_source
else:
raise ValueError(f"{data_source} is not a valid data source.")
self._mapping_dict = self.__get_mapping_dict()
self.invalid_entries = None
if not set(self._mapping_dict).issubset(self.data.columns):
raise ValueError("Please check your columns names!")
def get_invalid_entries(self):
# apply the passed functions to the respective columns and filter all
# values that are False (meaning that they have a wrong format)
temp = {ci: self.data.index[~self.data[ci].apply(fi)].tolist()
for ci, fi in self._mapping_dict.items()}
self.invalid_entries = temp
def __get_mapping_dict(self):
if self.source == "s1":
return {'ID': is_valid_name_s1,
'content': is_valid_sequence}
elif self.source == "s2":
return {'ID': is_valid_name_s2,
'content': is_valid_sequence}
elif self.source == "s3":
return {'ID': is_valid_name_s3,
'content': is_valid_sequence}
def is_valid_name_s1(name):
"""
This function checks whether *name* matches the following pattern:
(2letters)_(arbitrary number of letters/numbers/underscores).
(s1orig)(optional: underscore and for digits)
*name*: a string
return: True if *name matches the pattern, False if not
"""
regex_name = re.compile(r'^[A-Za-z]{2}_([A-Za-z0-9_]+)\.(s1orig)(_\d{4})?$')
if regex_name.match(name) is not None:
return True
return False
def is_valid_name_s2(name):
"""
This function checks whether *name* matches the following pattern:
(2 or 4 letters)_(arbitrary number of letters/numbers/underscores).
(s2orig)(optional: underscore and for digits)
*name*: a string
return: True if *name matches the pattern, False if not
"""
regex_name = re.compile(r'^([A-Za-z]{2}|[A-Za-z]{4})_([A-Za-z0-9_]+)\.(s2orig)(_\d{4})?$')
if regex_name.match(name) is not None:
return True
return False
def is_valid_name_s3(name):
"""
This function checks whether *name* matches the following pattern:
(3 letters)_(arbitrary number of letters/numbers/underscores).
(s3orig)(optional: underscore and for digits)
*name*: a string
return: True if *name matches the pattern, False if not
"""
regex_name = re.compile(r'^[A-Za-z]{3}_([A-Za-z0-9_]+)\.(s3orig)(_\d{4})?$')
if regex_name.match(name) is not None:
return True
return False
def is_valid_sequence(sequence):
"""
True if string consists of either only A, B, C or a, b, c
"""
regex_seq = re.compile(r'^([ABC]+|[abc]+)$')
if regex_seq.match(sequence) is not None:
return True
return False
if __name__ == '__main__':
df = pd.DataFrame({'ID': ["fo_stuff.s1orig_0987",
"cool_abc.s1orig",
"er_something.illegal.foo"],
'content': ['ABBBCCCCBB',
'CCBBAABBTY',
'BABCACCB']},
index=list('xyz'))
test = ProcessDataTables(df, "s1")
test.get_invalid_entries()
print(test.invalid_entries)