xcc.es
[READ-ONLY] Mirror of https://github.com/mrgnw/scripts.
1#!/usr/bin/env -S uv run --script
2# /// script
3# requires-python = ">=3.12"
4# dependencies = [
5# "boto3",
6# "duckdb"
7# ]
8# ///
9
10import sys
11import boto3
12import duckdb
13from typing import Dict, Any, List
14import argparse
15
16
17def scan_dynamodb_table(table_name: str, region: str = None) -> List[Dict[str, Any]]:
18 """
19 Scan entire DynamoDB table and return all items.
20 """
21 # Create DynamoDB client
22 if region:
23 dynamodb = boto3.client("dynamodb", region_name=region)
24 else:
25 dynamodb = boto3.client("dynamodb")
26
27 print(f"Scanning DynamoDB table: {table_name}")
28
29 items = []
30 response = dynamodb.scan(TableName=table_name)
31 items.extend(response.get("Items", []))
32
33 # Handle pagination
34 while "LastEvaluatedKey" in response:
35 print(f"Retrieved {len(items)} items so far...")
36 response = dynamodb.scan(TableName=table_name, ExclusiveStartKey=response["LastEvaluatedKey"])
37 items.extend(response.get("Items", []))
38
39 print(f"Total items retrieved: {len(items)}")
40 return items
41
42
43def convert_dynamodb_to_dict(dynamodb_item: Dict[str, Any]) -> Dict[str, Any]:
44 """
45 Convert DynamoDB item format to regular dictionary using modern Python patterns.
46
47 DynamoDB format: {'attr': {'S': 'value'}} or {'attr': {'N': '123'}}
48 Regular format: {'attr': 'value'} or {'attr': 123}
49 """
50
51 def convert_value(value: Any) -> Any:
52 """Convert a single DynamoDB value to Python native type."""
53 if not isinstance(value, dict) or len(value) != 1:
54 return value
55
56 type_key, actual_value = next(iter(value.items()))
57
58 match type_key:
59 case "S": # String
60 return actual_value
61 case "N": # Number
62 return int(actual_value) if "." not in actual_value else float(actual_value)
63 case "B": # Binary
64 return str(actual_value)
65 case "BOOL": # Boolean
66 return actual_value
67 case "NULL": # Null
68 return None
69 case "L": # List
70 return [convert_value(item) for item in actual_value]
71 case "M": # Map
72 return {k: convert_value(v) for k, v in actual_value.items()}
73 case "SS" | "NS" | "BS": # Sets (String Set, Number Set, Binary Set)
74 if type_key == "NS":
75 return [float(n) for n in actual_value]
76 elif type_key == "BS":
77 return [str(b) for b in actual_value]
78 else:
79 return actual_value
80 case _: # Unknown type
81 return actual_value
82
83 return {key: convert_value(value) for key, value in dynamodb_item.items()}
84
85
86def save_to_csv_with_duckdb(items: List[Dict[str, Any]], output_file: str):
87 """
88 Use DuckDB to process data and save to CSV with modern Python patterns.
89 """
90 if not items:
91 print("No items to save.")
92 return
93
94 # Convert DynamoDB items to regular dictionaries
95 converted_items = [convert_dynamodb_to_dict(item) for item in items]
96
97 # Create DuckDB connection
98 conn = duckdb.connect()
99
100 print(f"Processing {len(converted_items)} items...")
101
102 try:
103 # Get all unique keys and create schema
104 all_keys = sorted(set().union(*(item.keys() for item in converted_items)))
105
106 # Create table with proper column names (escaped)
107 columns_def = ", ".join(f'"{key}" VARCHAR' for key in all_keys)
108 conn.execute(f"CREATE TABLE dynamodb_data ({columns_def})")
109
110 # Prepare data for batch insert
111 rows = []
112 for item in converted_items:
113 row = []
114 for key in all_keys:
115 value = item.get(key)
116 # Convert complex types to JSON strings for CSV compatibility
117 if isinstance(value, (list, dict)):
118 import json
119
120 row.append(json.dumps(value, default=str))
121 else:
122 row.append(value)
123 rows.append(row)
124
125 # Batch insert all data
126 placeholders = ", ".join("?" for _ in all_keys)
127 conn.executemany(f"INSERT INTO dynamodb_data VALUES ({placeholders})", rows)
128
129 # Export to CSV
130 conn.execute(f"COPY dynamodb_data TO '{output_file}' (FORMAT CSV, HEADER)")
131
132 # Show results
133 row_count = conn.execute("SELECT COUNT(*) FROM dynamodb_data").fetchone()[0]
134 print(f"Columns: {all_keys}")
135 print(f"Data exported to: {output_file} ({row_count} rows)")
136
137 # Show sample data
138 print("\nSample data (first 3 rows):")
139 sample_rows = conn.execute("SELECT * FROM dynamodb_data LIMIT 3").fetchall()
140
141 for i, row in enumerate(sample_rows, 1):
142 print(f"Row {i}: {dict(zip(all_keys, row))}")
143
144 finally:
145 conn.close()
146
147
148def main():
149 parser = argparse.ArgumentParser(
150 description="Download DynamoDB table to CSV using DuckDB",
151 formatter_class=argparse.RawDescriptionHelpFormatter,
152 epilog="""
153Examples:
154 %(prog)s my-table
155 %(prog)s my-table --output my-data.csv
156 %(prog)s my-table --region us-west-2
157 %(prog)s my-table --output data.csv --region eu-west-1
158 """,
159 )
160
161 parser.add_argument("table_name", help="DynamoDB table name")
162 parser.add_argument(
163 "--output",
164 "-o",
165 help="Output CSV file (default: <table_name>.csv)",
166 )
167 parser.add_argument("--region", "-r", help="AWS region (uses default region if not specified)")
168
169 args = parser.parse_args()
170
171 # Set default output filename based on table name if not provided
172 if not args.output:
173 args.output = f"{args.table_name}.csv"
174
175 try:
176 # Scan DynamoDB table
177 items = scan_dynamodb_table(args.table_name, args.region)
178
179 if not items:
180 print("No items found in the table.")
181 return
182
183 # Save to CSV using DuckDB
184 save_to_csv_with_duckdb(items, args.output)
185
186 except Exception as e:
187 print(f"Error: {e}")
188 sys.exit(1)
189
190
191if __name__ == "__main__":
192 main()