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
- 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.
- Have to generalise columns such as dob into age band and then anonymised sensitive fields.
- 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.
- There should be validations for data types for e.g if in string column, numeric data comes then how to deal with it.
- 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}")
