diff options
Diffstat (limited to 'obfuscator')
| -rw-r--r-- | obfuscator/csv_reader.py | 19 | ||||
| -rw-r--r-- | obfuscator/csv_writer.py | 1 | ||||
| -rw-r--r-- | obfuscator/obfuscate.py | 5 | ||||
| -rw-r--r-- | obfuscator/utils.py | 1 |
4 files changed, 0 insertions, 26 deletions
diff --git a/obfuscator/csv_reader.py b/obfuscator/csv_reader.py index f8dd7d3..2b099c8 100644 --- a/obfuscator/csv_reader.py +++ b/obfuscator/csv_reader.py @@ -6,12 +6,6 @@ from typing import List, Dict from obfuscator.logger import get_logger from obfuscator.utils import Utilities -# Putting the CSV reading components into a class may seem like overkill -# for a simple script, but it allows for better organization and scalability. -# @staticmethod is used to define the method without an instance of the class -# being required. The methods could be defined just as functions, and this -# may still be changed. - class CSVReader: """ @@ -21,7 +15,6 @@ class CSVReader: def __init__(self, log_level=None): self.log_level = log_level - # Create the logger self.logger = get_logger("CSVREADER", log_level) def read_local(self, path) -> List[Dict[str, str]]: @@ -29,12 +22,8 @@ class CSVReader: A method to read a local CSV file and return the data as a list of dictionaries. """ - # Log the path of the file being read for debugging self.logger.debug(f"Reading local CSV from: {path}") - # Attempt to read the file and return the data as a list of dictionaries - # However, if the file isn't found or there is a generic exception, log - # the error and raise an exception try: with open(path, mode="r", encoding="utf-8") as f: reader = csv.DictReader(f) @@ -54,7 +43,6 @@ class CSVReader: bucket, key = utils.get_s3_path(path) self.logger.debug(f"Reading S3 CSV from: {bucket}/{key}") - # If LOCALSTACK=TRUE, use the localstack endpoint for testing if os.getenv("LOCALSTACK", "FALSE").upper() == "TRUE": localstack_endpoint = "http://localhost.localstack.cloud:4566" self.logger.debug("Using LocalStack endpoint for S3") @@ -69,15 +57,10 @@ class CSVReader: client = boto3.client("s3") try: - # Attempt to read the S3 object and return the data as a list of dictionaries response = client.get_object(Bucket=bucket, Key=key) self.logger.info("S3 object read successfully") - # Read and decode the content content = response["Body"].read().decode("utf-8") - # Even though the read_string method was only created for testing, - # it can be reused here to read and return the CSV data return CSVReader.read_string(content) - # TODO: Add more specific exceptions to catch except Exception as e: self.logger.error(f"Error reading S3 object: {e}") raise @@ -87,11 +70,9 @@ class CSVReader: A method to read CSV data from a string and return the data as a list of dictionaries. """ - # If the content is empty, return an empty list if not content.strip(): return [] - # Treat the string as a file-like object and return as list of dictionaries f = io.StringIO(content) reader = csv.DictReader(f) return [dict(row) for row in reader] diff --git a/obfuscator/csv_writer.py b/obfuscator/csv_writer.py index 099e910..56b3f1f 100644 --- a/obfuscator/csv_writer.py +++ b/obfuscator/csv_writer.py @@ -3,7 +3,6 @@ import io from typing import List, Dict from obfuscator.logger import get_logger -# Create the logger logger = get_logger("CSVWRITER") diff --git a/obfuscator/obfuscate.py b/obfuscator/obfuscate.py index 4f7e6c1..e964433 100644 --- a/obfuscator/obfuscate.py +++ b/obfuscator/obfuscate.py @@ -1,7 +1,6 @@ from typing import List, Dict from obfuscator.logger import get_logger -# Create the logger logger = get_logger("OBFUSCATE") @@ -12,14 +11,10 @@ def obfuscate( A function to obfuscate PII fields in a list of dictionaries, replacing sensitive values with a string of asterisks. """ - # If no data is provided, log a message and return an empty list if not data: logger.info("No valid data was provided to obfuscate") return [] - # Obfuscate the PII fields in each record using a list/dict comprehension - # This code is good but makes debugging a bit tricky. I may consider - # breaking it down into a for loop. return [ {k: ("***" if k in pii_fields else v) for k, v in record.items()} for record in data diff --git a/obfuscator/utils.py b/obfuscator/utils.py index 81eb04a..f61451b 100644 --- a/obfuscator/utils.py +++ b/obfuscator/utils.py @@ -5,7 +5,6 @@ from obfuscator.logger import get_logger class Utilities: def __init__(self, logger=None): - # Create the logger self.logger = get_logger("UTILITIES", logger) def get_s3_path(self, uri): |