Implement DynamoDB persistence backend for configuration management

# Implement DynamoDB persistence backend for configuration management

## Summary
Add DynamoDB as an alternative persistence backend for organization configuration, enabling centralized configuration management, version control, and multi-user collaboration. Include automatic migration capabilities between file and DynamoDB storage with transparent configuration-driven switching.

## Background
Currently, orgtool uses YAML files for configuration storage, which creates challenges for:
- **Multi-user collaboration**: File conflicts and synchronization issues
- **Centralized management**: No single source of truth for large organizations
- **Version control**: Limited audit trail and change tracking
- **Scalability**: File-based approach doesn't scale for enterprise environments
- **Concurrent access**: Race conditions when multiple users modify configurations

## Reference Implementation - Complete Code from AXA Fork

### 1. DynamoDB Model Class (from `awsorgs/spec.py`)

#### Core DynamoDBModel Class:
```python
def dynamodb_resource():
return boto3.resource('dynamodb')

class DynamoDBModel(BaseModel):
def __init__(self, log, config, autoload=True, recursive=False, spec_dir=None, master_account_id=None, table=None, access=dynamodb_resource):
config = config.split('/')

self.table = table
self.table_name = table.table_name if table else config[0]
self.entity = config[1]
config_key = config[2] if len(config) > 2 else None
self.access = access

super().__init__(
log,
config=config_key,
autoload=autoload,
recursive=recursive,
spec_dir=spec_dir,
master_account_id=master_account_id
)

def get_table(self):
if not self.table:
db = self.access()
self.table = db.Table(self.table_name)
return self.table

def scan_config_file(self, config_file=None):
if not config_file:
config_file = "config"
self.log.debug("loading config : %s", config_file)

try:
# load item from dynamodb
config = self.get_table().get_item(
Key={
'entity': self.entity,
'key': config_file
}
)['Item']
config.pop('entity')
config.pop('key')

except:
self.log.exception("cant load config %s:%s/%s",
self.table_name, self.entity, config_file)
return None
if self.log.isEnabledFor(logging.DEBUG):
self.log.debug("config: %s", json.dumps(config, indent=2))
return config

def load_specs(self):
try:
spec_object = self.get_table().get_item(
Key={
'entity': self.entity,
'key': self.spec_dir
}
)['Item']
spec_object.pop('entity')
spec_object.pop('key')
self.version = spec_object.pop('version')
except Exception as e:
self.log.exception("cannot load specs %s:%s/%s",
self.table_name, self.entity, self.spec_dir)
return (None, [str(e)])

self.log.debug("spec_object:\n%s", json.dumps(spec_object, indent=2))
return (spec_object, None)

def get_spec_dir(self):
spec_dir = self.spec_dir
if not spec_dir:
spec_dir = self.config['spec_dir']

if spec_dir and spec_dir.startswith("--/"):
spec_dir = spec_dir.replace("--/", "")

if not spec_dir:
spec_dir = "spec"

self.log.debug("spec_dir: %s", spec_dir)
return spec_dir

def child_model(self, child, spec_dir=None, master_account_id=None):
return DynamoDBModel(self.log, child, table=self.table, spec_dir=spec_dir, master_account_id=master_account_id)

def dump_specs(self, exec, message={}):
self.log.debug("dump specs to DB")

if not exec:
return

if self.has_changed():
self.dump_specs_in_transaction(message)

super().dump_specs(exec)

def _update_string(self):
values_to_update = [
f"{key} = :{key}"
for key in self.modified_specs
]
return "SET " + ", ".join(values_to_update)

def _history(self, models, table, message):
ts = timestamp()
ttl = int((datetime.datetime.now() + datetime.timedelta(days=30)).timestamp())

changes = [
f"{model.entity}.{change}"
for model in models
for change in model.modified_specs
]

return {
'Put': {
'Item': {
'entity': '#HISTORY',
'key': ts,
'changes': changes,
'message': message,
'ttl': ttl,
},
'TableName': table,
'ConditionExpression': 'attribute_not_exists(#key)',
'ExpressionAttributeNames': {'#key': 'key'},
}
}

def dump_specs_in_transaction(self, message):
self.log.debug("dump specs to DB in transaction")

table = self.get_table()
models = [self] + list(self.related_models)
models = [model for model in models if model.modified_specs]

if not models:
return

transact_items = []

for model in models:
expression_attribute_values = {
f":{key}": value
for key, value in model.specs.items()
if key in model.modified_specs
}
expression_attribute_values[':version'] = timestamp()

transact_items.append({
'Update': {
'Key': {
'entity': model.entity,
'key': model.spec_dir,
},
'UpdateExpression': model._update_string() + ", version = :version",
'ExpressionAttributeValues': expression_attribute_values,
'TableName': table.table_name,
}
})

transact_items.append(self._history(models, table.table_name, message))

try:
table.meta.client.transact_write_items(TransactItems=transact_items)
for model in models:
model.modified_specs = set()
except Exception as e:
self.log.exception("Error during transaction: %s", e)
raise
```

#### Factory Pattern Integration:
```python
@staticmethod
def create_model(log, args, recursive=False, autoload=True):
"""
Static factory method for creating a model based on args
"""
config = args["--config"]
if config.startswith('dynamodb:'):
return DynamoDBModel(
log,
config=config.split(":")[1],
spec_dir=args.get("--spec-dir"),
master_account_id=args.get("--master-account-id"),
recursive=recursive,
autoload=autoload,
)

return FileModel(
log,
config=config,
spec_dir=args.get("--spec-dir"),
master_account_id=args.get("--master-account-id"),
recursive=recursive,
autoload=autoload,
)
```

### 2. Migration Tools (from `awsorgs/migrate.py`)

#### Complete Migration Implementation:
```python
# Migration from file model to DynamoDB and inversely
import boto3
import os
import ruamel.yaml
from decimal import Decimal
from boto3.dynamodb.conditions import Attr, Not
from awsorgs.spec import upload_entity

PATH_PREFIX = 'organization/.orgtool' # Adapt to orgtool structure
VALID_FILES = [
'organizational_units',
'sc_policies',
'accounts',
'users',
'groups',
'delegations',
'local_users',
'custom_policies',
'policy_sets',
'stacks',
]

def from_file_to_dynamodb(path, table_name):
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)

org_path = os.path.join(path, PATH_PREFIX)
for dir_name in os.listdir(org_path):
entity_path = os.path.join(org_path, dir_name)
if os.path.isdir(entity_path):
upload_entity(table, org_path, dir_name)

# set the value of last update to 0
table.put_item(
Item={
'entity': '#LAST_UPDATE',
'key': '#LAST_UPDATE',
'value': '0',
},
)

def transform_config(config):
config.pop('entity')
config.pop('key')
if 'spec_dir' in config:
config['spec_dir'] = '--/spec.d'
return config

def transform_specs(specs):
specs.pop('entity')
specs.pop('key')
specs.pop('version')

ous = specs.get('organizational_units', [{}])
root = ous[0]
children = root.get('Child_OU', [])
for child in children:
include_path = child.get('IncludeConfigPath')
if include_path:
entity = include_path.split('/')[1]
child['IncludeConfigPath'] = '/'.join([PATH_PREFIX, entity, 'config.yaml'])

return specs

def from_dynamodb_to_file(table_name, path):
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)
yaml = ruamel.yaml.YAML()
org_path = os.path.join(path, PATH_PREFIX)

last_evaluated_key = "FIRST TIME"

while last_evaluated_key:
args = {
"FilterExpression": Not(Attr('entity').begins_with(':prefix')),
"ExpressionAttributeValues": {':prefix': '#'},
}
table_items = table.scan(**args)

for item in table_items['Items']:
if item['key'] == 'config':
entity_path = os.path.join(org_path, item['entity'])
if not os.path.exists(entity_path):
os.makedirs(entity_path)
with open(os.path.join(entity_path, 'config.yaml'), 'w') as f:
yaml.dump(transform_config(item), f)

elif item['key'] == 'spec':
entity_path = os.path.join(org_path, item['entity'], 'spec.d')
if not os.path.exists(entity_path):
os.makedirs(entity_path)
specs = transform_specs(item)
for key, value in specs.items():
if key in VALID_FILES:
with open(os.path.join(entity_path, f"{key}.yaml"), 'w') as f:
yaml.dump({key: value}, f)

last_evaluated_key = table_items.get('LastEvaluatedKey')
if last_evaluated_key:
args['ExclusiveStartKey'] = last_evaluated_key
else:
last_evaluated_key = None

def upload_entity(table, org_path, entity):
"""Upload entity configuration and specs to DynamoDB"""
yaml = ruamel.yaml.YAML()

# Upload config
config_path = os.path.join(org_path, entity, 'config.yaml')
if os.path.exists(config_path):
with open(config_path, 'r') as f:
config = yaml.load(f)

config['entity'] = entity
config['key'] = 'config'
table.put_item(Item=config)

# Upload specs
spec_dir = os.path.join(org_path, entity, 'spec.d')
if os.path.exists(spec_dir):
specs = {'entity': entity, 'key': 'spec', 'version': timestamp()}

for filename in os.listdir(spec_dir):
if filename.endswith('.yaml'):
spec_name = filename[:-5] # Remove .yaml extension
if spec_name in VALID_FILES:
with open(os.path.join(spec_dir, filename), 'r') as f:
spec_data = yaml.load(f)
specs[spec_name] = spec_data.get(spec_name, [])

table.put_item(Item=specs)

def timestamp():
"""Generate timestamp for versioning"""
return datetime.datetime.now().isoformat()
```

### 3. Table Management (from `awsorgs/cache.py` - adapt for config)

#### DynamoDB Table Creation:
```python
def init_config_table(table_name, log):
"""Initialize DynamoDB table for configuration storage"""
log.info("Initialising config table %s", table_name)
dyn_res = boto3.resource('dynamodb')

try:
table = dyn_res.Table(table_name)
table.table_status # Test if table exists
log.info("Table %s already exists", table_name)
return table
except dyn_res.meta.client.exceptions.ResourceNotFoundException:
log.info("Creating table %s", table_name)

table = dyn_res.create_table(
TableName=table_name,
KeySchema=[
{'AttributeName': 'entity', 'KeyType': 'HASH'},
{'AttributeName': 'key', 'KeyType': 'RANGE'}
],
AttributeDefinitions=[
{'AttributeName': 'entity', 'AttributeType': 'S'},
{'AttributeName': 'key', 'AttributeType': 'S'}
],
BillingMode='PAY_PER_REQUEST',
TimeToLiveSpecification={
'AttributeName': 'ttl',
'Enabled': True
}
)

# Wait for table to be created
table.wait_until_exists()
log.info("Table %s created successfully", table_name)
return table
except Exception as e:
log.error("Failed to initialize table %s: %s", table_name, e)
raise

def delete_config_table(table_name, log):
"""Delete DynamoDB configuration table"""
log.info("Deleting config table %s", table_name)
dyn_res = boto3.resource('dynamodb')

try:
table = dyn_res.Table(table_name)
table.delete()
table.wait_until_not_exists()
log.info("Table %s deleted successfully", table_name)
except dyn_res.meta.client.exceptions.ResourceNotFoundException:
log.info("Table %s does not exist", table_name)
except Exception as e:
log.error("Failed to delete table %s: %s", table_name, e)
raise
```

## Proposed Solution

### 1. Configuration-Driven Backend Selection
```yaml
# config.yaml - File backend (default)
persistence:
backend: "file"
config_path: "~/.orgtool"

# config.yaml - DynamoDB backend
persistence:
backend: "dynamodb"
table_name: "orgtool-config"
entity: "my-organization"
region: "us-east-1"
auto_create_table: true

# config.yaml - Migration configuration
migration:
enabled: true
source_backend: "file"
target_backend: "dynamodb"
source_path: "~/.orgtool"
target_table: "orgtool-config"
target_entity: "my-org"
backup_source: true
validate_migration: true
```

### 2. CLI Integration
```bash
# DynamoDB backend usage
orgtool --config dynamodb:table-name/entity-name/config-key organization
orgtoolaccounts --config dynamodb:orgtool-config/my-org/config create

# Migration commands
orgtoolconfigure migrate file-to-dynamodb --source ~/.orgtool --target-table orgtool-config --entity my-org [--exec]
orgtoolconfigure migrate dynamodb-to-file --source-table orgtool-config --entity my-org --target ~/.orgtool [--exec]

# Table management
orgtoolconfigure table create --table-name orgtool-config [--exec]
orgtoolconfigure table delete --table-name orgtool-config [--exec]
```

## Implementation Plan

### Files to create/modify:

#### 1. **`orgtool/persistence.py`** (new file)
**Backport complete DynamoDBModel class from fork**
- Copy entire `DynamoDBModel` class with all methods
- Adapt `dynamodb_resource()` function
- Include transactional write operations
- Add version control and history tracking

#### 2. **`orgtool/migrate.py`** (new file)
**Backport complete migration tools from fork**
- Copy `from_file_to_dynamodb()` function
- Copy `from_dynamodb_to_file()` function
- Copy `upload_entity()` function
- Copy transformation functions
- Adapt `PATH_PREFIX` and `VALID_FILES` for orgtool

#### 3. **`orgtool/table_manager.py`** (new file)
**Create table management utilities**
- Adapt `init_config_table()` from cache.py
- Add `delete_config_table()` function
- Add table validation and health checks

#### 4. **`orgtool/spec.py`** (modify existing)
**Add factory pattern and backend detection**
- Copy factory method from fork
- Add persistence configuration loading
- Integrate automatic table creation

#### 5. **`orgtool/configure.py`** (modify existing)
**Add migration and table management commands**
- Add migration CLI commands
- Add table management CLI commands
- Add configuration validation

### Code Backporting Checklist

**From fork `awsorgs/spec.py` (lines 713-850):**
- [ ] Complete `DynamoDBModel` class
- [ ] `scan_config_file()` method
- [ ] `load_specs()` method
- [ ] `dump_specs_in_transaction()` method
- [ ] `_history()` method for change tracking
- [ ] `_update_string()` method
- [ ] Factory pattern in `create_model()`

**From fork `awsorgs/migrate.py` (complete file):**
- [ ] `from_file_to_dynamodb()` function
- [ ] `from_dynamodb_to_file()` function
- [ ] `upload_entity()` function
- [ ] `transform_config()` function
- [ ] `transform_specs()` function
- [ ] `VALID_FILES` list adaptation

**From fork `awsorgs/cache.py` (table management):**
- [ ] Table creation logic
- [ ] Error handling patterns
- [ ] Wait conditions for table operations

## Benefits

1. **Centralized Configuration**: Single source of truth for organization config
2. **Multi-User Collaboration**: Concurrent access with proper versioning
3. **Version Control**: Built-in change tracking and audit trails
4. **Transactional Updates**: Atomic operations for data consistency
5. **Automatic Migration**: Seamless switching between backends
6. **Enterprise Ready**: Proven implementation from production environment

## Requirements

- DynamoDB permissions for table operations
- Backward compatibility with existing file-based configurations
- Atomic migration operations (all-or-nothing)
- Data validation during migration
- Rollback capability for failed migrations

## Acceptance Criteria

- [ ] Backport complete DynamoDBModel class from fork
- [ ] Backport complete migration tools from fork
- [ ] Configuration-driven backend selection
- [ ] Automatic DynamoDB table creation/deletion
- [ ] CLI commands for migration and table management
- [ ] Transactional updates with version control
- [ ] Change history tracking with TTL
- [ ] Comprehensive error handling and rollback
- [ ] Data validation during migration
- [ ] Performance benchmarks vs file backend
- [ ] Documentation for all persistence backends
- [ ] Unit tests for DynamoDB operations
- [ ] Integration tests for migration scenarios

## Priority
**Medium** - Valuable for enterprise environments with proven implementation available.

## Issue Type
**Feature**

## Labels
`enhancement`

Start by reading awsorgs/spec.py and the existing FileModel and create_model factory, then inspect the migration entry points in awsorgs/migrate.py. Compare the referenced DynamoDB model and file-to-DynamoDB conversion flow with the repository structure; done means the backend can be selected by configuration and migrations work in both directions with versioned updates.

Written by the indexing model from the issue text.