1. import json
  2. import logging
  4. class HeaderMetadataWrapper:
  5. def __init__(self, dry_run=False):
  6. self.dry_run = dry_run
  7. self.logger = logging.getLogger(__name__)
  9. def process_header_metadata(self, header_metadata):
  10. """
  11. Processes header metadata, wrapping errors and supporting dry-run mode.
  13. Args:
  14. header_metadata (dict): A dictionary containing header metadata.
  16. Returns:
  17. dict: The processed header metadata. If an error occurs, returns the original metadata and logs the error.
  18. """
  19. try:
  20. # Simulate some processing logic
  21. processed_metadata = self._validate_and_transform(header_metadata)
  23. if self.dry_run:
  24. self.logger.debug(f"Dry run: Processed metadata: {processed_metadata}")
  25. return processed_metadata
  26. else:
  27. self.logger.info(f"Processed metadata: {processed_metadata}")
  28. return processed_metadata
  30. except Exception as e:
  31. self.logger.error(f"Error processing metadata: {e}")
  32. if self.dry_run:
  33. self.logger.debug(f"Dry run: Returning original metadata due to error.")
  34. return header_metadata
  36. def _validate_and_transform(self, header_metadata):
  37. """
  38. Example validation and transformation logic.
  39. """
  40. if not isinstance(header_metadata, dict):
  41. raise TypeError("Header metadata must be a dictionary.")
  42. if 'key1' not in header_metadata:
  43. raise ValueError("Missing required key 'key1'.")
  45. # Simulate transformation
  46. header_metadata['transformed_key'] = header_metadata['key1'].upper()
  47. return header_metadata
  49. if __name__ == '__main__':
  50. # Example usage
  51. logging.basicConfig(level=logging.DEBUG) # Configure logging for debug output
  52. wrapper = HeaderMetadataWrapper(dry_run=True) # Set dry_run to True
  54. metadata1 = {'key1': 'hello'}
  55. processed_metadata1 = wrapper.process_header_metadata(metadata1)
  56. print(f"Processed metadata 1: {processed_metadata1}")
  58. metadata2 = {'key2': 'world'} # Missing key1
  59. processed_metadata2 = wrapper.process_header_metadata(metadata2)
  60. print(f"Processed metadata 2: {processed_metadata2}")
  62. metadata3 = "not a dictionary"
  63. processed_metadata3 = wrapper.process_header_metadata(metadata3)
  64. print(f"Processed metadata 3: {processed_metadata3}")

Add your comment