1. import json
  2. import logging
  3. import time
  4. from typing import List, Dict, Any
  5. # Configure logging
  6. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  7. def process_json_batch(json_list: List[Dict[str, Any]]):
  8. """
  9. Processes a batch of JSON objects, handling potential errors.
  10. Args:
  11. json_list: A list of JSON objects (dictionaries).
  12. Returns:
  13. A list of results. Each result corresponds to the processing of a JSON object.
  14. Returns an empty list if json_list is empty.
  15. """
  16. if not json_list:
  17. logging.info("Empty JSON list provided. Returning empty result list.")
  18. return []
  19. results = []
  20. for i, json_obj in enumerate(json_list):
  21. try:
  22. # Simulate processing (replace with your actual logic)
  23. start_time = time.time()
  24. result = process_single_json(json_obj, i)
  25. end_time = time.time()
  26. logging.info(f"Processed JSON object {i+1} in {end_time - start_time:.4f} seconds.")
  27. results.append(result)
  28. except Exception as e:
  29. logging.error(f"Error processing JSON object {i+1}: {e}")
  30. results.append({"error": str(e), "index": i}) # store error info
  31. return results
  32. def process_single_json(json_obj: Dict[str, Any], index: int) -> Dict[str, Any]:
  33. """
  34. Processes a single JSON object.
  35. Args:
  36. json_obj: A single JSON object (dictionary).
  37. index: The index of the JSON object in the batch.
  38. Returns:
  39. The result of processing the JSON object.
  40. Raises:
  41. ValueError: If the JSON object is missing a required field.
  42. TypeError: If a value is of the wrong type.
  43. """
  44. try:
  45. # Example processing: Check for required field and type
  46. if "name" not in json_obj:
  47. raise ValueError("Missing 'name' field")
  48. if not isinstance(json_obj["age"], int):
  49. raise TypeError("Age must be an integer")
  50. # Simulate some processing
  51. processed_data = {"name": json_obj["name"].upper(), "age": json_obj["age"] * 2, "processed_index": index}
  52. return processed_data
  53. except (ValueError, TypeError) as e:
  54. raise e # re-raise the exception to be caught in process_json_batch
  55. except Exception as e:
  56. raise e # re-raise other exceptions
  57. if __name__ == '__main__':
  58. # Example usage
  59. json_data = [
  60. {"name": "Alice", "age": 30},
  61. {"name": "Bob", "age": "25"}, # Simulate a type error
  62. {"age": 40}, # Simulate missing required field
  63. {"name": "Charlie", "age": 22}
  64. ]
  65. results = process_json_batch(json_data)
  66. print("\nResults:")
  67. for result in results:
  68. print(result)

Add your comment