lbry-sdk/lbrynet/metadata/StructuredDict.py

65 lines
2 KiB
Python
Raw Normal View History

import jsonschema
import logging
from jsonschema import ValidationError
log = logging.getLogger(__name__)
class StructuredDict(dict):
"""
A dictionary that enforces a structure specified by a schema, and supports
migration between different versions of the schema.
"""
# To be specified in sub-classes, an array in the format
# [(version, schema, migration), ...]
_versions = []
# Used internally to allow schema lookups by version number
_schemas = {}
version = None
def __init__(self, value, starting_version, migrate=True, target_version=None):
dict.__init__(self, value)
self.version = starting_version
self._schemas = dict([(version, schema) for (version, schema, _) in self._versions])
self.validate(starting_version)
if migrate:
self.migrate(target_version)
def _upgrade_version_range(self, start_version, end_version):
after_starting_version = False
for version, schema, migration in self._versions:
if not after_starting_version:
if version == self.version:
after_starting_version = True
continue
yield version, schema, migration
if end_version and version == end_version:
break
def validate(self, version):
jsonschema.validate(self, self._schemas[version])
def migrate(self, target_version=None):
if target_version:
2016-11-30 21:20:45 +01:00
assert self._versions.index(target_version) > self.versions.index(self.version), \
"Current version is above target version"
for version, schema, migration in self._upgrade_version_range(self.version, target_version):
migration(self)
try:
self.validate(version)
except ValidationError as e:
2016-11-30 21:20:45 +01:00
raise ValidationError(
"Could not migrate to version %s due to validation error: %s" %
(version, e.message))
self.version = version