xcc.es
[READ-ONLY] Mirror of https://github.com/mrgnw/scripts.
1#!/usr/bin/env -S uv run --script
2
3# /// script
4# requires-python = ">=3.13"
5# dependencies = [
6# "boto3",
7# ]
8# ///
9"""
10DynamoDB utilities for converting AWS DynamoDB JSON format to simple Python dictionaries.
11
12Clean, Pythonic approach using recursion and type converters.
13"""
14
15from decimal import Decimal
16from typing import Any, Dict, List, Union
17
18
19def deserialize_dynamodb_value(value: Dict[str, Any]) -> Any:
20 """
21 Convert a DynamoDB value to Python value using clean recursive approach.
22
23 Args:
24 value: DynamoDB value like {"S": "hello"}, {"N": "42"}, {"L": [...]}
25
26 Returns:
27 Native Python value
28 """
29 if not isinstance(value, dict) or len(value) != 1:
30 return value
31
32 type_key, type_value = next(iter(value.items()))
33
34 # Type converter mapping - much cleaner than if/elif chain
35 converters = {
36 "S": str, # String
37 "N": lambda x: int(x) if "." not in x else float(x), # Number (smart int/float)
38 "B": bytes, # Binary
39 "BOOL": bool, # Boolean
40 "NULL": lambda x: None, # Null
41 "SS": list, # String Set
42 "NS": lambda x: [int(n) if "." not in n else float(n) for n in x], # Number Set
43 "BS": list, # Binary Set
44 "L": lambda x: [deserialize_dynamodb_value(item) for item in x], # List
45 "M": lambda x: {k: deserialize_dynamodb_value(v) for k, v in x.items()}, # Map
46 }
47
48 converter = converters.get(type_key, lambda x: x)
49 return converter(type_value)
50
51
52def deserialize_dynamodb_item(item: Dict[str, Any]) -> Dict[str, Any]:
53 """
54 Convert a DynamoDB item to a clean Python dict.
55
56 Args:
57 item: DynamoDB item with typed attributes
58
59 Returns:
60 Clean Python dict with native values
61 """
62 return {key: deserialize_dynamodb_value(value) for key, value in item.items()}
63
64
65def deserialize_dynamodb_response(response: Dict[str, Any]) -> List[Dict[str, Any]]:
66 """
67 Convert a DynamoDB query/scan response to clean Python list of dicts.
68
69 Args:
70 response: DynamoDB response like {"Items": [{"id": {"S": "value"}}], "Count": 1}
71
72 Returns:
73 List of clean Python dicts like [{"id": "value"}]
74 """
75 items = response.get("Items", [])
76 return [deserialize_dynamodb_item(item) for item in items]
77
78
79def deserialize_dynamodb_json(json_data: Union[str, Dict[str, Any]]) -> List[Dict[str, Any]]:
80 """
81 Convert DynamoDB JSON (string or dict) to clean Python list of dicts.
82
83 Args:
84 json_data: JSON string or dict containing DynamoDB response
85
86 Returns:
87 List of clean Python dicts
88 """
89 import json
90
91 if isinstance(json_data, str):
92 try:
93 response = json.loads(json_data)
94 except json.JSONDecodeError as e:
95 raise ValueError(f"Invalid JSON string: {e}")
96 else:
97 response = json_data
98
99 return deserialize_dynamodb_response(response)
100
101
102# Alternative: Use boto3's built-in deserializer (if boto3 is available)
103def deserialize_with_boto3(response: Dict[str, Any]) -> List[Dict[str, Any]]:
104 """
105 Use boto3's built-in TypeDeserializer for the most accurate conversion.
106 This is the "official" way but requires boto3 dependency.
107 """
108 try:
109 from boto3.dynamodb.types import TypeDeserializer
110
111 deserializer = TypeDeserializer()
112 items = response.get("Items", [])
113 return [{k: deserializer.deserialize(v) for k, v in item.items()} for item in items]
114
115 except ImportError:
116 # Fallback to our custom implementation
117 return deserialize_dynamodb_response(response)
118
119
120def main():
121 """CLI interface for converting DynamoDB JSON files to clean format."""
122 import argparse
123 import json
124 import sys
125 from pathlib import Path
126
127 parser = argparse.ArgumentParser(description="Convert DynamoDB JSON format to clean Python dictionaries")
128 parser.add_argument("file", help="Path to DynamoDB JSON file to convert")
129 parser.add_argument(
130 "--use-boto3",
131 action="store_true",
132 help="Use boto3's TypeDeserializer instead of custom implementation",
133 )
134
135 args = parser.parse_args()
136
137 # Check if file exists
138 file_path = Path(args.file)
139 if not file_path.exists():
140 print(f"Error: File '{args.file}' not found", file=sys.stderr)
141 sys.exit(1)
142
143 try:
144 # Load the DynamoDB JSON file
145 with open(file_path, "r") as f:
146 raw_data = f.read()
147
148 # Convert to clean format
149 if args.use_boto3:
150 # Parse JSON first, then use boto3
151 response = json.loads(raw_data)
152 clean_data = deserialize_with_boto3(response)
153 else:
154 # Use our custom implementation
155 clean_data = deserialize_dynamodb_json(raw_data)
156
157 # Output clean JSON
158 print(json.dumps(clean_data, indent=2, default=str))
159
160 except json.JSONDecodeError as e:
161 print(f"Error: Invalid JSON in file '{args.file}': {e}", file=sys.stderr)
162 sys.exit(1)
163 except Exception as e:
164 print(f"Error processing file '{args.file}': {e}", file=sys.stderr)
165 sys.exit(1)
166
167
168if __name__ == "__main__":
169 main()