"""
The data that is manipulated in the PyConfigTool is held in a
"model" object (think Model-View-Controller paradigm). This package
defines the implementation of this object tree. It is held in a ZODB
database, which makes these objects magically persistent, and subject
to transactional semmantics.
"""
#
# Copyright Qualcomm Technologies Inc, 2019.
# All Rights Reserved
#
# Python import
import logging
from typing import Callable, Any, List, Optional
from pathlib import Path
import re
# Thirdparty imports
import transaction
import persistent
from persistent.list import PersistentList
from zope.component import getUtility
from zope.interface import implementer
import ZODB
import ZODB.FileStorage
# Package imports
from qct_interfaces.database import IDatabaseLocation
from qct_interfaces.state import ISavingModelState
# Imports from this package
from .configdomain import ConfigDomain, ConfigDomainContainer, ConfigAttributeBase
from .configdomain import ConfigAttributeCheckItem, ConfigAttributeIntegerItem
from .configdomain import ConfigAttributeComboboxItem, ConfigAttributeFloatItem
from .configdomain import ConfigAttributeTextItem
from .configdomain import ConfigAttributeTableItem, TableDomain
from .configdomain import ConfigAttributeStruct, StructDomain
from .configdomain import DependencyReferenceSpec
#from pyconfigtool.utils import trace
[docs]@implementer(IDatabaseLocation)
class DatabaseLocation(): # pylint: disable=too-few-public-methods
"""This implements the ``IDatabaseLocation`` marker interface.
"""
def __init__(self):
self.location = Path('.')
[docs]class DeviceDatabase:
"""This class holds the reference to the database file that has been
opened. The ``open()`` method returns the root domain object.
See the documentation on the ``persistent`` objects that are used
in this database connection.
This class can be used as a context manager, which provides the
root ConfigDomain object::
with DeviceDatabase('some_file') as root_domain:
assert isinstance(root_domain, ConfigDomain)
...
and the database will be correctly closed and cleaned up after the
context closes.
"""
DB_FORMAT_VERSION = '1.6'
def __init__(self, pth: Path):
self.file_path = pth
self._db = None
self._connection = None
self.root_domain = None
[docs] def open(self, create_if_needed=True) -> ConfigDomain:
"""Open the file that was passed to the constructor of this class. If
the file doesn't exist then the database file is created with
a root object.
:return: The root ConfigDomain object.
"""
initialize = not self.file_path.exists()
if initialize and not create_if_needed:
raise FileNotFoundError('%s does not exist' % str(self.file_path))
storage = ZODB.FileStorage.FileStorage(str(self.file_path))
self._db = ZODB.DB(storage)
# Consolidate changes older than 24 hours
self._db.pack(days=1)
self._connection = self._db.open()
root = self._connection.root
if initialize:
root.root_domain = ConfigDomain('root')
root.version = self.DB_FORMAT_VERSION
transaction.commit()
self.root_domain = root.root_domain
if root.version != self.DB_FORMAT_VERSION:
logging.warning('Upgrading database to latest version...')
if not self.__upgrade(root):
msg = 'Failed to upgrade from version %s to %s' % \
(root.version, self.DB_FORMAT_VERSION)
logging.error(msg)
raise ValueError(msg)
transaction.commit()
logging.warning('Done.')
logging.debug("open db: %s", self._db.connectionDebugInfo())
return self.root_domain
def __pre_commit_hook(self): # pylint:disable=no-self-use
getUtility(ISavingModelState).value = True
def __post_commit_hook(self, status): # pylint:disable=no-self-use
if not status:
logging.error("Transaction commit failed")
getUtility(ISavingModelState).value = False
[docs] def commit(self) -> None:
"""Commit the current model transaction.
Before calling commit on the global transaction manager, this
method attaches callbacks to its pre- and post-commit
hooks. These hooks have to be attached for every new
transaction, which is a bit irritating. These hooks then set and
clear the ISavingModelState variable, which the user interface
can use to indicate the saving of the model data.
"""
self._connection.transaction_manager.get().addBeforeCommitHook(self.__pre_commit_hook)
self._connection.transaction_manager.get().addAfterCommitHook(self.__post_commit_hook)
transaction.commit()
[docs] def close(self) -> None:
"""Close the connection and the database, and delete the ZODB temporary
working files for the db file.
"""
# Ensure any changes are stored before closing DB
transaction.commit()
logging.debug("close db: %s", self._db.connectionDebugInfo())
self._connection.close()
self._connection = None
self._db.close()
self._db = None
self.root_domain = None
self.__delete_db_working_files(['.index', '.lock', '.tmp'])
[docs] def walk_database(self, callback: Callable[[Any], None]) -> None:
"""Walk all the instance objects that are in the database tree and
call on the ``callback`` function with each object instance
found. The objects in the tree for which the callback will be
called are of type :py:class:`ConfigDomain`,
:py:class:`ConfigContainer`, and various classes derived from
:py:class:`ConfigAttributeBase`.
"""
if self.root_domain:
self.root_domain.walk_domains(callback)
[docs] def search_database(self, callback: Callable[[Any], Any]) -> Any:
"""Search through the instance objects that are in the database tree,
calling the ``callback`` function with each object instance
found. If the callback function returns a non-``None`` value,
then the recursive walk is abandoned and this value is
returned.
The objects in the tree for which the callback will be
called are of type :py:class:`ConfigDomain`,
:py:class:`ConfigContainer`, and various classes derived from
:py:class:`ConfigAttributeBase`.
:return: Whatever first non-``None`` value that a call to
the callback function returned.
"""
if self.root_domain:
return self.root_domain.search_domains(callback)
return None
[docs] def find_attr_by_key(self, attr_key: str) -> Optional[ConfigAttributeBase]:
"""Find the attribute in the database which has the key name
provided. This method uses the ``search_database`` generic
method to perform this search.
:return: The ConfigAttributeBase-derived object if it's found, or None.
"""
# Get the key and optionally an index
rgx = re.match(r'(?P<key>.*?)(\.(?P<index>\d+))?$', attr_key)
key = rgx.group('key')
index = rgx.group('index')
def _cb(obj):
if isinstance(obj, ConfigAttributeBase):
if obj.attr_key == key:
if index and len(obj.attr_value) <= int(index):
return None
return obj
return None
return self.search_database(_cb)
[docs] def find_attr_by_value(self, attr_value: str) -> Optional[ConfigAttributeBase]:
"""Find the attribute in the database which has the value
provided. This method uses the ``search_database`` generic
method to perform this search.
:return: The ConfigAttributeBase-derived object if it's found, or None.
"""
def _cb(obj):
if isinstance(obj, ConfigAttributeBase):
if isinstance(obj.attr_value, PersistentList):
if attr_value in obj.attr_value:
return obj
elif obj.attr_value == attr_value:
return obj
return None
return self.search_database(_cb)
[docs] def find_domain_by_name(self, domain_name: str) -> Optional[ConfigDomain]:
"""Find the domain with a particular name. This method uses the
``search_database`` generic method to perform this search.
:return: The ConfigDomain object if it's found, or None.
"""
def _cb(obj):
if isinstance(obj, ConfigDomain) and obj.name == domain_name:
return obj
return None
return self.search_database(_cb)
[docs] def find_by_url(self, url: str):
"""Find the object of a requested type by its object id. This method
uses the ``search_database`` generic method to perform this search.
:return: The object if it's found, or None.
"""
# Search for an object URL in the form "domain://643721"
matched = re.match(r'(?P<obj_type>.+):\/\/(?P<obj_id>\d+)', url)
if not matched:
logging.debug("No match for %s", url)
return None
obj_type_str, obj_id_str = matched.groups()
# Apply a type filter to the search results
obj_type = {
"domain": ConfigDomain,
}.get(obj_type_str, None)
if obj_type is None:
return None
obj_id = int(obj_id_str)
def _cb(obj):
if isinstance(obj, obj_type) and id(obj) == obj_id:
return obj
return None
return self.search_database(_cb)
def __enter__(self):
return self.open()
def __exit__(self, *args):
self.close()
def __delete_db_working_files(self, suffixes: List[str]) -> None:
for suffix in suffixes:
pth = Path(str(self.file_path) + suffix)
if pth.exists():
pth.unlink()
def __upgrade(self, root) -> bool:
"""An older version of the database is being read in, so the memory
copy needs to be upgraded.
N.B. this needs careful maintenance to ensure that changes to
any of the objects in the database tree have code here to
manage the corresponding version upgrade.
N.B. there is no downgrade capability.
:return: Return whether we have upgraded to the current supported
version of this code.
"""
if root.version == '1.0':
# upgrade up to '1.1'. Two attributes were added to the
# ConfigAttributeBase class.
self.__update_class_with_attr(ConfigAttributeBase,
attr_visibility_spec=None,
attr_enablement_spec=None)
root.version = '1.1'
if root.version == '1.1':
# upgrade to 1.2. Each domain object now supports a
# _parent attibute.
_upgrade_set_parent(None, root.root_domain)
root.version = '1.2'
if root.version == '1.2':
# upgrade to the 1.3. Each ConfigAttributeBase supports
# the 'attr_tags' attribute, which is a set of strings.
self.__update_class_with_attr(ConfigAttributeBase,
attr_tags=set())
root.version = '1.3'
if root.version == '1.3':
# upgrade to the 1.4. The attr_tags attribute is now
# renamed to be attr_meta_data whose value is just a
# string or None. The attr_tags has not actually been used
# yet, so we just delete this attr from all
# ConfigAttributeBase objects, and add the attr_meta_data
# instead.
self.__update_class_with_attr(ConfigAttributeBase,
attr_meta_data=None)
self.__update_class_deleting_attr(ConfigAttributeBase,
'attr_tags')
self.__update_class_with_attr(ConfigAttributeBase,
index=None)
root.version = '1.4'
if root.version == '1.4':
# upgrade to 1.5.
# Each ConfigAttributeBase object supports the 'locked'
# attribute, which is a boolean denoting non-modifiable
# attribute data fields
# Each ConfigAttributeIntegerItem and ConfigAttributeFloatItem
# supports the 'wraps' attribute, which is a boolean.
self.__update_class_with_attr(ConfigAttributeIntegerItem,
wraps=False)
self.__update_class_with_attr(ConfigAttributeFloatItem,
wraps=False)
self.__update_class_with_attr(ConfigAttributeBase,
locked=False)
root.version = '1.5'
if root.version == '1.5':
# upgrade to 1.6.
# Each ConfigAttributeBase object supports the 'css'
# attribute, which is a cascading style sheet string
self.__update_class_with_attr(ConfigAttributeBase,
css='')
root.version = '1.6'
transaction.commit()
## Copy this template for each upgrade action:
## if root.version == '1.n - 1':
## # upgrade to the 1.n. Document what that change is here.
## root.version = '1.n'
# Return whether we have upgraded to the current supported
# version of this code
return root.version == self.DB_FORMAT_VERSION
def __update_class_with_attr(self, cls, **attrs) -> None:
"""Utility for the upgrade feature. If a particular class just needs
to have a number of attributes added, then this method will
walk the database tree adding these attributes to all instance
of the specified class.
This uses the :py:meth:`walk_database` method to walk over the
database with a callback that does the attribute updates.
"""
def _cb(instance):
if isinstance(instance, cls):
logging.info("Upgrading %s with %s", instance.attr_key, attrs)
for key, val in attrs.items():
setattr(instance, key, val)
self.walk_database(_cb)
def __update_class_deleting_attr(self, cls, *attrs) -> None:
"""Utility for the upgrade feature. If a particular class just needs
to have a number of attributes removed, then this method will
walk the database tree deleting these attributes from all instances
of the specified class.
This uses the :py:meth:`walk_database` method to walk over the
database with a callback that deltes the attribute.
"""
def _cb(instance):
if isinstance(instance, cls):
for key in attrs:
delattr(instance, key)
self.walk_database(_cb)
def _upgrade_set_parent(parent, domain):
"""Utility method to upgrade a 1.1 to a 1.2 version of the database. A
_parent attribute was introduced at this point, and this function
patches the existing ConfigDomain objects.
"""
setattr(domain, '_parent', parent)
for dom in domain.child_domains:
_upgrade_set_parent(domain, dom)