transmog 1.0.6

A data transformation library for flattening complex nested structures into tabular formats while preserving hierarchical relationships

Transmog

A Python library for transforming complex nested data structures into flat, tabular formats while preserving hierarchical relationships.

Features

• Multiple Input Formats: JSON, JSONL, CSV
• Nested Structure Handling: Flattens deeply nested objects with customizable separators
• Array Processing: Extracts arrays as child tables with parent-child relationships maintained
• Output Options: Python dictionaries, PyArrow tables, JSON, CSV, Parquet
• Performance Features: Chunked processing, streaming output, memory optimization
• Data Integrity: Deterministic ID generation, consistent parent-child linking
• Error Recovery: Configurable strategies for handling malformed data

Installation

pip install transmog

Optional dependencies:

pip install transmog[dev]  # Development tools

Quick Example

import transmog as tm

# Sample nested data
data = {
    "user": {
        "id": 1,
        "name": "John Doe",
        "contact": {
            "email": "[e-mail]"
        },
        "orders": [
            {"id": 101, "amount": 99.99},
            {"id": 102, "amount": 45.50}
        ]
    }
}

# Process the data
processor = tm.Processor()
result = processor.process(data)

# Access the data
tables = result.to_dict()
main_table = tables["main"]
orders = tables["user_orders"]

# Export to different formats
result.write_all_json("output/json")
result.write_all_csv("output/csv")
result.write_all_parquet("output/parquet")

Configuration

# Use pre-configured modes
config = tm.TransmogConfig.memory_optimized()
# or
config = tm.TransmogConfig.performance_optimized()

# Custom configuration
config = (
    tm.TransmogConfig.default()
    .with_naming(separator=".")
    .with_processing(cast_to_string=True)
    .with_metadata(id_field="custom_id")
    .with_error_handling(max_retries=3)
)

processor = tm.Processor(config=config)

Large Dataset Processing

# Memory-optimized processing
processor = tm.Processor.memory_optimized()

# Chunked processing
result = processor.process_chunked(
    "large_data.jsonl",
    entity_name="records",
    chunk_size=1000
)

# Streaming output
processor.stream_process_file(
    "large_data.jsonl",
    entity_name="records",
    output_format="parquet",
    output_destination="output_dir"
)

Error Handling

# Skip and log errors
processor = tm.Processor().with_error_handling(recovery_strategy="skip")

# Partial recovery (preserves valid portions)
processor = tm.Processor.with_partial_recovery()

Documentation

• Installation Guide
• Getting Started
• Configuration Guide
• API Reference
• Examples

License

MIT License