# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2009 SnapLogic, Inc. All rights reserved.
#
# See http://www.snaplogic.org for more information about
# the SnapLogic project.
#
# This program is free software, distributed under the terms of
# the GNU General Public License Version 2. See the LEGAL file
# at the top of the source tree.
#
# "SnapLogic" is a trademark of SnapLogic, Inc.
#
#
# $
# $Id: request_proxy.py 9423 2009-10-22 20:28:19Z dhiraj $
"""
Provides the RequestProxy class used by L{snaplogic.server.repository.get_instance} to execute API requests
against an internal store object.
"""
from __future__ import with_statement
from snaplogic.common.snap_exceptions import *
from snaplogic.common.config import snap_config
from snaplogic.common import snap_log
from snaplogic.common import snap_http_lib
from snaplogic.common import uri_prefix
from snaplogic.server.http_request import HttpRequest
from snaplogic.server import auth,uri_checker
from snaplogic.server.auth import PERM_READ,PERM_WRITE,PERM_EXEC
from snaplogic.snapi_base import resdef
from snaplogic.snapi_base import keys
class LogHandle(object):
request_count = 0
def __init__(self, log, section, *params):
self._log = log
self._section = section
self._params = params
def __enter__(self):
if self._log is not None:
LogHandle.request_count += 1
self._request_id = LogHandle.request_count
self._log(snap_log.LEVEL_DEBUG, "Entering %s [%d] (%s)" % (self._section,
self._request_id,
repr(self._params)))
def __exit__(self, exc_type, exc_val, exc_tb):
if self._log is not None:
if exc_type is None:
self._log(snap_log.LEVEL_DEBUG, "Exiting %s [%d]" % (self._section, self._request_id))
else:
self._log(snap_log.LEVEL_DEBUG, "Exiting %s [%d] (Error: %s)" % (self._section,
self._request_id,
str(exc_val)))
class RequestProxy(object):
"""
A proxy between the public repository API and the internal store instances and remote servers.
When a call to L{snaplogic.server.repository.get_instance} is made, an instance of this class is
returned. Objects of this provide a single API for retrieving local resources. Local resources
are those located in the main process repository used by the calling process.
All request methods of the proxy object take a resource URI. This URI is used to determine if a resource
is local or remote. If the URI contains no server URL (i.e. http://server.company.com:8088) or the URL
points to the local server, the resource is considered local.
"""
def __init__(self, store, repo_log):
super(RequestProxy, self).__init__()
self._store = store
self._repo_log = repo_log
common = snap_config.get_instance().get_section('common')
parsed = snap_http_lib.parse_uri(common['main_process_uri'])
self._main_server_url = common['main_process_uri']
self._main_server_hostname = parsed.hostname.lower()
self._main_server_port = parsed.port
self._is_main_server = common['i_am_main_process']
# Make sure the server URL does not end with a slash
self._main_server_url.rstrip('/')
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, exc_tb):
self.close()
return False
def close(self):
if self._store is not None:
self._store.close()
def _logger(self, section, *params):
return LogHandle(self._repo_log, section, *params)
def _parse_uri(self, uri, strict=False):
"""
Parse and validate the URI.
The URI is parsed into a tuple of the form (server_url, res_uri). The server_url element gives the
base URL of the server the resource exists on. If server_url would point to the main server and this
is the main server process, then None is returned for server_url to indicate a local request to the
store should be performed. If the hostname and port are omitted the local server is assumed. The
res_uri contains the remaining portion of the original URI giving the location of the resource on the
server.
If strict is given and True, additional checks are performed to insure a URI is not using any internal
namespaces. This is typically only used when creating a resource.
@param uri: A resource URI.
@type uri: string
@param strict: Flag indicating if strict URI namespace restrictions should be used to not allow URIs in
internal namespaces.
@type strict: bool
@return: A tuple of (server_url, res_uri)
@rtype: tuple
@raise SnapInvalidURIError: The URI is malformed.
@raise SnapPermissionError: The URI is using a restricted namespace and strict is True.
"""
if not uri:
raise SnapValueError("Resource URI is empty.")
elif uri[0] == '/':
# This is a relative URI.
if strict and uri.startswith('/__snap__'):
raise SnapValueError("Resource URI '%s' is located in a restricted namespace." % uri)
else:
snap_http_lib.check_relative_uri(uri)
return (None, uri)
parsed = snap_http_lib.parse_uri(uri, 'http')
if parsed.scheme != 'http' and parsed.scheme != 'https':
raise SnapValueError("Resource URI '%s' uses an unsupported scheme." % uri)
elif parsed.hostname is None:
raise SnapValueError("Resource URI '%s' missing hostname or malformed relative URI." % uri)
elif parsed.query != '':
raise SnapValueError("Resource URI '%s' cannot contain a query string." % uri)
elif not parsed.path or parsed.path == '/':
raise SnapValueError("Resource URI '%s' must contain a valid path." % uri)
elif strict and parsed.path.startswith('/__snap__'):
raise SnapValueError("Resource URI '%s' is located in a restricted namespace." % uri)
if self._is_main_server and uri_checker.is_mine(uri):
return (None, parsed.path)
else:
return ('%s://%s' % (parsed.scheme, parsed.netloc), parsed.path)
def _parse_uri_list(self, uri_list):
local_uris = {}
remote_uris = []
for uri in uri_list:
(server_url, res_uri) = self._parse_uri(uri)
if server_url is None:
local_uris[res_uri] = uri
else:
remote_uris.append(uri)
return (local_uris, remote_uris)
def _make_absolute_uri(self, uri):
"""
Ensure that a URI is absolute.
If the URI is relative, the main server's base URL will be prepended. If the URI is absolute,
it is returned as-is.
@param uri: A relative or absolute URI.
@type uri: str
@return: Absolute URI if uri is relative, or uri as-is if already absolute.
@rtype: str
"""
if uri.startswith('/'):
return self._main_server_url + uri
else:
return uri
def _validate_resdef(self, resdef_dict):
"""
Validates the ResDef for use with the repository.
Validation in this context does not validate that the ResDef is valid for pipeline execution. Only validation
of information needed for storage is considered. This method raises an exception if any error is found.
Otherwise, the method returns without error.
@param resdef_dict: A value to check for validity as a ResDef dictionary.
@type resdef_dict: dict
@raises SnapObjTypeError: The resdef value is not a dictionary.
@raises SnapValueError: The resdef value contains invalid values.
"""
if type(resdef_dict) is not dict:
raise SnapObjTypeError("ResDef must be dictionary; received type '%s' instead." % type(resdef_dict).__name__)
else:
try:
res = resdef.create_from_dict(resdef_dict)
except Exception, e:
# Problem parsing resdef_dict.
raise SnapValueError("Invalid ResDef dictionary: " + str(e))
if isinstance(res, resdef.PipelineResDef):
for dep_name in res.list_resource_names():
# Let snap_http_lib validate the URI and throw exception if bad
snap_http_lib.parse_uri(res.get_resource_uri(dep_name))
def _map_dict_keys(self, src_dict, key_map):
dest_dict = {}
for (key, value) in src_dict.iteritems():
dest_dict[key_map[key]] = value
return dest_dict
def _recurse_dependencies(self, uri, local_only, dependencies=None):
"""
Recusively find the dependency resource set of pipelines.
@param uri: URI of current resource to examine.
@type uri: str
@param local_only: Flag indicating if only local resources should be returned.
@type local_only: bool
@param dependencies: Set of dependencies already found or None if this is the first call.
@type dependencies: set
@return: Set of all dependencies within the (nested) pipeline located uri.
@rtype: set
"""
if dependencies is None:
dependencies = set()
result = self._store.read_resources([uri])
if result[keys.ERROR]:
raise SnapObjNotFoundError("Resource '%s' does not exist" % uri)
res = resdef.create_from_dict(result[keys.SUCCESS][uri][keys.RESDEF])
if isinstance(res, resdef.PipelineResDef):
for res_name in res.list_resource_names():
sub_uri = res.get_resource_uri(res_name)
sub_absolute = self._make_absolute_uri(sub_uri)
if sub_absolute not in dependencies:
(server_url, sub_path) = self._parse_uri(sub_uri)
if server_url is None:
# Local resource.
dependencies.add(sub_absolute)
try:
self._recurse_dependencies(sub_uri, local_only, dependencies)
except SnapObjNotFoundError:
# Don't throw an error for missing dependencies.
pass
elif not local_only:
dependencies.add(sub_uri)
return dependencies
def _check_uri(self, uri, credentials):
"""
Check credentials for a URI.
If credentials is not None, this method returns the same as auth.check_uri. However, if credentials is None,
this is considered a special condition of server permissions that have access to the entire repository.
The full permission set (READ, WRITE, EXEC) is returned in this case.
@param uri: A URI string.
@type uri: str
@param credentials: A credential tuple or None for unrestricted access (server permissions).
@type credentials: tuple
@return: Permission bit mask for the URI based on credentials.
@rtype: int
"""
if credentials is None:
return PERM_READ | PERM_WRITE | PERM_EXEC
else:
return auth.check_uri(uri, *credentials)
def create_resource(self, uri, resdef, force_flag=False):
"""
Create a resource in the store.
A new resource is created in the store with resdef as the object. If uri is None, a default URI is
generated. The final URI, GUID, and gen_id are returned.
If force_flag is given and True and a resource already exists with the given URI, it will be overwritten
with the new value. A new GUID will be generated and the Generation ID will be reset to 0.
@param uri: URI to associate with the resource or None to generate one.
@type uri: string
@param resdef: A ResDef dictionary.
@type resdef: dict
@param force_flag: A flag indicating if the operation should be forced despite guid/gen_id conflicts.
@type force_flag: bool
@return: A dictionary containing the uri, GUID, and gen_id.
@rtype: tuple
@raise SnapObjExistsError: A resource is already present at the given URI.
@raise SnapObjTypeError: The resdef is not a dictionary.
@raise SnapValueError: The resdef is invalid.
@raise SnapObjPermissionError: URL is remote and only local requests are allowed.
"""
with self._logger("create resource", uri, force_flag):
if uri is not None:
(server_url, res_uri) = self._parse_uri(uri, True)
if server_url is not None:
raise SnapObjPermissionError("Creation of remote resource '%s' via this server is not allowed." % uri)
else:
uri = res_uri
self._validate_resdef(resdef)
return self._store.create_resource(uri, resdef, force_flag)
def read_resources(self, uri_list, credentials=None):
"""
Read resources from the store.
Read resources out the store and return a list of ResDefs for each. For each URI listed in url_list,
the store performs a lookup and translates the stored resource into a ResDef. The result is a dictionary
where the keys are the URIs found in uri_list and the value another dictionary of the following for:
{keys.GUID: '1df91d3613b14bc8b87e454df4cbe919',
keys.GEN_ID: 5,
keys.RESDEF: <ResDef dictionary>}
If a resource is not found, the value for its URI key in the result will be None.
@param uri_list: A list of URIs of resources to read.
@type uri_list: list
@param credentials: A credential tuple of (username, groups) or None to ignore credentials.
@type credentials: tuple
@return: A dictionary mapping URI to resource data.
@rtype: dict
@raise SnapObjPermissionError: URL is remote and only local requests are allowed.
"""
with self._logger("read resources", uri_list):
(local_uris, remote_uris) = self._parse_uri_list(uri_list)
if remote_uris:
raise SnapObjPermissionError("Reading of remote resource '%s' via this server is not allowed" % remote_uris[0])
temp_result = self._store.read_resources(local_uris.keys())
error = self._map_dict_keys(temp_result[keys.ERROR], local_uris)
# May need to restrict access or amount of information returned depending on credentials
filtered_success = {}
for (uri, info) in temp_result[keys.SUCCESS].iteritems():
perms = self._check_uri(uri, credentials)
if auth.can_read(perms):
filtered_success[uri] = info
elif auth.can_exec(perms):
info[keys.RESDEF] = resdef.create_from_dict(info[keys.RESDEF]).describe()
filtered_success[uri] = info
else:
error[uri] = "Not authorized"
success = self._map_dict_keys(filtered_success, local_uris)
return {keys.SUCCESS: success, keys.ERROR: error}
def list_resources(self, uri_list=None, credentials=None):
"""
Return dictionary of resources contained in store.
Return a dictionary of resources stored in the repository. Each element in the returned list is also a
dictionary providing minimal information about the resource. An example follows:
{keys.GUID: '1df91d3613b14bc8b87e454df4cbe919',
keys.GEN_ID: 5}
Additional fields may be added later, but the above four should always be available.
@param uri_list: A list of URIs to restrict the listing to or None for all resources in repository.
@type uri_list: list
@param credentials: A credential tuple of (username, groups) or None to ignore credentials.
@type credentials: tuple
@return: A dictionary of dictionaries describing the resources in the store.
@rtype: dict
@raise SnapObjPermissionError: The uri_list contains a remote URI.
"""
with self._logger("list resources", uri_list):
if uri_list is None:
result = self._store.list_resources()
else:
(local_uris, remote_uris) = self._parse_uri_list(uri_list)
if remote_uris:
raise SnapObjPermissionError("Listing of remote resource '%s' via this server is not allowed." % remote_uris[0])
else:
result = self._store.list_resources(local_uris.keys())
# Filter URIs that the credentials don't have access to.
filtered = {}
for (uri, info) in result.iteritems():
if (self._check_uri(uri, credentials) & (PERM_READ | PERM_WRITE | PERM_EXEC)) != 0:
filtered[uri] = info
if uri_list is None:
return filtered
else:
return self._map_dict_keys(filtered, local_uris)
def summarize_resources(self, uri_list=None, details=None, credentials=None):
"""
Return dictionary of resources contained in store with specified details.
Return a dictionary of resources stored in the repository. Each element in the returned list is also a
dictionary providing minimal information about the resource. An example follows:
{keys.GUID: '1df91d3613b14bc8b87e454df4cbe919',
keys.GEN_ID: 5,
keys.SUMMARY: {...}}
The keys.SUMMARY key of the dictionary will contain a subset of the resdef dictionary keys if
the details paramter is given. Otherwise, the key will not be present.
The amount of information returned depends on what permissions the user has for this data:
- READ: Can access anything within the resdef.
- WRITE: Only has permission to see that the resource exists. No resdef data.
- EXEC: Can see the resource exists and any resdef data related to execution (known as DESCRIBE view).
@param uri_list: A list of URIs to restrict the listing to or None for all resources in repository.
@type uri_list: list
@param details: A list of detail keys to include in the summary of the resource.
@type details: list
@param credentials: A credential tuple of (username, groups) or None to ignore credentials.
@type credentials: tuple
@return: A dictionary of dictionaries describing the resources in the store.
@rtype: dict
@raise SnapObjPermissionError: The uri_list contains a remote URI.
"""
with self._logger("summarize resources", uri_list):
# Don't read in the resdef (can be slow) if no details are needed.
if uri_list is None:
if details:
result = self._store.read_all_resources()
else:
result = self._store.summarize_resources()
else:
(local_uris, remote_uris) = self._parse_uri_list(uri_list)
if remote_uris:
raise SnapObjPermissionError("Summary of remote resource '%s' via this server is not allowed." % remote_uris[0])
else:
if details:
read_result = self._store.read_resources(local_uris.keys())
result = read_result[keys.SUCCESS]
# Map all the error resources to not found value None.
for uri in read_result[keys.ERROR]:
result[uri] = None
else:
result = self._store.summarize_resources(local_uris.keys())
# Need to run through each resource and filter it depending on user's permission for the URI.
filtered = {}
for (uri, info) in result.iteritems():
perms = self._check_uri(uri, credentials)
res = info.pop(keys.RESDEF, None) if info is not None else None
if auth.can_read(perms):
# Read can access of the resource.
pass
elif auth.can_exec(perms):
# Filter resdef to only contain execute-level (DESCRIBE) data.
if res is not None:
res = resdef.create_from_dict(res).describe()
elif auth.can_write(perms):
# Write permissions can only see the resource and its identification data. ResDef information
# is not returned.
res = None
else:
# Credentials have no permissions for this URI, so do not include it in results.
continue
filtered[uri] = info
if details and info is not None:
# Only create a summary key if details are asked for.
if res is not None:
# Only send the elements requested.
summary = dict(item for item in res.iteritems() if item[0] in details)
info[keys.SUMMARY] = summary
else:
info[keys.SUMMARY] = {}
if uri_list is None:
return filtered
else:
return self._map_dict_keys(filtered, local_uris)
def update_resource(self, uri, guid, gen_id, resdef, force_flag=False):
"""
Update the resource associated with uri.
First checks that the guid and gen_id values stored for the uri match those provided as arguments.
If they do not, an exception is thrown. Otherwise, the resource stored at uri is updated with the new value
defined by resdef.
If force_flag is given and True, a resource already present at the given URI will be overwritten depending
on the value of guid:
- If guid == None, any resource present at the URI will be overwritten. This case is treated the same
as if create_resource was called with the given URI and force_flag == True.
- If guid != None, a resource will only be overwritten if the stored GUID matches the one given. Otherwise,
a SnapObjExistsError will be raised instead.
@param uri: URI of resource.
@type uri: string
@param guid: GUID expected in the store.
@type guid: string
@param gen_id: Generation ID expected in the store.
@type gen_id: integer
@param resdef: ResDef dictionary defining resource.
@type resdef: dict
@param force_flag: A flag indicating if the operation should be forced despite guid/gen_id conflicts.
@type force_flag: bool
@return: A tuple of (uri, new_guid, new_gen_id).
@rtype: tuple
@raise SnapObjNotFoundError: No resource is located at uri within the store.
@raise SnapResDefGUIDError: Resource in store at uri has a different GUID than provided in guid.
@raise SnapResDefGenIDError: Resource in store at uri has different generation ID than provided in gen_id.
@raise SnapObjTypeError: The resdef value is not a dictionary.
@raise SnapValueError: The resdef contains an invalid value.
@raise SnapObjPermissionError: URL is remote and only local requests are allowed.
"""
with self._logger("update resource", uri, guid, gen_id, force_flag):
(server_url, res_uri) = self._parse_uri(uri)
if server_url is not None:
raise SnapObjPermissionError("Updating remote resource '%s' via this server is not allowed." % uri)
self._validate_resdef(resdef)
return self._store.update_resource(res_uri, guid, gen_id, resdef, force_flag)
def delete_resource(self, uri, guid, gen_id, force_flag=False):
"""
Delete a resource from the store.
First checks that the guid and gen_id values stored for the uri match those provided as arguments.
If they do not, an exception is thrown. Otherwise, the resource at uri is removed from the store.
If force_flag is given and True, a resource may be deleted without specifying the GUID or Generation ID.
If guid is given and not None, only a resource with a GUID matching the one given will be deleted. If guid
is None, any resource present will be deleted. If a resource cannot be deleted due to a mismatched
GUID when guid is not None, an SnapResDefGUIDError is raised.
@param uri: URI of resource.
@type uri: string
@param guid: GUID expected in the store.
@type guid: string
@param gen_id: Generation ID expected in the store.
@type gen_id: integer
@param force_flag: A flag indicating if the operation should be forced despite guid/gen_id conflicts.
@type force_flag: bool
@raise SnapObjNotFoundError: No resource is located at uri within the store.
@raise SnapResDefGUIDError: Resource in store at uri has a different GUID than provided in guid.
@raise SnapResDefGenIDError: Resource in store at uri has different generation ID than provided in gen_id.
@raise SnapObjPermissionError: URL is remote and only local requests are allowed.
"""
with self._logger("delete resource", uri, guid, gen_id, force_flag):
(server_url, res_uri) = self._parse_uri(uri)
if server_url is not None:
raise SnapObjPermissionError("Deletion of remote resource '%s' via this server is not allowed." % uri)
self._store.delete_resource(res_uri, guid, gen_id, force_flag)
def get_resource_dependencies(self, uri, local_only=False):
"""
Calculate the list of dependencies for a given resource URI.
If the resource identified by uri is a pipeline, a recursive algorithm will calculate all resources
the pipeline depends on. The list will contain the URI of every resource used within the pipeline. If
there are nested pipelines, their dependencies will be added to the list recursively.
If the resource identified by uri is not a pipeline, the list will be empty.
When local_only is given and True, only resources URIs local to this server will be returned.
@param uri: URI of resource to calculate dependencies of.
@type uri: str
@param local_only: Flag indicating if only local resource URIs should be returned.
@type local_only: bool
@return: List of resource URIs the given resource URI depends on.
@rtype: list
"""
(server_url, res_uri) = self._parse_uri(uri)
if server_url is not None:
raise SnapObjPermissionError("Dependency list calculation of remote resource '%s' via this server is not allowed." % uri)
dependencies = self._recurse_dependencies(uri, local_only)
return list(dependencies)
def set_resource_validated_flag(self, uri, guid, gen_id):
"""
Sets the validated flag on a resource.
Set a flag indicating the resource has been validated for execution. If the resource is later
modified or deleted, the flag will automatically return to a False value until this method
is called again.
Only metadata external to the resdef is modified by this call. The generation ID will not be affected.
@param uri: URI of resource.
@type uri: string
@param guid: GUID expected in the repository.
@type guid: string
@param gen_id: Generation ID expected in the repository.
@type gen_id: integer
@raise SnapObjNotFoundError: No resource is located at uri within the store.
@raise SnapResDefGUIDError: Resource in store at uri has a different GUID than provided in guid.
@raise SnapResDefGenIDError: Resource in store at uri has different generation ID than provided in gen_id.
@raise SnapObjPermissionError: URL is remote and only local requests are allowed.
"""
with self._logger("set resource validated", uri, gen_id, guid):
(server_url, res_uri) = self._parse_uri(uri)
if server_url is not None:
raise SnapObjPermissionError("Setting validation flag of remote resource '%s' via this server is not allowed." % uri)
self._store.set_resource_validated_flag(res_uri, guid, gen_id)
def is_resource_validated(self, uri, guid, gen_id):
"""
Check if a resource has been validated.
Checks to see if the resource has been validated as set by L{RequestProxy.set_resource_validated_flag}.
If the resource's validated flag is set, then True is returned. If not, or if the resource's GUID or GenID
do not match the provided values, False is returned.
@param uri: URI of resource.
@type uri: string
@param guid: GUID expected in the repository.
@type guid: string
@param gen_id: Generation ID expected in the repository.
@type gen_id: integer
@return: True if resource given by (uri, guid, gen_id) has its validation flag set. False if not set or
resource's GUID/GenID do not match.
"""
with self._logger("is resource validated", uri, guid, gen_id):
(server_url, res_uri) = self._parse_uri(uri)
if server_url is not None:
raise SnapObjPermissionError("Reading validation flag of remote resource '%s' via this server is not allowed." % uri)
return self._store.read_resource_validated_flag(uri, guid, gen_id)
def create_scheduler_event(self, uri, event):
"""
Store a scheduler event into the repository.
@param uri: URI for scheduler event.
@type uri: str
@param event: An dictionary representation of the scheduler event object.
@type event: dict
@raise SnapObjExistError: An event with the given URI already exists.
"""
self._store.create_scheduler_event(uri, event)
def read_scheduler_event(self, uri):
"""
Read a scheduler event from the repository.
@param uri: URI for scheduler event.
@type uri: str
@return: A dictionary representation of the scheduler event.
@rtype: dict
@raise SnapObjNotFoundError: No event with the given URI was found.
"""
return self._store.read_scheduler_event(uri)
def update_scheduler_event(self, uri, updates):
"""
Update fields of a scheduler event.
The updates parameter should be a dictionary mapping a subset of the keys from an event object
to their new values. It can also be an entirely new event dictionary. Any keys that are not
part of the updates dictionary will retain their current value in the repository.
@param uri: URI for scheduler event.
@type uri: str
@param updates: A dictionary mapping event object fields to their new value.
@type updates: dict
@raise SnapObjNotFoundError: No event with the given URI was found.
"""
return self._store.update_scheduler_event(uri, updates)
def delete_scheduler_event(self, uri):
"""
Delete a scheduler event from the repository.
@param uri: URI for the scheduler event.
@type uri: str
@raise SnapObjNotFoundError: No event with the given URI was found.
"""
self._store.delete_scheduler_event(uri)
def list_scheduler_events(self, uri_prefix=None):
"""
List scheduler events stored in the repository.
If the uri_prefix is not given, all scheduler event URIs stored in the repository are returned.
If the uri_prefix is given, it should be a string URI prefix. Any scheduler event URI stored in the
repository that begins with uri_prefix is returned.
@param uri_prefix: An optional URI prefix string to limit result.
@type uri_prefix: str
@return: List of scheduler event URIs stored in repository (and optionally beginning with uri_prefix).
@rtype: list
"""
return self._store.list_scheduler_events(uri_prefix)
def set_registry_key(self, key, value):
"""
Set the registry key to a (new) value.
Sets the registry key to the value given. If the registry key is already set, the new value will replace
the old one.
@param key: Registry key name.
@type key: str
@param value: A python value.
@type value: any
"""
self._store.set_registry_key(key, value)
def get_registry_key(self, key):
"""
Get the value of a registry key.
@param key: Registry key name.
@type key: str
@return: Python value previously set for given registry key.
@rtype: any
@raises SnapObjNotFoundError: The key is not defined in the registry.
"""
return self._store.get_registry_key(key)
def delete_registry_key(self, key):
"""
Delete the registry key and its value from the registry.
The registry key and its value are removed from the registry. The set_registry_key() method may be
later used to set a new value.
@param key: Registry key name.
@type key: str
@raises SnapObjNotFoundError: The key is not defined in the registry.
"""
self._store.delete_registry_key(key)
|