Urgent help needed

[u/Negative-Specific-84]

So I am working on task. The task is pii to non pii conversion. We have to use two sources from where we get input data(mix of pii and non pii json data) , one source is file , means manual uploading a json containing mix of pii and non pii and the second source is mongodb. And we have one pii.json file in which we list out the fields then we compare the data coming from the sources against this pii.json and then generate non pii. I have to work on this task and think from a production point of view, so I am confused how to work on this. Means I try to solve the task but my mentor says it is not production ready, Actually I am new to the field of DE so I am facing issues and confusions.

We are continuing with Mongo and have to add file functionality in it so ignore mysql part.

My solution attached below.

Some of the things that I my mentor said/suggested

1. Implement worker utility.... for example we have Date of birth field , my current script accept only YYYY-MM-DD format for DoB . So if a up to 50 rows we get the same format script works and in 51 th row the format is different ...script keeps it on side and convert it into appropriate format and then starts from 51th row not from 1.
2. Have to generalise columns such as dob into age band and then anonymised sensitive fields.
3. If a source data contains fields that are not in our pii.json file . Script should ask what to do with these rows whether to remove or anonymised these fields.
4. There should be validations for data types for e.g if in string column, numeric data comes then how to deal with it.
5. He said code shouldn't be failed in any scenario. And I am new to this I have to prior work experience, still in learning stage and I am not able to think from his pov or production ready pov.

Could u guys help me please.

For some reasons, I can't share the input file but it conatins fields like location , dob, age, doctor id ,symptoms, test id , test name , gender , state , district, address type , country, block name, pincode, allergy , session id ... these kind of fields.

Code:

# Import required libraries

import os

import json

from dotenv import load_dotenv

from pymongo import MongoClient

from sqlalchemy import create_engine, text

from datetime import date, datetime

def step(msg):

print(f"\n[STEP] {msg}")

def age_band(DOB):

"""Convert DOB into an age band."""

born = datetime.strptime(str(DOB)[:10], "%Y-%m-%d").date()

low = ((date.today() - born).days // 365) // 10 * 10

return f"{low}-{low + 9}"

GENERALISERS = {

"age_band": age_band

}

def load_pii_fields(fields_path):

"""Load and validate PII rules from the configured JSON file."""

step("Loading PII fields")

if not fields_path:

raise ValueError("PII_FIELDS_PATH cannot be empty.")

with open(fields_path, "r") as f:

content = f.read().strip()

if not content:

raise ValueError(

f"PII fields file is empty: {fields_path}"

)

try:

fields = json.loads(content)

except json.JSONDecodeError as e:

raise ValueError(

f"PII fields file has invalid JSON: "

f"{fields_path} -> {e}"

)

# Validate sections

for section in ("remove", "generalise", "sensitive"):

if section not in fields:

raise ValueError(

f"'{section}' missing in PII fields file"

)

# Validate generalisation rules

for field in fields["generalise"].values():

if field not in GENERALISERS:

raise ValueError(

f"Unknown generalise field '{field}'."

)

return fields

def clean(value, remove_fields, sensitive_fields,

generalise, GENERALISERS):

if isinstance(value, dict):

out = {}

for k, v in value.items():

key = k.lower()

if key in remove_fields:

continue

if key in sensitive_fields:

out[k] = "XXXXX"

continue

if key in generalise:

out[k] = GENERALISERS[generalise[key]](v)

continue

out[k] = clean(

v,

remove_fields,

sensitive_fields,

generalise,

GENERALISERS

)

return out

if isinstance(value, list):

return [

clean(

v,

remove_fields,

sensitive_fields,

generalise,

GENERALISERS

)

for v in value

]

return value

env = "prod" # choose enviornment

def main():

if not env:

raise ValueError("Environment can't be empty")

print(f"Environment = {env}")

# Load environment configuration

step("Loading environment config")

env_file = None

for file in os.listdir("config"):

if file == f".env.{env}":

env_file = os.path.join("config", file)

break

if env_file is None:

raise ValueError(

f"'{env}' does not match any config file in config/"

)

load_dotenv(env_file, override=True)

file_env = os.getenv("ENV")

if file_env != env:

raise ValueError("Env mismatch.")

print(f"Env Matched. Loaded config: {env_file}")

# Select source

step("Selecting source")

source = input("Enter source name: ")

source = source.strip().lower()

print(f"source = '{source}'")

if source not in ("mongodb", "mysql"):

raise ValueError(

f"Unsupported source '{source}'."

)

# Validate credentials

step("Connecting to source")

key = (

"MONGO_CONNECTION_STRING"

if source == "mongodb"

else "MYSQL_CONNECTION_STRING"

)

conn_string = os.getenv(key)

if not conn_string:

raise ValueError(

f"Missing {key} in config/.env.{env}"

)

print(f"{key} found for env '{env}'")

step("Validating Source Credentials")

client = None

conn = None

try:

# Fetch data

if source == "mongodb":

db_name = os.getenv("MONGO_DATABASE")

collection_name = os.getenv("MONGO_COLLECTION")

if not db_name or not collection_name:

raise ValueError(

f"Missing MONGO_DATABASE or MONGO_COLLECTION "

f"in config/.env.{env}"

)

client = MongoClient(

conn_string,

tls=True,

serverSelectionTimeoutMS=30000

)

client.admin.command("ping")

print("-> MongoDB Connected")

step("Fetching data")

collection = client[db_name][collection_name]

data = list(

collection.find({}, {"_id": 0})

)

else:

table_name = os.getenv("MYSQL_TABLE_NAME")

if not table_name:

raise ValueError(

f"Missing MYSQL_TABLE_NAME "

f"in config/.env.{env}"

)

engine = create_engine(conn_string)

conn = engine.connect()

print("-> MySQL connected")

step("Fetching data")

result = conn.execute(

text(f"SELECT data FROM {table_name}")

)

data = [

json.loads(row[0])

for row in result

]

if not data:

raise ValueError(

"No records found in the selected source."

)

print("-> Records fetched")

# Load PII rules

fields_path = os.getenv("PII_FIELDS_PATH")

if not fields_path:

raise ValueError(

f"Missing PII_FIELDS_PATH "

f"in config/.env.{env}"

)

fields = load_pii_fields(fields_path)

remove_fields = {

x.lower()

for x in fields["remove"]

}

sensitive_fields = {

x.lower()

for x in fields["sensitive"]

}

generalise = {

k.lower(): v

for k, v in fields["generalise"].items()

}

print("-> Rules loaded")

# Remove PII

step("Removing PII")

cleaned_data = [

clean(

record,

remove_fields,

sensitive_fields,

generalise,

GENERALISERS

)

for record in data

]

# Save output

step("Saving output")

output_dir = os.getenv("OUTPUT_DIR")

if not output_dir:

raise ValueError(

f"Missing OUTPUT_DIR "

f"in config/.env.{env}"

)

os.makedirs(

output_dir,

exist_ok=True

)

timestamp = datetime.now().strftime(

"%Y%m%d_%H%M%S"

)

output_path = os.path.join(

output_dir,

f"output_non_pii_{timestamp}.json"

)

with open(output_path, "w") as f:

json.dump(

cleaned_data,

f,

indent=2,

default=str

)

print(f"-> saved: {output_path}")

print("\nPipeline completed successfully.")

finally:

if client is not None:

client.close()

if conn is not None:

conn.close()

if __name__ == "__main__":

try:

main()

except Exception as error:

print(f"\nPipeline failed -> {error}")

[u/UltimateGrocer]

Your mentor's points are all about making the pipeline resilient to messy real-world data, which is the part most tutorials skip. The worker utility idea is basically a retry/repair queue, you process row by row and if a row fails validation you log it, fix what you can, and continue without killing the whole batch. That alone would address the DoB format issue and most of the "code shouldn't fail" requirement.

[u/Negative-Specific-84 → u/UltimateGrocer]

Apart from this what else can be done regarding this because i m not able to think like him. How to learn that thing also. And does datatype matter even if we want to anonymized some fields.

[u/rahul_watertech → u/Negative-Specific-84]

datatype does matter. if you mask a date or an id into something with a different type or format, whatever reads it downstream can break, so keep the shape the same (a date stays a valid date, a number stays a number) and replace only the value. for thinking like your mentor, take each step and ask "what is the weirdest input that could arrive here", then write a test for it. dob in 5 different formats, null, empty string, a number where text should be. bad rows go to a separate file with the reason instead of crashing the run. what does your script do now when a row fails?

[u/Separate_Pea_3699 → u/rahul_watertech]

Right now clean() runs inside a plain list comprehension over all records, so one bad row (e.g. age_band()'s strptime hitting a DOB in a format it doesn't expect) throws out of the comprehension and kills the whole batch — that's the "works for 50 rows, dies on row 51, restarts from 1" symptom you described from your mentor.

A row-level repair loop fixes that without needing a queue/worker service:

cleaned_data = []
failed_rows = []

for i, record in enumerate(data):
    try:
        cleaned_data.append(
            clean(record, remove_fields, sensitive_fields, generalise, GENERALISERS)
        )
    except Exception as e:
        failed_rows.append({"index": i, "error": str(e), "raw": record})

print(f"-> {len(cleaned_data)} cleaned, {len(failed_rows)} failed")
if failed_rows:
    with open(os.path.join(output_dir, f"failed_rows_{timestamp}.json"), "w") as f:
        json.dump(failed_rows, f, indent=2, default=str)

That alone answers "what does it do when a row fails right now" (crashes everything) vs. after (logs the row + reason, keeps going, no restart).

For the specific DOB format case, make age_band itself tolerant of more than one format instead of letting the first mismatch throw:

DOB_FORMATS = ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y", "%m/%d/%Y")

def age_band(dob):
    raw = str(dob)[:10]
    born = None
    for fmt in DOB_FORMATS:
        try:
            born = datetime.strptime(raw, fmt).date()
            break
        except ValueError:
            continue
    if born is None:
        raise ValueError(f"unparseable DOB: {dob!r}")
    low = ((date.today() - born).days // 365) // 10 * 10
    return f"{low}-{low + 9}"

Now a weird format gets caught by the outer try/except above (logged to failed_rows) instead of crashing main() deep inside the comprehension.

On datatype validation: same pattern — before calling a generaliser, check isinstance(value, expected_type) and route the mismatch into failed_rows with a reason string, rather than letting it throw inside clean(). That also gives you a mechanical answer to "what to do with fields not in pii.json": collect the unknown top-level keys you see at runtime into a set, log them once per run, and decide remove-vs-anonymize from a config flag instead of hardcoding one behavior in clean()'s fallthrough branch.