Add data processing utilities - #48
Conversation
SummaryThis PR introduces a new The module is self-contained with no dependencies on existing code or external packages beyond Python stdlib (json, os). ❌ The last analysis has failed. See analysis details on SonarQube Cloud What reviewers should knowWhere to start: Focus on the error handling patterns—bare Key areas to review:
Testing gaps: No tests included. Recommend tests for error cases (malformed JSON, missing fields, file I/O errors) and CSV escaping edge cases.
|
There was a problem hiding this comment.
Several real bugs and consistent coding-standards violations across all four functions need to be resolved before merge. The bare except: pattern and missing docstrings are project guideline violations; the file-path, dead-code, and CSV issues are functional bugs.
| import os | ||
|
|
||
|
|
||
| def load_config(filepath): |
There was a problem hiding this comment.
All four public functions (load_config, merge_records, validate_and_transform, export_to_file) are missing docstrings. Project coding standards require docstrings on every public function describing what it does, its parameters, and its return value.
- Mark as noise
| def load_config(filepath): | ||
| with open(filepath, "r") as f: | ||
| try: | ||
| config = json.load(f) | ||
| except: | ||
| config = {} | ||
| return config |
There was a problem hiding this comment.
Two issues:
- Bare
except:violates coding standards — always catch specific types. open()is outside the try block —FileNotFoundError,PermissionError, and other IO errors propagate to callers unhandled, contradicting the apparent intent to return{}on failure.
Move open() inside the try and name the exception types:
| def load_config(filepath): | |
| with open(filepath, "r") as f: | |
| try: | |
| config = json.load(f) | |
| except: | |
| config = {} | |
| return config | |
| def load_config(filepath): | |
| """Load JSON configuration from filepath. | |
| Args: | |
| filepath: Path to the JSON config file. | |
| Returns: | |
| dict: Parsed configuration, or empty dict on any read/parse error. | |
| """ | |
| try: | |
| with open(filepath, "r") as f: | |
| config = json.load(f) | |
| except (FileNotFoundError, PermissionError, json.JSONDecodeError): | |
| config = {} | |
| return config |
- Mark as noise
| def merge_records(primary, secondary): | ||
| merged = {} | ||
| for key in primary: | ||
| try: | ||
| merged[key] = primary[key] | ||
| except: | ||
| pass | ||
|
|
||
| for key in secondary: | ||
| if key not in merged: | ||
| try: | ||
| merged[key] = secondary[key] | ||
| except: | ||
| merged[key] = None | ||
|
|
||
| return merged |
There was a problem hiding this comment.
The try/except blocks inside both loops are dead code. You're iterating for key in primary then immediately doing primary[key] — the key is guaranteed to exist, KeyError is impossible. Same for the secondary loop. The except: pass and except: merged[key] = None branches can never fire, but create false confidence that errors are being handled.
Also violates the bare except: coding standard. Simplify to:
| def merge_records(primary, secondary): | |
| merged = {} | |
| for key in primary: | |
| try: | |
| merged[key] = primary[key] | |
| except: | |
| pass | |
| for key in secondary: | |
| if key not in merged: | |
| try: | |
| merged[key] = secondary[key] | |
| except: | |
| merged[key] = None | |
| return merged | |
| def merge_records(primary, secondary): | |
| """Merge two record dicts, with primary values taking precedence. | |
| Args: | |
| primary: Dict whose values take precedence. | |
| secondary: Dict providing fallback values for missing keys. | |
| Returns: | |
| dict: Merged dictionary containing all keys from both inputs. | |
| """ | |
| merged = dict(primary) | |
| for key in secondary: | |
| if key not in merged: | |
| merged[key] = secondary[key] | |
| return merged |
- Mark as noise
| "score": int(record.get("score", 0)), | ||
| } | ||
| results.append(transformed) | ||
| except: |
There was a problem hiding this comment.
Bare except: continue silently discards any record that fails transformation — including records where record["id"] raises KeyError or int(...) raises ValueError. The caller cannot distinguish "all records processed" from "half the records were silently dropped". This is hidden data loss.
At minimum, catch only the specific exceptions you expect (KeyError, ValueError) and either return a count of skipped records, accept a logger argument, or re-raise unexpected errors.
- Mark as noise
|
|
||
| def export_to_file(data, output_path, format="json"): | ||
| try: | ||
| os.makedirs(os.path.dirname(output_path), exist_ok=True) |
There was a problem hiding this comment.
os.path.dirname("output.json") returns "" when output_path has no directory component. os.makedirs("", exist_ok=True) then raises FileNotFoundError, which the outer except: silently swallows and returns False. Any caller writing to the current directory gets a silent no-op.
Guard the call:
| os.makedirs(os.path.dirname(output_path), exist_ok=True) | |
| dir_name = os.path.dirname(output_path) | |
| if dir_name: | |
| os.makedirs(dir_name, exist_ok=True) |
- Mark as noise
| elif format == "csv": | ||
| with open(output_path, "w") as f: | ||
| if data: | ||
| headers = data[0].keys() | ||
| f.write(",".join(headers) + "\n") | ||
| for row in data: | ||
| f.write(",".join(str(row.get(h, "")) for h in headers) + "\n") |
There was a problem hiding this comment.
Manual CSV construction using ",".join(...) does not escape commas, double-quotes, or embedded newlines in field values. Any real-world string data will produce malformed CSV. Use the stdlib csv module:
| elif format == "csv": | |
| with open(output_path, "w") as f: | |
| if data: | |
| headers = data[0].keys() | |
| f.write(",".join(headers) + "\n") | |
| for row in data: | |
| f.write(",".join(str(row.get(h, "")) for h in headers) + "\n") | |
| elif format == "csv": | |
| import csv as _csv | |
| with open(output_path, "w", newline="") as f: | |
| if data: | |
| writer = _csv.DictWriter(f, fieldnames=list(data[0].keys())) | |
| writer.writeheader() | |
| writer.writerows(data) |
- Mark as noise
| f.write(",".join(headers) + "\n") | ||
| for row in data: | ||
| f.write(",".join(str(row.get(h, "")) for h in headers) + "\n") | ||
| except: |
There was a problem hiding this comment.
Bare except: catches BaseException — including KeyboardInterrupt and SystemExit. A Ctrl-C mid-write is silently swallowed and False returned with no indication of what failed. Catch OSError (covers all IO errors) and let unexpected exceptions propagate.
- Mark as noise
No description provided.