# $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:__init__.py 1764 2008-03-21 04:27:56Z dhiraj $
from StringIO import StringIO
from urllib import quote
from snaplogic.common import snap_http_lib
from snaplogic.common.snap_http_lib import urlopen,concat_paths,parseHostAndScheme,parse_host_and_path
from snaplogic import rp
from exceptions import SnapiException,SnapiHttpException
from snaplogic.common.headers import SNAPI_HEADER_PREFIX,CC_TOKEN_HEADER,STATUS_HEADER
import keys
from snaplogic.common import uri_prefix,snap_crypt
from snaplogic.server import cc_list
JSON_CONTENT_TYPE = 'application/json'
SNAPI_CONTENT_TYPE = JSON_CONTENT_TYPE
GET_HEADERS = { 'Accept' : SNAPI_CONTENT_TYPE }
POST_HEADERS = {'Accept' : SNAPI_CONTENT_TYPE,
'Content-type': SNAPI_CONTENT_TYPE }
json_rp = rp.get_rp(SNAPI_CONTENT_TYPE)
VERSION = "1.0"
__statuses = ['Failed','Stopped','Completed']
def _send_request(method, uri, data=None, custom_headers=None, cred=None, iterate_data=False):
"""
Convenience method to send an HTTP request to SnapServer.
@param method: HTTP method to use (POST, PUT, GET, etc. -- see RFC 2616)
@type method: str
@param uri: URI to send the request to
@type uri: str
@param data: Data to send in the body of the request. Note that, in full compliance with RFC
2616, we allow data to be send even on GET and DELETE requests, while some implementations
don't deal with this correctly. TODO.
@type data: obj
@param custom_headers: Additional headers to send with the request.
@type custom_headers: dict
@param cred: A 2-tuple containing (username, password)
@type cred: tuple
@param iterate_data: If set to True, the function will treat the data as a sequence and iterate
over it, and write out each entry individually to the RP stream.If False, the object is
written as a whole to the stream.
@type iterate_data: bool
@return: Response data to the request, as python objects generated by RP.
@rtype: python objects.
@raise SnapiHttpException: If an HTTP error is received.
"""
if custom_headers is None:
custom_headers = {}
if custom_headers:
new_headers = {}
# This is because NGINX, for instance, in violation of the RFC:
# http://www.w3.org/Protocols/rfc2616/rfc2616-sec2.html#sec2.1
# http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2
# does not like header fields with underscores in them
for header_field in custom_headers.keys():
new_header_field = header_field.replace('_','-')
new_headers[new_header_field] = custom_headers[header_field]
custom_headers = new_headers
if data is not None:
str = StringIO()
writer = json_rp.Writer(str)
writer.initialize()
if iterate_data:
for d in data:
writer.write(d)
else:
writer.write(data)
writer.end()
data = str.getvalue()
if method in [ "POST", "PUT" ]:
h = POST_HEADERS
else:
h = GET_HEADERS
custom_headers.update(h)
response = urlopen(method, uri, data, custom_headers, cred)
try:
headers = response.getHeaders()
try:
content_type = headers['content-type']
rp_plugin = rp.get_rp(content_type)
reader = rp_plugin.Reader(response)
retval = [obj for obj in reader]
if len(retval) == 1:
retval = retval[0]
except:
retval = ''
status = response.getStatus()
if status >= 400:
reason = response.getReason()
snap_status = headers.get(STATUS_HEADER)
exc = SnapiHttpException(status, reason, snap_status, retval)
raise exc
finally:
response.close()
return retval
def send_req(method, uri, data=None, custom_headers=None, cred=None):
"""
Convenience method to send an HTTP request to SnapServer.
@param method: HTTP method to use (POST, PUT, GET, etc. -- see RFC 2616)
@type method: str
@param uri: URI to send the request to
@type uri: str
@param data: Data to send in the body of the request. Note that, in full
compliance with RFC 2616, we allow data to be send even on GET and
DELETE requests, while some implementations don't deal with this
correctly. TODO.
@type data: obj
@param custom_headers: Additional headers to send with the request.
@type custom_headers: dict
@param cred: A 2-tuple containing (username, password)
@type cred: tuple
@return: Response data to the request, as python objects generated by RP.
@rtype: python objects.
@raise SnapiHttpException: If an HTTP error is received.
"""
return _send_request(method, uri, data, custom_headers, cred)
def send_list_req(method, uri, data=None, custom_headers=None, cred=None):
"""
Convenience method to send a list object in HTTP request to SnapServer.
@param method: HTTP method to use (POST, PUT, GET, etc. -- see RFC 2616)
@type method: str
@param uri: URI to send the request to
@type uri: str
@param data: List data to send in the body of the request. Note that, in full compliance
with RFC 2616, we allow data to be send even on GET and DELETE requests, while some
implementations don't deal with this correctly. TODO.
@type data: obj
@param custom_headers: Additional headers to send with the request.
@type custom_headers: dict
@param cred: A 2-tuple containing (username, password)
@type cred: tuple
@return: Response data to the request, as python objects generated by RP.
@rtype: python objects.
@raise SnapiHttpException: If an HTTP error is received.
"""
return _send_request(method, uri, data, custom_headers, cred, True)
SERVER_URI_MAP = {}
LOG_URI_MAP = {}
def get_server_uri_map(server_uri, cred=None):
"""
Get the map of special server URIs.
@param server_uri: URI of the server root
@type server_uri: str
@param cred: Credentials
@return: a mapping of L{keys} to the server URIs
@rtype: dict
"""
full_uri_map = _send_request('GET', server_uri, None, None, cred)
# The full URI map contains human readable descriptions. All entries
# are of the form: { 'name' : { 'uri' : <uri>, 'description' : <desc> }, ... }
# We need to translate this to a simple dictionary for lookup from name
# straight to URI.
uri_map = {}
for name in full_uri_map.keys():
uri_map[name] = full_uri_map[name]['uri']
return uri_map
def _get_server_uri(server_uri, key, cred=None):
"""
Get a URI corresponding to a specific function on the
server. If necessary, call L{server_uri_map}.
@param server_uri: Root uri of a server
@type server_uri: str
@param key: Key describing the functionality (see L{get_server_uri_map}).
@type key: str
"""
# TODO
if key == keys.SERVER_COMPONENT_LIST:
return concat_paths(server_uri, '/__snap__/component/list')
try:
server_uri_map = SERVER_URI_MAP[server_uri]
except KeyError:
server_uri_map = get_server_uri_map(server_uri, cred)
SERVER_URI_MAP[server_uri] = server_uri_map
uri = server_uri_map[key]
return uri
def get_log_uri_map(log_uri, cred=None):
"""
Get the map of special log URIs.
@param log_uri: URI of the server's log root
@type log_uri: str
@param cred: Credentials
@return: a mapping of L{keys} to the log URIs
@rtype: dict
"""
full_uri_map = _send_request('GET', log_uri, None, None, cred)
# The full URI map contains keys 'all' and "last_40_lines".
# We need to translate this to a simple dictionary for lookup from name
# staright to URI specified for "all".
uri_map = {}
for name in full_uri_map.keys():
uri_map[name] = full_uri_map[name]['all']
return uri_map
def _get_log_uri(log_uri, key, cred=None):
"""
Get a URI corresponding to a specific log on the
server. If necessary, call L{server_uri_map}.
@param log_uri: Log uri of a server
@type log_uri: str
@param key: Key for specific log file.
@type key: str
"""
try:
log_uri_map = LOG_URI_MAP[log_uri]
except KeyError:
log_uri_map = get_log_uri_map(log_uri, cred)
LOG_URI_MAP[log_uri] = log_uri_map
uri = log_uri_map[key]
return uri
def get_snapi_info():
"""
Get information about this implementation of the SnAPI Library.
@return information about the SnAPI Library. Version - the value is the version of
the SnAPI Library implements - a list of categories of API that
this implementation supports, for instance: ['info','resource_editing',
'pipeline_control']
@rtype dict
"""
return {
'version' : VERSION,
'implements' : ['info', 'resource_editing', 'pipeline_control']
}
def get_server_info(server_uri, cred=None):
"""
@param server_uri: URI of the Data Server
@type server_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@return:
a dictionary, with the following keys, whose values are the appropriate versions:
snap_server_version
server_snapi_version
@rtype: dict
"""
info_uri = _get_server_uri(server_uri, keys.SERVER_INFO, cred)
response = _send_request('GET', info_uri, None, None, cred)
return response
AUTH_URI_MAP = {}
def get_auth_uri_map(auth_uri, cred=None):
"""
Get the map of special auth URIs.
@param auth_uri: URI of the server's auth root.
@type log_uri: string
@param cred: Credentials, consisting of username and password.
@type cred: tuple
@return: A mapping of L{keys} to the auth URIs.
@rtype: dict
"""
global AUTH_URI_MAP
# We only want to have to bother the server once with a request for
# the map. Therefore, we will cache the result and return it instead
# if we have it in future invocations.
if AUTH_URI_MAP:
return AUTH_URI_MAP
else:
uri_map = _send_request('GET', auth_uri, None, None, cred)
AUTH_URI_MAP.update(uri_map)
return AUTH_URI_MAP
def auth_check(server_uri, cred=None):
"""
Perform check to see if the user is known to the system.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@return:
'KNOWN' or 'UNKNOWN' depending on whether the user exists and the correct password is used.
@rtype: str
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
response = _send_request('GET', map_uris[keys.SERVER_AUTH_CHECK]['uri'], None, None, cred)
return response
def auth_user_list(server_uri, cred=None):
"""
Return list of known users.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: List of known users.
@rtype: list
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
response = _send_request('GET', map_uris[keys.SERVER_AUTH_USER_LIST]['uri'], None, None, cred)
return response
def auth_user_entry_get(server_uri, username, cred=None):
"""
Return information about an auth user entry.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param username: Name of the user entry.
@type username: string
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: Dictionary with entry description,
containing elements such as 'username'
and 'password' (encrypted), 'description',
'email' group list and a 'genid'.
@rtype: dict
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
uri = map_uris[keys.SERVER_AUTH_USER_ENTRY]['uri'] + "/" + quote(username)
response = _send_request('GET', uri, None, None, cred)
response['uri'] = uri
return response
def auth_user_entry_create(server_uri, userdef, cred=None):
"""
Create a new user in the system.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param userdef: A dictionary defining the new user to be
created. Mandatory elements are: 'username',
'password' (in clear text). Optional elements
are 'description' and 'email'.
@type userdef: dict
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: A tuple with three elements:
1. The URI of the new entry
2. The genid of the new entry
3. A clear-text success message
In case of error an exception will be raised.
@rtype: tuple
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
# Sanity check the user definition.
mandatory_elems = [ 'name', 'password' ]
optional_elems = [ 'description', 'email' ]
allowable_elems = mandatory_elems + optional_elems
for e in mandatory_elems:
if e not in userdef:
raise SnapiException("Mandatory element '%s' is missing from user definition." % e)
for e in userdef:
if e not in allowable_elems:
raise SnapiException("Unknown element '%s' in user definition." % e)
# Assemble the user-definition that is sent to the server. Some things
# are added and changed here automatically, which is why we don't just
# send the definition as it was passed in by the user.
data = {
"name" : userdef['name'],
"password" : snap_crypt.obfuscate(userdef['password']),
"password_obfuscated" : "Yes",
"description" : userdef.get('description', ""),
"email" : userdef.get('email', ""),
}
response = _send_request('POST', map_uris[keys.SERVER_AUTH_USER_ENTRY]['uri'] + "/" + quote(data['name']), data, None, cred)
return response
def auth_user_entry_edit(server_uri, change_dict, cred=None):
"""
Modify an already existing user in the system.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param change_dict: Dictionary that contains all the elements of the entry
that should be changed. 'name' and 'genid' are mandatory
entries. For example:
{
"name" : "foo",
"genid" : "123",
"password" : "bar"
}
@type change_dict: dict
@param cred: Credentials to use for this request
@type cred: tuple (username, password)
@return: A tuple with three elements:
1. The URI of the entry
2. The new genid of the entry
3. A clear-text success message
In case of error an exception will be raised.
@rtype: tuple
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
mandatory_elems = [ 'name', 'genid' ]
optional_elems = [ 'password', 'password_obfuscated', 'description', 'email' ]
allowable_elems = mandatory_elems + optional_elems
for e in mandatory_elems:
if e not in change_dict:
raise SnapiException("Mandatory element '%s' is missing from user definition." % e)
for e in change_dict:
if e not in allowable_elems:
raise SnapiException("Unknown element '%s' in user definition." % e)
username = change_dict["name"]
# Make sure that if the password is present it is obfuscated before
# it goes over the wire.
if "password" in change_dict:
if "password_obfuscated" not in change_dict or change_dict["password_obfuscated"].lower() == "no":
change_dict["password"] = snap_crypt.obfuscate(change_dict["password"])
change_dict["password_obfuscated"] = "Yes"
response = _send_request('PUT', map_uris[keys.SERVER_AUTH_USER_ENTRY]['uri'] + "/" + quote(username), change_dict, None, cred)
return response
def auth_user_entry_delete(server_uri, username, cred=None):
"""
Delete information about an auth user entry.
Note that if a user is deleted than the genid of any group (!)
that this user was a member in changes as well, since in effect
the group definition has changed, too.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param username: Name of the auth entry.
@type username: string
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: Success message. In case of failure an exception is raised.
@rtype: string
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
response = _send_request('DELETE', map_uris[keys.SERVER_AUTH_USER_ENTRY]['uri'] + "/" + quote(username), None, None, cred)
return response
def auth_group_list(server_uri, cred=None):
"""
Return list of known groups.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: Dictionary of known groups. Each entry is a
contains the 'uri' and 'description' element
of the group. The groupname is the index.
@rtype: dict
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
response = _send_request('GET', map_uris[keys.SERVER_AUTH_GROUP_LIST]['uri'], None, None, cred)
return response
def auth_group_entry_get(server_uri, groupname, cred=None):
"""
Return information about an auth group entry.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param username: Name of the group entry.
@type username: string
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: Dictionary with entry description,
containing elements such as 'groupname'
user list and genid.
@rtype: dict
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
uri = map_uris[keys.SERVER_AUTH_GROUP_ENTRY]['uri'] + "/" + quote(groupname)
response = _send_request('GET', uri, None, None, cred)
response['uri'] = uri
return response
def auth_group_entry_create(server_uri, groupdef, cred=None):
"""
Create a new group in the system.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param userdef: A dictionary defining the new group to be
created. Mandatory elements are: 'groupname',
'users' (a list that may be empty). Optional
element is 'description'.
@type userdef: dict
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: A tuple with three elements:
1. The URI of the new entry
2. The genid of the new entry
3. A clear-text success message
In case of error an exception will be raised.
@rtype: tuple
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
# Sanity check the group definition.
mandatory_elems = [ 'name', 'users' ]
optional_elems = [ 'description' ]
allowable_elems = mandatory_elems + optional_elems
for e in mandatory_elems:
if e not in groupdef:
raise SnapiException("Mandatory element '%s' is missing from group definition." % e)
for e in groupdef:
if e not in allowable_elems:
raise SnapiException("Unknown element '%s' in group definition." % e)
response = _send_request('POST', map_uris[keys.SERVER_AUTH_GROUP_ENTRY]['uri'] + "/" + quote(groupdef['name']), groupdef, None, cred)
return response
def auth_group_entry_edit(server_uri, change_dict, cred=None):
"""
Modify an already existing group in the system.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param change_dict: Dictionary that contains all the elements of the entry
that should be changed. 'groupname' and 'genid' are mandatory
entries. For example:
{
"name" : "foo",
"genid" : "123",
"description" : "bar",
"users" : [ 'user1', 'user2' ]
}
Note that the 'users' list may be empty, resulting in an
empty group.
If a user is removed or added to a group then the genid of the
user entry (!) also changes (since the user in effect has changed).
@type change_dict: dict
@param cred: Credentials to use for this request
@type cred: tuple (username, password)
@return: A tuple with three elements:
1. The URI of the entry
2. The new genid of the entry
3. A clear-text success message
In case of error an exception will be raised.
@rtype: tuple
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
mandatory_elems = [ 'name', 'genid' ]
optional_elems = [ 'users', 'description' ]
allowable_elems = mandatory_elems + optional_elems
for e in mandatory_elems:
if e not in change_dict:
raise SnapiException("Mandatory element '%s' is missing from group definition." % e)
for e in change_dict:
if e not in allowable_elems:
raise SnapiException("Unknown element '%s' in group definition." % e)
groupname = change_dict["name"]
response = _send_request('PUT', map_uris[keys.SERVER_AUTH_GROUP_ENTRY]['uri'] + "/" + quote(groupname), change_dict, None, cred)
return response
def auth_group_entry_delete(server_uri, groupname, cred=None):
"""
Delete information about an auth group entry.
Note that if a group is deleted than the genid of any user (!)
that was a member of this group changes as well, since in effect
the user definition has changed, too.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param username: Name of the auth entry.
@type username: string
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: Success message. In case of failure an exception is raised.
@rtype: string
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
response = _send_request('DELETE', map_uris[keys.SERVER_AUTH_GROUP_ENTRY]['uri'] + "/" + quote(groupname), None, None, cred)
return response
def auth_acl_list(server_uri, cred=None):
"""
Return list of known ACLs.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: List of known groups.
@rtype: list
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
response = _send_request('GET', map_uris[keys.SERVER_AUTH_ACL_LIST]['uri'], None, None, cred)
return response
def auth_acl_entry_get(server_uri, name, cred=None):
"""
Return information about an auth ACL entry.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param name: Name of the ACL entry. This is a relative
path on the server, always starting with
a '/' character.
@type name: string
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: Dictionary with entry description,
containing elements such as 'name', 'rules',
'description', etc.
@rtype: dict
"""
if not name.startswith("/"):
raise SnapiException("The name of the ACL '%s' has to start with a '/'." % name)
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
uri = map_uris[keys.SERVER_AUTH_ACL_ENTRY]['uri'] + quote(name)
response = _send_request('GET', uri, None, None, cred)
response['uri'] = uri
return response
def auth_acl_entry_create(server_uri, acldef, cred=None):
"""
Create a new ACL rule in the system.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param acldef: A dictionary defining the new ACL to be
created. Mandatory element are: 'name'
(which is always a path on the server,
starting with "/") and rules (which is a
list of textual rule definitions. This
list can be empty.) Optional element is
'description'.
@type acldef: dict
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: A tuple with three elements:
1. The URI of the new entry
2. The genid of the new entry
3. A clear-text success message
In case of error an exception will be raised.
@rtype: tuple
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
# Sanity check the group definition.
mandatory_elems = [ 'name', 'rules' ]
optional_elems = [ 'description' ]
allowable_elems = mandatory_elems + optional_elems
for e in mandatory_elems:
if e not in acldef:
raise SnapiException("Mandatory element '%s' is missing from ACL definition." % e)
for e in acldef:
if e not in allowable_elems:
raise SnapiException("Unknown element '%s' in ACL definition." % e)
if not acldef['name'].startswith("/"):
raise SnapiException("The name of the ACL '%s' has to start with a '/'." % acldef['name'])
response = _send_request('POST', map_uris[keys.SERVER_AUTH_ACL_ENTRY]['uri'] + quote(acldef['name']), acldef, None, cred)
return response
def auth_acl_entry_edit(server_uri, change_dict, cred=None):
"""
Modify an already existing ACL in the system.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param change_dict: Dictionary that contains all the elements of the entry
that should be changed. 'name' and 'genid' are mandatory
entries. For example:
{
"name" : "/foo/bar",
"genid" : "123",
"description" : "bar",
"rules" : [ 'allow user foo permissions read' ]
}
Note that the 'rules' list may be empty, resulting in an
ACL without rules.
@type change_dict: dict
@param cred: Credentials to use for this request
@type cred: tuple (username, password)
@return: A tuple with three elements:
1. The URI of the entry
2. The new genid of the entry
3. A clear-text success message
In case of error an exception will be raised.
@rtype: tuple
"""
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
mandatory_elems = [ 'name', 'genid' ]
optional_elems = [ 'rules', 'description' ]
allowable_elems = mandatory_elems + optional_elems
for e in mandatory_elems:
if e not in change_dict:
raise SnapiException("Mandatory element '%s' is missing from ACL definition." % e)
for e in change_dict:
if e not in allowable_elems:
raise SnapiException("Unknown element '%s' in ACL definition." % e)
name = change_dict["name"]
if not name.startswith("/"):
raise SnapiException("The name of the ACL '%s' has to start with a '/'." % name)
response = _send_request('PUT', map_uris[keys.SERVER_AUTH_ACL_ENTRY]['uri'] + quote(name), change_dict, None, cred)
return response
def auth_acl_entry_delete(server_uri, name, cred=None):
"""
Delete information about an auth ACL entry.
@param server_uri: The root URI of the Data Server
@type server_uri: str
@param name: Name of the auth entry. The name always
has to start with a '/'.
@type name: string
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: Success message. In case of failure an exception is raised.
@rtype: string
"""
if not name.startswith("/"):
raise SnapiException("The name of the ACL '%s' has to start with a '/'." % name)
map_uris = get_auth_uri_map(server_uri + uri_prefix.AUTH, cred)
response = _send_request('DELETE', map_uris[keys.SERVER_AUTH_ACL_ENTRY]['uri'] + quote(name), None, None, cred)
return response
def list_resources(server_uri, uri_list=None, cred=None):
"""
List resources available at the given server.
@param uri: URI of the Data Server.
@type uri: str
@param uri_list: if provided, retrieve info only for resources that
match this list
@type uri_list: sequence
@return: a dict keyed by a URI, whose values are a dictionary containing gen_id, guid and description of
the resource
@rtype: dict
"""
list_uri = _get_server_uri(server_uri, keys.SERVER_RESOURCE_LIST, cred)
method = 'GET'
data = None
if uri_list:
method = 'POST'
data = uri_list
result = _send_request(method, list_uri, data, None, cred)
if result:
result2 = {}
for res_uri in result.keys():
if res_uri.startswith('/'):
abs_res_uri = concat_paths(server_uri, res_uri)
else:
abs_res_uri = res_uri
result2[abs_res_uri] = result[res_uri]
result = result2
return result
def summarize_resources(server_uri, uri_list=None, details=None, cred=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.
@param server_uri: URI of the Data Server.
@type server_uri: str
@param uri_list: A list of relative 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.
@return: A dictionary of dictionaries describing the resources in the store.
@rtype: dict
"""
summary_uri = _get_server_uri(server_uri, keys.SERVER_RESOURCE_SUMMARY, cred)
method = 'GET'
data = None
if uri_list or details:
method = 'POST'
data = [uri_list, details]
result = _send_request(method, summary_uri, data, None, cred)
if result:
result2 = {}
for res_uri in result.keys():
if res_uri.startswith('/'):
abs_res_uri = concat_paths(server_uri, res_uri)
else:
abs_res_uri = res_uri
result2[abs_res_uri] = result[res_uri]
result = result2
return result
def upgrade_resources(server_uri, uri_list, credentials=None):
"""
Upgrade specified resources
@param server_uri: URI of the Data Server.
@type server_uri: str
@param resource_list: URIs of the resources to be upgrade
@type resource_list: list
@param credentials: Credentials for the request, consisting of a username/password
tuple.
@type credentials: tuple
@return:
a dictionary with the upgrade results where:
- key is the resource URI
- value is a tuple consisting of three objects:
(needed_upgrade, upgrade_succeeded, error_message)
@rtype: dict
"""
# If uri list is None, upgrade the entire repository
if uri_list is None:
# Sending in an empty list upgrades the entire repository
uri_list = []
# If one URI given instead of a list, convert to a list
if uri_list is not None and type(uri_list) is not list and type(uri_list) is not tuple:
uri_list = [ uri_list ]
upgrade_uri = _get_server_uri(server_uri, keys.SERVER_RESOURCE_UPGRADE, credentials)
return _send_request("POST", upgrade_uri, uri_list, None, credentials)
def get_server_logs(server_uri, cred=None):
"""
Parameter:
server_uri - URI of the Data Server
"""
log_uri = _get_server_uri(server_uri, keys.SERVER_LOGFILES, cred)
result = _send_request('GET', log_uri, None, None, cred)
return result
def get_statistics(server_uri, cred=None):
"""
Get various statistics e.g. pipeline execution-related numbers from the main server and component container(s).
@param server_uri: Snaplogic server URI, e.g. http://host:8088
@type server_uri: str
@param cred: Credentials to make the request, as a tuple (username, password).
@type cred: tuple
"""
uri = _get_server_uri(server_uri, keys.SERVER_STATS, cred)
return _send_request('GET', uri, None, None, cred)
def get_resource_logs(rid, resource_uri, include_facilities=None, exclude_facilities=None, cred=None,
min_level=None, max_level=None, start_date=None, end_date=None, last_lines=None):
"""
Fetch all log messages related to a particular execution of a resource.
@param rid: Runtime id of the resource that was executed.
@type rid: str
@param uri: URI of the server.
@type uri: str
@param include_facilties: List of log facility names to include (optional). The response will be limited
to messages from these facilities.
@type include_facilties: list
@param exclude_facilities: List of log facility names to exclude (optional). The response will exclude
log messages from these facilities.
@param cred: Credentials to make the log request, as a tuple (username, password).
@type cred: tuple
@param min_level: Minimum log level messages to return.
@type min_level: str
@param max_level: Maximum log level messages to return.
@type max_level: str
@param start_date: Earliest timestamps of log message to return. Date must be expressed in ISO 8601 format string
YYYY-MM-DDTHH:MM:SS
@type start_date: str
@param end_date: Latest timestamp of log messages to return. Date must be expressed in ISO 8601 format string
YYYY-MM-DDTHH:MM:SS
@type end_date: str
@param last_lines: Number of last few lines to return (sort of a tail)
@type min_level: int
@return: List of log messages retrieved.
@rtype: list
"""
# First, extract the server URI from the resource uri.
(scheme, netloc, url, p, q, f) = snap_http_lib.parse_uri(resource_uri)
server_uri = snap_http_lib.unparse_uri((scheme, netloc, "", "", "", ""))
# Fetch the log URI from the server and then fetch the pipeline log URI from that location.
log_uri = _get_server_uri(server_uri, keys.SERVER_LOGFILES, cred)
pipeline_log_uri = _get_log_uri(log_uri, keys.PIPELINE_LOGFILE, cred)
# Now process the params for the log entry request.
if exclude_facilities:
final_list = [ "-" + f for f in exclude_facilities]
else:
final_list = []
if include_facilities is not None:
final_list += include_facilities
facility_arg = ",".join(final_list)
params = {"rid" : rid}
if len(facility_arg) > 0:
params["facility"] = facility_arg
if min_level is not None:
params["min_level"] = min_level
if max_level is not None:
params["max_level"] = max_level
if start_date is not None:
params["start_date"] = start_date
if end_date is not None:
params["end_date"] = end_date
if last_lines is not None:
params["last_lines"] = last_lines
pipeline_log_uri = snap_http_lib.add_params_to_uri(pipeline_log_uri, params)
message_list = send_req("GET", pipeline_log_uri, cred=cred)
return message_list
def get_runtime_status(server_uri, cred = None):
"""
Get all runtime status information available on the server
(this is similar to visiting the http://localhost:8088/__snap__/runtime/status page)
"""
uri = _get_server_uri(server_uri, keys.SERVER_RUNTIME_STATUS, cred)
return send_req("GET", uri, cred = cred)
def list_components(server_uri, cred=None, custom_hdr=None):
"""
List components available at the given server.
@param uri: URI of the Data Server or of the Component Container.
@type uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@param custom_hdr: HTTP Headers to be passed with the request.
@type custom_hdr: dict
@return:
a list of available components, as a list of dictionaries, each containing values of the
following keys for every component:
name - user-visible component name
uri - component URI
description - short overview of the component's functionality
capabilities - same as capabilities as a read-only field of a ResDef.
More keys can be added if desired. Here is a sample return value from this function:
@rtype: list
"""
list_uri = _get_server_uri(server_uri, keys.SERVER_COMPONENT_LIST, cred)
if custom_hdr is None:
cc_token = cc_list.get_token_by_uri(list_uri)
if cc_token is not None:
custom_hdr = { CC_TOKEN_HEADER : cc_token }
else:
custom_hdr = None
result = _send_request("GET", list_uri, None, custom_hdr, cred)
return result
def get_component_template(template_uri, cred=None, custom_hdr=None):
"""
Get a component template.
@param template_uri: uri that handles get_component_template functionality (client knows
this after a call to L{list_components}).
@type template_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@param custom_hdr: HTTP Headers to be passed with the request.
@type custom_hdr: dict
@return:
a ResDef structure for the specified component.
"""
result = _send_request("POST", template_uri, {}, custom_hdr, cred)
return result
def get_pipeline_template(uri, cred=None):
"""
Special case of L{get_component_template}, for getting a pipeline template.
@param uri: Server URI
@type uri: str
"""
from snaplogic.snapi_base.resdef import PipelineResDef
# TODO: get it from the server in fact
return PipelineResDef().dict
def read_resource(uri, cred=None, custom_hdr=None):
"""
Retrieve the resource.
@param uri: URI of the resource
@type uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@param custom_hdr: HTTP Headers to be passed with the request.
@type custom_hdr: dict
@return: a dictionary, keyed as follows:
resdef : a ResDef structure for the resource located at the specified URI, and
guid : the GUID of the resource
genid : the current Generation ID of the resource
@rtype: dict
@raise: SnapiException if the resource at the provided URI is not found.
"""
result = _send_request('GET', uri, None, custom_hdr, cred)
if result[keys.SUCCESS]:
retval = result[keys.SUCCESS]
rel_uri = retval.keys()[0]
value = retval[rel_uri]
retval[uri] = value
retval.__delitem__(rel_uri)
return retval
else:
raise SnapiHttpException(404, 'NOT FOUND', None, result[keys.ERROR])
def read_resources(uris, cred=None):
"""
@param uris: list of URIs
@type uris: list
@param cred: Credentials to use
@type cred: tuple (username, password)
@return: a dictionary, with two keys, SUCCESS and ERROR. Under SUCCESS, a dictionary keyed by URIs of successfully
read resources, with values same as result of L{read_resource}. Under ERROR, a dictionary keyed by URIs of resources
with errors, with SnapiException as values.
@rtype: dictionary
"""
requests = {}
retval = {
keys.SUCCESS : {},
keys.ERROR : {}
}
if not isinstance(uris, list):
raise SnapiException("The 'uris' param must be set to a list")
for uri in uris:
(host,path) = parse_host_and_path(uri)
try:
res_uris = requests[host]
except KeyError:
requests[host] = []
res_uris = requests[host]
res_uris.append(path)
for server in requests.keys():
try:
read_uri = _get_server_uri(server, keys.SERVER_RESOURCE_READ, cred)
server_response = _send_request('POST', read_uri, requests[server], None, cred)
absolute_response = {keys.SUCCESS : {}, keys.ERROR : {}}
for res_uri in server_response[keys.SUCCESS].keys():
absolute_response[keys.SUCCESS][concat_paths(server, res_uri)] = server_response[keys.SUCCESS][res_uri]
for res_uri in server_response[keys.ERROR].keys():
absolute_response[keys.ERROR][concat_paths(server, res_uri)] = server_response[keys.ERROR][res_uri]
server_response = absolute_response
retval.update(server_response)
except SnapiException, e:
if len(server) == 1:
raise
for res_uri in requests[server]:
retval[keys.ERROR][concat_paths(server, res_uri)] = e
return retval
def write_resource(resdef, guid=None, gen_id=None, uri=None, force_flag=False, cred=None, custom_hdr=None):
"""
Saves the ResDef under the given URI. The save is successful if the guid and gen_id parameters match the current ones associated
with the URI (or if the URI does not exist and the guid and gen_id specified are None).
If the force_flag is given and True, a resource already present at the given URI will be overwritten. If
the guid is None, then any resource that exists at that URI will be overwritten. If guid is not None then
the GUID stored within the repository must match the one given or a conflict error will occur. In both cases,
the gen_id is ignored.
NOTE: This does not validate anything; you are allowed to save an invalid resource.
@param resdef: a ResDef structure
@type resdef: dict
@param guid: L{keys.GUID} of the resource received via the last L{write_resource} or L{read_resource}, or
None if this is the first write.
@type guid: str
@param gen_id: L{keys.GEN_ID} of the resource received via the last L{write_resource} or L{read_resource}, or
None if this is the first write.
@type gen_id: int
@param uri: URI under which to save this resource, or None to use an automatically generated URI.
@type uri: str
@param force_flag: Flag to indicate a resource be overwritten if it exists.
@type force_flag: bool
@param cred: Credentials to use
@type cred: tuple (username, password)
@param custom_hdr: HTTP Headers to be passed with the request.
@type custom_hdr: dict
@return: a dictionary containing the following:
L{keys.GUID} -- if this is a newly created resource (L{keys.GUID} sent was None), a new L{keys.GUID}. Otherwise,
the same L{keys.GUID} as was sent
L{keys.GEN_ID} - incremented gen_id (or newly created one - 0) if the gen_id sent was None
uri - under which the resource was created (if uri parameter was provided, the same URI is returned;
otherwise the automatically generated URI is returned).
@rtype: dict
@raise SnapiException: if:
- the IDs do not match as described above. This means there is a conflict -
another client has modified or deleted the resource. It is up to the client now to perform a
read_resource() operation to determine the current state of the resource and act accordingly.
- saving under this URI is disallowed (it is an internally used URI).
- This is a request for an update (GUID and GEN_ID are not None) but the resource under the URI is
not found
"""
# TODO... need to do proper thing for relative uri...
result = _send_request('PUT', uri, {keys.RESDEF : resdef,
keys.GUID : guid,
keys.GEN_ID : gen_id,
keys.FORCE: force_flag},
custom_hdr,
cred)
# TODO temp hack
if result[keys.URI].startswith('/'):
result[keys.URI] = uri
return result
def delete_resource(guid, gen_id, uri, force_flag=False, cred=None, custom_hdr=None):
"""
Deletes a resource at the specified URI if the ID matches.
If the force_flag is given and True, a resource already present at the given URI will be overwritten. If
the guid is None, then any resource that exists at that URI will be overwritten. If guid is not None then
the GUID stored within the repository must match the one given or a conflict error will occur. In both cases,
the gen_id is ignored.
@param guid: L{keys.GUID} of the resource received via the last L{write_resource} or L{read_resource}.
@type guid: str
@param gen_id: L{keys.GEN_ID} of the resource received via the last L{write_resource} or L{read_resource}.
@type gen_id: int
@param uri: URI under which to save this resource (None to use an automatically generated URI).
@type uri: str
@param force_flag: Flag to force deletion of resource depending on value of guid and ignoring gen_id.
@type force_flag: bool
@param cred: Credentials to use
@type cred: tuple (username, password)
@param custom_hdr: HTTP Headers to be passed with the request.
@type custom_hdr: dict
@raise exception: when the IDs do not match (see the behavior of L{write_resource}).
"""
hdr = {
SNAPI_HEADER_PREFIX + keys.GUID : guid,
SNAPI_HEADER_PREFIX + keys.GEN_ID : gen_id,
SNAPI_HEADER_PREFIX + keys.FORCE : force_flag
}
if custom_hdr is not None:
hdr.update(custom_hdr)
result = _send_request('DELETE', uri, None, hdr, cred)
return result
def get_resource_dependencies(uri, local_only=False, cred=None):
"""
Get the list of URIs that the resource depends on.
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
@raise SnapiException: There is no resource at uri.
"""
if local_only:
dependencies_uri = uri + keys.RESOURCE_PROPERTY_DEPENDENCIES_LOCAL
else:
dependencies_uri = uri + keys.RESOURCE_PROPERTY_DEPENDENCIES
return _send_request('GET', dependencies_uri, cred=cred)
def validate(resdef, validate_uri, cred=None, custom_hdr=None):
"""
Validates the ResDef.
@param resdef: a ResDef dictionary
@type resdef: dict
@param validate_uri: uri that handles suggest_resource_values functionality (client knows
this after a call to L{list_components}).
@type validate_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@param custom_hdr: HTTP Headers to be passed with the request.
@type custom_hdr: dict
@return: a ResDef containing errors and modifications to the sent ResDef, if any. For instance,
a suggest_resource_values() call may result in a creation of a new view.
@return: the ResDef structure, with errors if warranted.
@rtype: dict
"""
validated = _send_request('POST', validate_uri, resdef, custom_hdr, cred)
return validated
def validate_pipeline(resdef, server_uri, cred=None, custom_hdr=None):
"""
Similar to L{validate} for a Pipeline.
"""
validate_uri = _get_server_uri(server_uri, keys.SERVER_PIPELINE_VALIDATE, cred)
retval = validate(resdef, validate_uri, cred, custom_hdr)
return retval
def suggest_pipeline_values(resdef, server_uri, params={}, cred=None, custom_hdr=None):
"""
Similar to L{suggest_resource_values} for a Pipeline.
"""
suggest_uri = _get_server_uri(server_uri, keys.SERVER_PIPELINE_SUGGEST, cred)
retval = suggest_resource_values(resdef, suggest_uri, params, cred)
return retval
def suggest_resource_values(resdef, suggest_uri, params=None, cred=None, custom_hdr=None):
"""
Performs a suggest_resource_values scenario.
@param resdef: a ResDef dictionary
@type resdef: dict
@param suggest_uri: uri that handles suggest_resource_values functionality (client knows
this after a call to L{list_components}).
@type suggest_uri: str
@param cred: Credentials to use
@type cred: tuple (username, password)
@param custom_hdr: HTTP Headers to be passed with the request.
@type custom_hdr: dict
@return: a ResDef containing errors and modifications to the sent ResDef, if any.
For instance, a suggest_resource_values() call may result in a creation of a new view.
See also description of autoFill() method in Component Container API spec.
"""
if params is None:
params = {}
suggested = _send_request('POST', suggest_uri, [resdef, params], custom_hdr, cred, True)
return suggested
def scheduler_list_events(server_uri, cred=None):
"""
Parameter:
server_uri - URI of the Data Server
"""
sched_uri = _get_server_uri(server_uri, keys.SERVER_SCHEDULER, cred)
retval = _send_request('GET', sched_uri, None, None, cred)
return retval
def scheduler_list_event(event_uri, cred=None):
"""
Parameter:
event_uri - URI of the event
"""
retval = _send_request('GET', event_uri, None, None, cred)
return retval
def scheduler_run_event(event_start_uri, cred=None):
"""
Parameter:
event_start_uri - URI of the event with start argument appended
"""
retval = _send_request('GET', event_start_uri, None, None, cred)
return retval
def scheduler_create_event(server_uri, params, cred=None):
"""
Parameter:
server_uri - URI of the Data Server
params - dictionary containing the event definition
"""
sched_uri = _get_server_uri(server_uri, keys.SERVER_SCHEDULER, cred)
retval = _send_request('POST', sched_uri, params, None, cred)
return retval
def scheduler_delete_event(event_uri, cred=None):
"""
Parameter:
event_uri - URI of the event to delete
"""
retval = _send_request('DELETE', event_uri, None, None, cred)
return retval
def scheduler_update_event(event_uri, params, cred=None):
"""
Parameter:
event_uri - URI of the event
params - dictionary containing the event definition
"""
retval = _send_request('PUT', event_uri, params, None, cred)
return retval
def get_notification_types(server_uri, cred=None):
"""
Parameter:
server_uri - URI of the Data Server
"""
uri = _get_server_uri(server_uri, keys.SERVER_NOTIFICATION, cred)
retval = _send_request('GET', uri, None, None, cred)
return retval
def diff_member_resource(server_uri, member_uri, member_res, credentials=None):
"""
Get the diff of a potentially stale member resource against the image of the resource def in repository.
The response returns tow things
1) The refreshed image of the member resource that can be directly set into the pipeline resdef.
2) The diff that was found from the stale image sent to the server.
@param server_uri: URI of the Data Server.
@type server_uri: str
@param resource_list: URIs of the resources to be upgrade
@type resource_list: list
@param credentials: Credentials for the request, consisting of a username/password
tuple.
@type credentials: tuple
"""
diff_uri = _get_server_uri(server_uri, keys.RESOURCE_DIFF, credentials)
return _send_request("POST", diff_uri, [member_uri, member_res], None, credentials)
class ValidationDict(dict):
"""
A print friendly derivation of dictionary, customized for validation errors.
"""
def print_with_context(self, entry, stk, message_list, print_context = True):
if keys.ERROR_MESSAGE in entry and entry[keys.ERROR_MESSAGE]:
if entry[keys.LABEL] is not None:
l = stk + [entry[keys.LABEL]]
else:
l = stk
context_of_error = " / ".join(l)
if context_of_error and print_context:
message_list.append( "%-60s => %s" % (entry[keys.ERROR_MESSAGE], context_of_error))
else:
message_list.append(entry[keys.ERROR_MESSAGE])
if keys.LIST_ERROR in entry:
for i in range(len(entry[keys.LIST_ERROR])):
if entry[keys.LIST_ERROR][i] is not None:
if entry[keys.LABEL] is not None:
label = entry[keys.LABEL]
else:
label = "List"
stk.append("%s[%s]" % (label, i))
self.print_with_context(entry[keys.LIST_ERROR][i], stk, message_list, print_context)
stk.pop()
elif keys.DICT_ERROR in entry and entry[keys.DICT_ERROR] is not None:
for k in entry[keys.DICT_ERROR]:
if entry[keys.DICT_ERROR] is not None:
if entry[keys.LABEL] is not None:
label = entry[keys.LABEL]
else:
label = "Dictionary"
stk.append("%s[%s]" % (label, k))
self.print_with_context(entry[keys.DICT_ERROR][k], stk, message_list, print_context)
stk.pop()
else:
return
def __str__(self):
"""Returns a brief list of error message strings in the dictionary."""
message_list = []
stk = []
self.print_with_context(self, stk, message_list, False)
message_list.sort()
s = "\n".join(message_list)
return s
def verbose(self):
"""Returns a more verbose string containing error messages and the location of the message in the resdef."""
message_list = []
stk = []
self.print_with_context(self, stk, message_list, True)
message_list.sort()
s = "\n".join(message_list)
return s
def has_messages(self, expected_list):
"""
Method for comparing with an expected list of error messages.
@param expected_list: List of expected message strings.
@type expected_list: list
@return: None if the expected list matches the mesages in the error object, else, a tuple of 2 lists
is returned- (list of expected messages that were not found, list of unexpected messages that were found).
@rtype: 2-tuple or None
"""
elist = list(expected_list)
unexpected_messages = []
message_list = []
stk = []
self.print_with_context(self, stk, message_list, False)
for m in message_list:
if m in elist:
elist.remove(m)
else:
unexpected_messages.append(m)
if len(unexpected_messages) > 0 or len(elist) > 0:
elist.sort()
unexpected_messages.sort()
return (elist, unexpected_messages)
else:
return None
if __name__ == "__main__":
print read_resources(["http://localhost:8088/resdef/writer_test/writer"])
|