1. import json
  3. def batch_json_dry_run(json_responses, operations):
  4. """
  5. Performs batch operations on a list of JSON responses for dry-run scenarios.
  7. Args:
  8. json_responses (list): A list of JSON strings or dictionaries.
  9. operations (list): A list of operations to perform on each JSON response.
  10. Each operation is a dictionary with 'type' and 'value' keys.
  11. 'type' can be 'replace', 'add', or 'remove'.
  12. 'value' contains the operation's details.
  14. Returns:
  15. list: A list of modified JSON responses (strings or dictionaries).
  16. """
  17. modified_responses = []
  18. for i, response in enumerate(json_responses):
  19. try:
  20. # Load JSON if it's a string
  21. if isinstance(response, str):
  22. data = json.loads(response)
  23. else:
  24. data = response # Assume it's already a dictionary
  26. for operation in operations:
  27. operation_type = operation['type']
  28. if operation_type == 'replace':
  29. # Replace a key's value
  30. key = operation['value']['key']
  31. new_value = operation['value']['value']
  32. if key in data:
  33. data[key] = new_value
  34. else:
  35. print(f"Warning: Key '{key}' not found in response {i+1}")
  36. elif operation_type == 'add':
  37. # Add a new key-value pair
  38. key = operation['value']['key']
  39. value = operation['value']['value']
  40. data[key] = value
  41. elif operation_type == 'remove':
  42. # Remove a key
  43. key = operation['value']['key']
  44. if key in data:
  45. del data[key]
  46. else:
  47. print(f"Warning: Key '{key}' not found in response {i+1}")
  48. else:
  49. print(f"Warning: Unknown operation type '{operation_type}'")
  51. # Convert back to JSON string if the original was a string
  52. if isinstance(response, str):
  53. modified_responses.append(json.dumps(data, indent=4)) # indent for readability
  54. else:
  55. modified_responses.append(data)
  57. except json.JSONDecodeError:
  58. print(f"Error decoding JSON in response {i+1}")
  59. modified_responses.append(response) #Return original if decode fails.
  60. except Exception as e:
  61. print(f"An unexpected error occurred while processing response {i+1}: {e}")
  62. modified_responses.append(response) #Return original if any other error.
  64. return modified_responses
  67. if __name__ == '__main__':
  68. # Example usage
  69. json_responses = [
  70. '{"name": "Alice", "age": 30}',
  71. {"city": "New York", "population": 8400000}
  72. ]
  74. operations = [
  75. {'type': 'replace', 'value': {'key': 'age', 'value': 31}},
  76. {'type': 'add', 'value': {'key': 'occupation', 'value': 'Engineer'}},
  77. {'type': 'remove', 'value': {'key': 'population'}} #remove population from the second response
  78. ]
  80. modified_responses = batch_json_dry_run(json_responses, operations)
  81. for i, response in enumerate(modified_responses):
  82. print(f"Modified Response {i+1}:\n{response}\n")

Add your comment