Data Processing Scenarios
Real-world dictionary transformations often involve complex data processing tasks across various domains like data analysis, configuration management, and API interactions.
JSON Configuration Management
import json
def merge_configurations(base_config, override_config):
def deep_merge(dict1, dict2):
for key, value in dict2.items():
if isinstance(value, dict):
dict1[key] = deep_merge(dict1.get(key, {}), value)
else:
dict1[key] = value
return dict1
return deep_merge(base_config.copy(), override_config)
base_config = {
"database": {
"host": "localhost",
"port": 5432
},
"logging": {
"level": "INFO"
}
}
override_config = {
"database": {
"port": 6432
},
"logging": {
"level": "DEBUG"
}
}
merged_config = merge_configurations(base_config, override_config)
Data Normalization Pipeline
def normalize_user_data(users):
normalized_users = {}
for user in users:
normalized_key = user['email'].lower().replace('.', '_')
normalized_users[normalized_key] = {
"full_name": f"{user['first_name']} {user['last_name']}",
"contact": {
"email": user['email'],
"phone": user.get('phone', 'N/A')
},
"active": user.get('status', 'inactive') == 'active'
}
return normalized_users
raw_users = [
{"first_name": "John", "last_name": "Doe", "email": "[email protected]"},
{"first_name": "Jane", "last_name": "Smith", "email": "[email protected]", "status": "active"}
]
normalized_data = normalize_user_data(raw_users)
def transform_api_response(response):
transformed_data = {
"total_records": len(response['results']),
"records": [
{
"id": record['id'],
"name": record['name'],
"metadata": {
key: value
for key, value in record.items()
if key not in ['id', 'name']
}
}
for record in response['results']
]
}
return transformed_data
api_response = {
"results": [
{"id": 1, "name": "Product A", "price": 100, "category": "Electronics"},
{"id": 2, "name": "Product B", "price": 200, "category": "Clothing"}
]
}
transformed_response = transform_api_response(api_response)
graph TD
A[Raw Data] --> B[Validation]
B --> C[Normalization]
C --> D[Transformation]
D --> E[Processed Data]
Technique |
Use Case |
Complexity |
Generator Expressions |
Large Datasets |
O(1) Memory |
Functional Transformations |
Immutable Data |
Declarative |
Parallel Processing |
CPU-Intensive Tasks |
Scalable |
from functools import reduce
from operator import itemgetter
def complex_data_aggregation(data_list):
return reduce(
lambda acc, item: {
**acc,
item['category']: acc.get(item['category'], 0) + item['value']
},
data_list,
{}
)
transaction_data = [
{"category": "food", "value": 50},
{"category": "transport", "value": 30},
{"category": "food", "value": 25}
]
aggregated_data = complex_data_aggregation(transaction_data)
Error Handling and Validation
def safe_transform(data, transform_func, default=None):
try:
return transform_func(data)
except Exception as e:
print(f"Transformation error: {e}")
return default
Best Practices
- Use immutable transformation approaches
- Implement comprehensive error handling
- Optimize for memory efficiency
- Validate input data before transformation
LabEx recommends adopting these strategies for robust dictionary transformations in production environments.