Source code for capsul.engine.database_json
# -*- coding: utf-8 -*-
from __future__ import absolute_import
import os
import os.path as osp
import six
import uuid
import json
from capsul.engine.database import DatabaseEngine
[docs]
class JSONDBEngine(DatabaseEngine):
'''
A JSON dictionary implementation of :py:class:`capsul.engine.database.DatabaseEngine`
'''
def __init__(self, json_filename):
if json_filename is not None:
self.json_filename = osp.normpath(osp.abspath(json_filename))
else:
self.json_filename = None
self.read_json()
def read_json(self):
if self.json_filename is not None and osp.exists(self.json_filename):
self.json_dict = json.load(open(self.json_filename))
self.modified = False
else:
self.json_dict = {}
self.modified = True
def commit(self):
if self.modified and self.json_filename is not None:
parent = osp.dirname(self.json_filename)
if not osp.exists(parent):
os.makedirs(parent)
json.dump(self.json_dict, open(self.json_filename, 'w'), indent=2)
self.modified = False
def rollback(self):
self.read_json()
[docs]
def set_named_directory(self, name, path):
if path:
path = osp.normpath(osp.abspath(path))
self.json_dict.setdefault('named_directory', {})[name] = {'name': name,
'path': path}
self.modified = True
else:
named_directory = self.json_dict.get('named_directory')
if named_directory is not None:
named_directory.pop(name, None)
self.modified = True
[docs]
def named_directory(self, name):
return self.json_dict.get('named_directory', {}).get(name, {}).get('path')
[docs]
def named_directories(self):
return self.json_dict.get('named_directory', {}).values()
[docs]
def set_json_value(self, name, json_value):
self.json_dict.setdefault('json_value', {})[name] = json_value
self.modified = True
[docs]
def json_value(self, name):
return self.json_dict.get('json_value', {}).get(name)