Skip to content

json_storage

JsonFileBackend

Stores operations as JSON files in a specific directory.

Examples:

>>> from wintermute.core import Operation
>>> from wintermute.backends.json_storage import JsonFileBackend
>>> backend = JsonFileBackend()
>>> Operation.register_backend("json_local", backend, make_default=True)
>>> op = Operation("Project_Test1")
>>> op.addAnalyst("Ripley", "ripley", "ripley@foobar.com")
True
>>> op.save()
True
>>> op2 = Operation("Project_Test1")
>>> op2.load()
True
>>> op2
<wintermute.core.Operation object at 0x745c92ded7c0>
>>> op2.to_dict()
{'operation_name': 'Project_Test1',
    'operation_id': '8b5b3662-f597-11f0-a793-54b2030b4724',
    'start_date': '01/19/2026', 'end_date': '01/19/2026',
    'ticket': None,
    'analysts': [
        {
            'name': 'Ripley',
            'userid': 'ripley',
            'email': 'ripley@foobar.com'
        }
    ],
    'devices': [],
    'users': [],
    'cloud_accounts': [],
    'test_plans': [],
    'test_runs': []
}
Source code in wintermute/backends/json_storage.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class JsonFileBackend:
    """Stores operations as JSON files in a specific directory.

    Examples:
        >>> from wintermute.core import Operation
        >>> from wintermute.backends.json_storage import JsonFileBackend
        >>> backend = JsonFileBackend()
        >>> Operation.register_backend("json_local", backend, make_default=True)
        >>> op = Operation("Project_Test1")
        >>> op.addAnalyst("Ripley", "ripley", "ripley@foobar.com")
        True
        >>> op.save()
        True
        >>> op2 = Operation("Project_Test1")
        >>> op2.load()
        True
        >>> op2
        <wintermute.core.Operation object at 0x745c92ded7c0>
        >>> op2.to_dict()
        {'operation_name': 'Project_Test1',
            'operation_id': '8b5b3662-f597-11f0-a793-54b2030b4724',
            'start_date': '01/19/2026', 'end_date': '01/19/2026',
            'ticket': None,
            'analysts': [
                {
                    'name': 'Ripley',
                    'userid': 'ripley',
                    'email': 'ripley@foobar.com'
                }
            ],
            'devices': [],
            'users': [],
            'cloud_accounts': [],
            'test_plans': [],
            'test_runs': []
        }

    """

    def __init__(self, base_path: str = ".wintermute_data"):
        self.base_path = base_path
        os.makedirs(self.base_path, exist_ok=True)

    def _get_path(self, op_id: str) -> str:
        # Sanitize filename if needed
        safe_name = "".join(x for x in op_id if x.isalnum() or x in "-_")
        return os.path.join(self.base_path, f"{safe_name}.json")

    def save(self, operation_id: str, data: Dict[str, Any]) -> bool:
        path = self._get_path(operation_id)
        try:
            with open(path, "w", encoding="utf-8") as f:
                json.dump(data, f, default=str)
            return True
        except Exception as e:
            log.error(f"Error saving to {path}: {e}")
            return False

    def load(self, operation_id: str) -> Optional[Dict[str, Any]]:
        path = self._get_path(operation_id)
        if not os.path.exists(path):
            return None
        try:
            with open(path, "r", encoding="utf-8") as f:
                data = json.load(f)
                if isinstance(data, dict):
                    return data
                log.error(f"Data in {path} is not a valid dictionary.")
                return None
        except Exception as e:
            log.error(f"Error loading {path}: {e}")
            return None

    def list_all(self) -> List[str]:
        if not os.path.exists(self.base_path):
            return []
        return [
            f.replace(".json", "")
            for f in os.listdir(self.base_path)
            if f.endswith(".json")
        ]

    def delete(self, operation_id: str) -> bool:
        path = self._get_path(operation_id)
        if not os.path.exists(path):
            log.warning(f"Operation {operation_id} not found at {path}")
            return False
        try:
            os.remove(path)
            return True
        except Exception as e:
            log.error(f"Error deleting {path}: {e}")
            return False