You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oodt.apache.org by ma...@apache.org on 2017/07/17 23:06:28 UTC
[73/88] [abbrv] oodt git commit: OODT-753: Move FM and WM Python APIs
into agility component by kelly, mattmann.
OODT-753: Move FM and WM Python APIs into agility component by kelly, mattmann.
Project: http://git-wip-us.apache.org/repos/asf/oodt/repo
Commit: http://git-wip-us.apache.org/repos/asf/oodt/commit/d329e1ab
Tree: http://git-wip-us.apache.org/repos/asf/oodt/tree/d329e1ab
Diff: http://git-wip-us.apache.org/repos/asf/oodt/diff/d329e1ab
Branch: refs/heads/feature/zookeeper-config
Commit: d329e1ab90a0aacd601859e37e3acac978d68c94
Parents: 2fc7bce
Author: Chris Mattmann <ma...@apache.org>
Authored: Sun Jul 16 14:03:23 2017 -0700
Committer: Chris Mattmann <ma...@apache.org>
Committed: Sun Jul 16 14:03:23 2017 -0700
----------------------------------------------------------------------
CHANGES.txt | 2 +
agility/oodt/conf/fm.conf | 43 ++
agility/oodt/conf/setclasspath.sh | 17 +
agility/oodt/fm.py | 554 +++++++++++++++++++
agility/oodt/metadata.py | 29 +
agility/oodt/workflow.py | 134 +++++
filemgr/src/main/python/fm.conf | 43 --
filemgr/src/main/python/fm.py | 554 -------------------
filemgr/src/main/python/setclasspath.sh | 17 -
.../main/python/xmlrpc-wrapper/oodt_metadata.py | 29 -
.../xmlrpc-wrapper/oodt_workflow_xmlrpc.py | 134 -----
11 files changed, 779 insertions(+), 777 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 7ceb353..c4be36b 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -3,6 +3,8 @@ Apache OODT Change Log
Release 1.1 - Current Development
+* OODT-753 Move FM and WM Python APIs into "agility" component (kelly, mattmann)
+
* OODT-563 Task editing and Workflow execution (varun, mattmann)
* OODT-212 Rollback capability for Workflows (sherylj via mattmann)
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/conf/fm.conf
----------------------------------------------------------------------
diff --git a/agility/oodt/conf/fm.conf b/agility/oodt/conf/fm.conf
new file mode 100644
index 0000000..dae6438
--- /dev/null
+++ b/agility/oodt/conf/fm.conf
@@ -0,0 +1,43 @@
+# encoding: utf-8
+#/*
+# * Licensed to the Apache Software Foundation (ASF) under one or more
+# * contributor license agreements. See the NOTICE file distributed with
+# * this work for additional information regarding copyright ownership.
+# * The ASF licenses this file to You under the Apache License, Version 2.0
+# * (the "License"); you may not use this file except in compliance with
+# * the License. You may obtain a copy of the License at
+# *
+# * http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+
+[index]
+# The path gives the name of the directory where the Lucene search engine
+# will build its indexes. The path refers to a directory; it will be created
+# if it doesn't exist. If it does exist, it must refer to an
+# existing Lucene index with correct sub-files present. So, don't make this
+# directory manually, or you'll be confused by error messages!
+path = @INDEX_PATH@
+pageSize = 20
+
+[policies]
+# The 'repo' and 'validation' are comma-separated lists of URLs. Typically, they
+# contain a single "file:" style URL that names a directory with policy XML files
+# within. The 'user' identifies the user database. It'll be created if it
+# doesn't exist.
+repo = @REPO_POLICY_URLS@
+validation = @VALIDATION_POLICY_URLS@
+user = @USER_DB_PATH@
+
+[factories]
+# There's no need to change anything here unless you really know what you're doing,
+# and trust me, you don't. Don't take it personally. It's just a fact.
+catalog = org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory
+repository = org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory
+datatransfer = org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory
+validation =org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/conf/setclasspath.sh
----------------------------------------------------------------------
diff --git a/agility/oodt/conf/setclasspath.sh b/agility/oodt/conf/setclasspath.sh
new file mode 100755
index 0000000..1fbfc48
--- /dev/null
+++ b/agility/oodt/conf/setclasspath.sh
@@ -0,0 +1,17 @@
+#/*
+# * Licensed to the Apache Software Foundation (ASF) under one or more
+# * contributor license agreements. See the NOTICE file distributed with
+# * this work for additional information regarding copyright ownership.
+# * The ASF licenses this file to You under the Apache License, Version 2.0
+# * (the "License"); you may not use this file except in compliance with
+# * the License. You may obtain a copy of the License at
+# *
+# * http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+setenv CLASSPATH ${HOME}/.m2/repository/org.apache.oodt.cas-filemgr/1.7.0-dev/cas-filemgr-1.7.0-dev.jar:${HOME}/.m2/repository/xmlrpc/xmlrpc/2.1-dev/xmlrpc-2.1-dev.jar:${HOME}/.m2/repository/org.apache.oodt.cas-metadata/1.5.0/cas-metadata-1.5.0.jar:${HOME}/.m2/repository/org/apache/lucene/lucene-core/2.0.0/lucene-core-2.0.0.jar:${HOME}/.m2/repository/jug/jug-asl/2.0.0/jug-asl-2.0.0.jar
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/fm.py
----------------------------------------------------------------------
diff --git a/agility/oodt/fm.py b/agility/oodt/fm.py
new file mode 100644
index 0000000..5eed046
--- /dev/null
+++ b/agility/oodt/fm.py
@@ -0,0 +1,554 @@
+#!/usr/bin/env python
+# encoding: utf-8
+#/*
+# * Licensed to the Apache Software Foundation (ASF) under one or more
+# * contributor license agreements. See the NOTICE file distributed with
+# * this work for additional information regarding copyright ownership.
+# * The ASF licenses this file to You under the Apache License, Version 2.0
+# * (the "License"); you may not use this file except in compliance with
+# * the License. You may obtain a copy of the License at
+# *
+# * http://www.apache.org/licenses/LICENSE-2.0
+# *
+# * Unless required by applicable law or agreed to in writing, software
+# * distributed under the License is distributed on an "AS IS" BASIS,
+# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# * See the License for the specific language governing permissions and
+# * limitations under the License.
+# */
+'''
+CAS Filemgr Python Server
+
+This is the Python server API for an authenticated Catalog and Archive System. It uses the
+OODT Catalog and Archive System as its core, adding user/password-based authentication
+and role-based authorization.
+'''
+
+import sys, os, os.path
+import getopt, sha, pickle
+
+from ConfigParser import ConfigParser
+from org.apache.oodt.cas.filemgr.system.auth import SecureWebServer, Dispatcher, Result
+from org.apache.oodt.cas.filemgr.datatransfer import TransferStatusTracker
+from org.apache.oodt.cas.filemgr.structs import Product
+from org.apache.oodt.cas.filemgr.util import GenericFileManagerObjectFactory
+from org.apache.oodt.cas.filemgr.util import XmlRpcStructFactory as Structs
+from org.apache.oodt.cas.metadata import Metadata
+from java.lang import Boolean, Double, Integer
+from java.util import Hashtable, Vector
+
+# We choose these default factory classes because it minimizes our dependencies
+# on heavyweight external packages, like smelly old SQL databases.
+_defaultFactories = {
+ 'catalog': 'org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory',
+ 'repository': 'org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory',
+ 'datatransfer': 'org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory',
+ 'validation': 'org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory'
+}
+
+# All available permissions. By default, the "root" user will be in the "wheel"
+# group and will have these permissions.
+_allPerms = [
+ 'filemgr.addMetadata',
+ 'filemgr.addProductReferences',
+ 'filemgr.addProductType',
+ 'filemgr.catalogProduct',
+ 'filemgr.getCurrentFileTransfer',
+ 'filemgr.getCurrentFileTransfers',
+ 'filemgr.getElementById',
+ 'filemgr.getElementByName',
+ 'filemgr.getElementsByProductType',
+ 'filemgr.getFirstPage',
+ 'filemgr.getLastPage',
+ 'filemgr.getMetadata',
+ 'filemgr.getNextPage',
+ 'filemgr.getNumProducts',
+ 'filemgr.getPrevPage',
+ 'filemgr.getProductById',
+ 'filemgr.getProductByName',
+ 'filemgr.getProductPctTransferred',
+ 'filemgr.getProductReferences',
+ 'filemgr.getProductsByProductType',
+ 'filemgr.getProductTypeById',
+ 'filemgr.getProductTypeByName',
+ 'filemgr.getProductTypes',
+ 'filemgr.getRefPctTransferred',
+ 'filemgr.getTopNProducts',
+ 'filemgr.handleRequest',
+ 'filemgr.hasProduct',
+ 'filemgr.ingestProduct',
+ 'filemgr.isTransferComplete',
+ 'filemgr.pagedQuery',
+ 'filemgr.query',
+ 'filemgr.removeFile',
+ 'filemgr.removeProductTransferStatus',
+ 'filemgr.setProductTransferStatus',
+ 'filemgr.transferFile',
+ 'filemgr.transferringProduct',
+ 'usermgr.addGroup',
+ 'usermgr.addPermissionToGroup',
+ 'usermgr.addUser',
+ 'usermgr.addUserToGroup',
+ 'usermgr.removeGroup',
+ 'usermgr.removePermissionFromGroup',
+ 'usermgr.removeUser',
+ 'usermgr.removeUserFromGroup'
+]
+
+def _toJavaBoolean(truthiness):
+ '''Convert a Python boolean into the string format that Java uses: true or false.
+ '''
+ if truthiness:
+ return 'true'
+ else:
+ return 'false'
+
+
+def _encodePassword(pw):
+ '''Encode a password using an SHA-1 digest.
+ '''
+ return sha.new(pw).digest()
+
+
+class User:
+ '''A user of the CAS. Users don't have permissions directly; instead they receive
+ them implicitly by being members of groups, which do have permissions.
+ '''
+ def __init__(self, userID, name, email, password, groups=[]):
+ self.userID, self.name, self.email, self.password, self.groups = userID, name, email, password, groups
+
+ def __cmp__(self, other):
+ return cmp(self.userID, other.userID)
+
+ def __hash__(self):
+ return hash(self.userID)
+
+ def __repr__(self):
+ return 'User(userID=%s,name=%s,email=%s,password=%s,groups=%r)' % (
+ self.userID, self.name, self.email, self.password, self.groups
+ )
+
+
+class Group:
+ '''A CAS group. The group contains a sequence of permissions, which are strings
+ that name the XML-RPC methods that the group is allowed to call.
+ '''
+ def __init__(self, groupID, name, email, perms=[]):
+ self.groupID, self.name, self.email, self.perms = groupID, name, email, perms
+
+ def __cmp__(self, other):
+ return cmp(self.groupID, other.groupID)
+
+ def __hash__(self):
+ return hash(self.groupID)
+
+ def __repr__(self):
+ return 'Group(groupdID=%s,name=%s,email=%s,perms=%r)' % (self.groupdID, self.name, self.email, self.perms)
+
+
+class UserDB:
+ '''The user database records all the users and groups.
+ '''
+ def __init__(self, users, groups, filename):
+ self.users, self.groups, self.filename = users, groups, filename
+
+ def authenticate(self, name, password):
+ '''Authenticate a user by checking the user name and password.
+ Return true if the user's password matches the given one. The
+ password given should be in SHA-1 digest format.
+ '''
+ user = self.users[name]
+ return user.userID == name and user.password == password
+
+ def authorize(self, name, perm):
+ '''Authorize if the user has the given permission. Return true if
+ the user can do it, false otherwise.
+ '''
+ user = self.users[name]
+ for group in user.groups:
+ if perm in group.perms:
+ return True
+ return False
+
+ def save(self):
+ '''Save the user database to disk.
+ '''
+ f = file(self.filename, 'wb')
+ pickle.dump(self, f)
+ f.close()
+
+
+class FileMgrDispatcher(Dispatcher):
+ '''The file manager dispatcher handles all XML-RPC calls.
+ '''
+ def __init__(self, catalog, repo, xfer, userDB):
+ self.catalog, self.repo, self.xfer, self.userDB = catalog, repo, xfer, userDB
+ self.tracker = TransferStatusTracker(self.catalog)
+
+ def handleRequest(self, methodSpecifier, params, user, password):
+ '''Handle an XML-RPC request. First, authenticate the user. If the user's
+ authentic (by dint of providing a correct user ID and password pair), then
+ authorize if the method the user is trying to call is available.
+ '''
+ password = _encodePassword(password)
+ if self.userDB.authenticate(user, password):
+ if self.userDB.authorize(user, methodSpecifier):
+ obj, method = methodSpecifier.split('.')
+ if obj not in ('filemgr', 'usermgr'):
+ raise ValueError('Unknown object')
+ func = getattr(self, method)
+ return func(params)
+ raise ValueError('Not authorized for "%s"' % methodSpecifier)
+ raise ValueError('Illegal user name "%s" and/or password' % user)
+
+ def getProductTypeByName(self, params):
+ return Result(None, Structs.getXmlRpcProductType(self.repo.getProductTypeByName(params[0])))
+
+ def ingestProduct(self, params):
+ productHash, metadata, clientXfer = params
+ p = Structs.getProductFromXmlRpc(productHash)
+ p.setTransferStatus(Product.STATUS_TRANSFER)
+ self.catalog.addProduct(p)
+
+ m = Metadata()
+ m.addMetadata(metadata)
+ self.catalog.addMetadata(m, p)
+
+ if not clientXfer:
+ versioner = GenericFileManagerObjectFactory.getVersionerFromClassName(p.getProductType().getVersioner())
+ versioner.createDataStoreReferences(p, m)
+ self.catalog.addProductReferences(p)
+ self.xfer.transferProduct(p)
+ p.setTransferStatus(Product.STATUS_RECEIVED)
+ self.catalog.setProductTranfserStatus(p)
+ return Result(None, p.getProductId())
+
+ def addProductReferences(self, params):
+ self.catalog.addProductReferences(Structs.getProductFromXmlRpc(params[0]))
+ return Result(Boolean, 'true')
+
+ def transferringProduct(self, params):
+ self.tracker.transferringProduct(Structs.getProductFromXmlRpc(params[0]))
+ return Result(Boolean, 'true')
+
+ def removeProductTransferStatus(self, params):
+ self.tracker.removeProductTransferStatus(Structs.getProductFromXmlRpc(params[0]))
+ return Result(Boolean, 'true')
+
+ def setProductTransferStatus(self, params):
+ self.catalog.setProductTransferStatus(Structs.getProductFromXmlRpc(params[0]))
+ return Result(Boolean, 'true')
+
+ def getCurrentFileTransfer(self, params):
+ status = self.tracker.getCurrentFileTransfer()
+ if status is None:
+ return Result(None, Hashtable())
+ else:
+ return Result(None, Structs.getXmlRpcFileTransferStatus(status))
+
+ def getCurrentFileTransfers(self, params):
+ xfers = self.tracker.getCurrentFileTransfers()
+ if xfers is not None and len(xfers) > 0:
+ return Result(None, Structs.getXmlRpcFileTransferStatuses(xfers))
+ else:
+ return Result(None, Vector())
+
+ def getProductPctTransferred(self, params):
+ return Result(Double, str(self.tracker.getPctTransferred((Structs.getProductFromXmlRpc(params[0])))))
+
+ def getRefPctTransferred(self, params):
+ pct = self.tracker.getPctTransferred(Structs.getReferenceFromXmlRpc(params[0]))
+ return Result(Double, str(pct))
+
+ def isTransferComplete(self, params):
+ return Result(Boolean, _toJavaBoolean(self.tracker.isTransferComplete(Structs.getProductFromXmlRpc(params[0]))))
+
+ def pagedQuery(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[1])
+ query = Structs.getQueryFromXmlRpc(params[0])
+ return Result(None, Structs.getXmlRpcProductPage(self.catalog.pagedQuery(query, ptype, params[2])))
+
+ def getFirstPage(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ return Result(None, Structs.getXmlRpcProductPage(self.catalog.getFirstPage(ptype)))
+
+ def getLastPage(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ return Result(None, Structs.getXmlRpcProductPage(self.catalog.getLastProductPage(ptype)))
+
+ def getNextPage(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ page = Structs.getProductPageFromXmlRpc(params[1])
+ return Result(None, Structs.getXmlRpcProductPage(self.catalog.getNextPage(ptype, page)))
+
+ def getPrevPage(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ page = Structs.getProductPageFromXmlRpc(params[1])
+ return Result(None, Structs.getXmlRpcProductPage(self.catalog.getPrevPage(ptype, page)))
+
+ def addProductType(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ self.repo.addProductType(ptype)
+ return Result(None, ptype.getProductTypeId())
+
+ def getNumProducts(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ return Result(Integer, str(self.catalog.getNumProducts(ptype)))
+
+ def getTopNProducts(self, params):
+ if len(params) == 1:
+ return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0])))
+ ptype = Structs.getProductTypeFromXmlRpc(params[1])
+ return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0], ptype)))
+
+ def hasProduct(self, params):
+ p = self.catalog.getProductByName(params[0])
+ return Result(Boolean, _toJavaBoolean(p is not None and p.transferStatus == Product.STATUS_RECEIVED))
+
+ def getMetadata(self, params):
+ return Result(None, self.catalog.getMetadata(Structs.getProductFromXmlRpc(params[0])).getHashtable())
+
+ def getProductTypes(self, params):
+ return Result(None, Structs.getXmlRpcProductList(self.repo.getProductTypes()))
+
+ def getProductReferences(self, params):
+ p = Structs.getProductFromXmlRpc(params[0])
+ return Result(None, Structs.getXmlRpcReferences(self.catalog.getProductReferences(p)))
+
+ def getProductById(self, params):
+ return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductById(params[0])))
+
+ def getProductByName(self, params):
+ return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductByName(params[0])))
+
+ def getProductsByProductType(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ return Result(None, Structs.getXmlRpcProductList(self.catalog.getProductsByProductType(ptype)))
+
+ def getElementsByProductType(self, params):
+ ptype = Structs.getProductTypeFromXmlRpc(params[0])
+ return Structs.getXmlRpcElementList(self.catalog.getValidationLayer().getElements(ptype))
+
+ def getElementById(self, params):
+ return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementById(params[0]))
+
+ def getElementByName(self, params):
+ return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementByName(params[0]))
+
+ def query(self, params):
+ q = Structs.getQueryFromXmlRpc(params[0])
+ ptype = Structs.getProductTypeFromXmlRpc(params[1])
+ ids = self.catalog.query(q, ptype)
+ if ids is not None and len(ids) > 0:
+ return Result(None, [self.catalog.getProductById(i) for i in ids])
+ return Result(None, Vector())
+
+ def getProductTypeById(self, params):
+ ptype = self.repo.getProductTypeById(params[0])
+ return Result(None, Structs.getXmlRpcProductType(ptype))
+
+ def catalogProduct(self, params):
+ p = Structs.getProductFromXmlRpc(params[0])
+ return Result(None, self.catalog.addProduct(p))
+
+ def addMetadata(self, params):
+ p = Structs.getProductFromXmlRpc(params[0])
+ m = Metadata()
+ m.addMetadata(params[1])
+ self.catalog.addMetadata(m, p)
+ return Result(Boolean, 'true')
+
+ def transferFile(self, params):
+ outFile, data, offset, numBytes = params
+ if os.path.exists(outFile):
+ out = file(outFile, 'ab')
+ else:
+ dirPath = os.dirname(outFile)
+ os.makedirs(dirPath)
+ out = file(outFile, 'wb')
+ out.seek(offset)
+ out.write(data)
+ out.close()
+ return Result(Boolean, 'true')
+
+ def removeFile(self, params):
+ os.remove(params[0])
+ return Result(Boolean, 'true')
+
+ def addUser(self, params):
+ userID = params[0]
+ user = User(userID, params[1], params[2], _encodePassword(params[3]), [])
+ self.userDB.users[userID] = user
+ self.userDB.save()
+ return Result(Boolean, 'true')
+
+ def removeUser(self, params):
+ del self.userDB.users[params[0]]
+ return Result(Boolean, 'true')
+
+ def addGroup(self, params):
+ groupID = params[0]
+ group = Group(groupID, params[1], params[2])
+ self.userDB.groups[groupID] = group
+ self.userDB.save()
+ return Result(Boolean, 'true')
+
+ def removeGroup(self, params):
+ groupID = params[0]
+ del self.userDB.groups[groupID]
+ for user in self.userDB.users.itervalues():
+ indexes = []
+ index = 0
+ for group in user.groups:
+ if group.groupID == groupID:
+ indexes.append(index)
+ index += 1
+ indexes.reverse()
+ for index in indexes:
+ del user.groups[index]
+ self.userDB.save()
+ return Result(Boolean, 'true')
+
+ def addUserToGroup(self, params):
+ self.userDB.users[params[0]].groups.append(self.userDB.groups[params[1]])
+ return Result(Boolean, 'true')
+
+ def removeUserFromGroup(self, params):
+ groupID = params[1]
+ user = self.userDB.users[params[0]]
+ indexes = []
+ index = 0
+ for group in user.group:
+ if group.groupID == groupID:
+ indexes.append(index)
+ index += 1
+ indexes.reverse()
+ for index in indexes:
+ del user.groups[index]
+ self.userDB.save()
+ return Result(Boolean, 'true')
+
+ def addPermissionToGroup(self, params):
+ self.userDB.groups[params[0]].perms.append(params[1])
+ self.userDB.save()
+ return Result(Boolean, 'true')
+
+ def removePermissionFromGroup(self, params):
+ permName = params[1]
+ group = self.userDB.groups[params[0]]
+ indexes = []
+ index = 0
+ for perm in group.perms:
+ if perm == permName:
+ indexes.append(index)
+ indexes.reverse()
+ for index in indexes:
+ del group.perms[index]
+ self.userDB.save()
+ return Result(Boolean, 'true')
+
+
+def _usage():
+ '''Show a usage message to the stderr and quit.
+ '''
+ print >>sys.stderr, 'Usage: %s [-c <configFile>]' % sys.argv[0]
+ print >>sys.stderr, ' or: %s [--config=<configFile>]' % sys.argv[0]
+ sys.exit(2)
+
+
+def _parseCommandLine():
+ '''Parse the command line options. If any. The only option is -c (or --config)
+ that names a configuration file to use. If none given, reasonable defaults are
+ used. Well, mostly reasonable.
+ '''
+ try:
+ opts, args = getopt.getopt(sys.argv[1:], 'c:', 'config=')
+ except getopt.GetoptError:
+ _usage()
+ configFile = None
+ for option, arg in opts:
+ if option in ('-c', '--config'):
+ configFile = arg
+ if configFile is None or len(configFile) == 0:
+ _usage()
+ return configFile
+
+
+def _getConfig(configFile):
+ '''Get the configuration. This populates a configuration with default values
+ and then overwrites them with the configFile, which may be None (in which case,
+ no overwriting happens).
+ '''
+ configParser = ConfigParser()
+ configParser.add_section('factories')
+ for key, val in _defaultFactories.iteritems():
+ configParser.set('factories', key, val)
+
+ configParser.add_section('index')
+ configParser.set('index', 'path', 'index')
+ configParser.set('index', 'pageSize', '20')
+
+ current = '/'.join(os.path.split(os.getcwd()))
+ configParser.add_section('policies')
+ configParser.set('policies', 'repo', 'file:%s/policy' % current)
+ configParser.set('policies', 'validation', 'file:%s/policy' % current)
+ configParser.set('policies', 'user', '%s/user.db' % current)
+
+ if configFile is not None:
+ configParser.readfp(file(configFile))
+ return configParser
+
+
+def _setJavaProperties(config):
+ '''Set Java-based properties. The Java-based cas expects a whole bunch of
+ system properties to be set, sort of like global variables. Woot! Global
+ variables!
+ '''
+ from java.lang import System
+ System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.idxPath', config.get('index', 'path'))
+ System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.pageSize', config.get('index', 'pageSize'))
+ System.setProperty('org.apache.oodt.cas.filemgr.repositorymgr.dirs', config.get('policies', 'repo'))
+ System.setProperty('org.apache.oodt.cas.filemgr.validation.dirs', config.get('policies', 'validation'))
+ System.setProperty('org.apache.oodt.cas.filemgr.datatransfer.remote.chunkSize', '1024')
+ System.setProperty('filemgr.repository.factory', config.get('factories', 'repository'))
+ System.setProperty('filemgr.catalog.factory', config.get('factories', 'catalog'))
+ System.setProperty('filemgr.datatransfer.factory', config.get('factories', 'datatransfer'))
+ System.setProperty('filemgr.validationLayer.factory', config.get('factories', 'validation'))
+
+def _getUserDB(path):
+ '''Get the user database, creating it if necessary.
+ '''
+ try:
+ f = file(path, 'rb')
+ db = pickle.load(f)
+ f.close()
+ db.filename = path
+ db.save()
+ except:
+ wheel = Group('wheel', 'Administrators', 'teh.power@power.users', _allPerms)
+ root = User('root', 'Super User', 'teh.root@power.users', _encodePassword('poipu'), [wheel])
+ db = UserDB({'root': root}, {'wheel': wheel}, path)
+ db.save()
+ return db
+
+
+def main():
+ '''Start the CAS Filemgr Backend.
+ '''
+ configFile = _parseCommandLine()
+ configParser = _getConfig(configFile)
+ _setJavaProperties(configParser)
+
+ catalog = GenericFileManagerObjectFactory.getCatalogServiceFromFactory(configParser.get('factories', 'catalog'))
+ repo = GenericFileManagerObjectFactory.getRepositoryManagerServiceFromFactory(configParser.get('factories', 'repository'))
+ xfer = GenericFileManagerObjectFactory.getDataTransferServiceFromFactory(configParser.get('factories', 'datatransfer'))
+ userDB = _getUserDB(configParser.get('policies', 'user'))
+
+ ws = SecureWebServer(1999)
+ ws.addDispatcher(FileMgrDispatcher(catalog, repo, xfer, userDB))
+ ws.start()
+
+
+if __name__ == '__main__':
+ main()
+
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/metadata.py
----------------------------------------------------------------------
diff --git a/agility/oodt/metadata.py b/agility/oodt/metadata.py
new file mode 100644
index 0000000..1f1a9a9
--- /dev/null
+++ b/agility/oodt/metadata.py
@@ -0,0 +1,29 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+class MetaData(object):
+ def __init__(self):
+ self.data = {}
+
+ def addMetaData(self, key, value):
+ if key not in self.data:
+ self.data[key] = []
+ self.data[key].append(value)
+
+ def toXmlRpc(self):
+ return self.data
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/agility/oodt/workflow.py
----------------------------------------------------------------------
diff --git a/agility/oodt/workflow.py b/agility/oodt/workflow.py
new file mode 100644
index 0000000..ae83c23
--- /dev/null
+++ b/agility/oodt/workflow.py
@@ -0,0 +1,134 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+import xmlrpclib
+from oodt_metadata import MetaData
+
+
+class OODTWorkFlowWrapper(object):
+ def __init__(self, host):
+ self._server = xmlrpclib.Server(host)
+
+ #String executeDynamicWorkflow(Vector<String> taskIds, Hashtable metadata)
+ def executeDynamicWorkflow(self, taskIds, metadata):
+ return self._server.workflowmgr.executeDynamicWorkflow(taskIds, metadata)
+ #Hashtable getConditionById(String conditionId)
+ def getConditionById(self, conditionId):
+ return self._server.workflowmgr.getConditionById(conditionId)
+ #Hashtable getFirstPage()
+ def getFirstPage(self):
+ return self._server.workflowmgr.getFirstPage()
+ #Hashtable getLastPage()
+ def getLastPage(self):
+ return self._server.workflowmgr.getLastPage()
+ #Hashtable getNextPage(Hashtable currentPage)
+ def getNextPage(self, currentPage):
+ return self._server.workflowmgr.getNextPage()
+ #int getNumWorkflowInstances()
+ def getNumWorkflowInstances(self):
+ return self._server.workflowmgr.getNumWorkflowInstances()
+ #int getNumWorkflowInstancesByStatus(String status)
+ def getNumWorkflowInstancesByStatus(self, status):
+ return self._server.workflowmgr.getNumWorkflowInstancesByStatus(status)
+ #Hashtable getPrevPage(Hashtable currentPage)
+ def getPrevPage(self, currentPage):
+ return self._server.workflowmgr.getPrevPage(currentPage)
+ #Vector getRegisteredEvents()
+ def getEventNames(self):
+ return self._server.workflowmgr.getRegisteredEvents()
+ #Hashtable getTaskById(String taskId)
+ def getTaskById(self, taskId):
+ return self._server.workflowmgr.getTaskById(taskId)
+ #Hashtable getWorkflowById(String workflowId)
+ def getWorkflowById(self, workflowId):
+ return self._server.workflowmgr.getWorkflowById(workflowId)
+ #double getWorkflowCurrentTaskWallClockMinutes(String workflowInstId)
+ def getWorkflowCurrentTaskWallClockMinutes(self, workflowInstId):
+ return self._server.workflowmgr.getWorkflowCurrentTaskWallClockMinutes(workflowInstId)
+ #Hashtable getWorkflowInstanceById(String wInstId)
+ def getWorkflowInstanceById(self,wInstId):
+ return self._server.workflowmgr.getWorkflowInstanceById(wInstId)
+ #Hashtable getWorkflowInstanceMetadata(String wInstId)
+ def getWorkflowInstanceMetadata(self, wInstId):
+ return self._server.workflowmgr.getWorkflowInstanceMetadata(wInstId)
+ #Vector getWorkflowInstances()
+ def getWorkflowInstances(self):
+ return self._server.workflowmgr.getWorkflowInstances()
+ #Vector getWorkflowInstancesByStatus(String status)
+ def getWorkflowInstancesByStatus(self, status):
+ return self._server.workflowmgr.getWorkflowInstancesByStatus(status)
+ #Vector getWorkflows()
+ def getWorkflows(self):
+ return self._server.workflowmgr.getWorkflows()
+ #Vector getWorkflowsByEvent(String eventName)
+ def getWorkflowsByEvent(self, eventName):
+ return self._server.workflowmgr.getWorkflowsByEvent(eventName)
+ #double getWorkflowWallClockMinutes(String workflowInstId)
+ def getWorkflowWallClockMinutes(self,workflowInstId):
+ return getWorkflowWallClockMinutes(workflowInstId)
+ #boolean handleEvent(String eventName, Hashtable metadata)
+ def startEvent(self, eventName, metaData):
+ return self._server.workflowmgr.handleEvent(eventName, metaData)
+ #Hashtable paginateWorkflowInstances(int pageNum)
+ def paginateWorkflowInstances(self, pageNum):
+ return self._server.workflowmgr.paginateWorkflowInstances(pageNum)
+ #Hashtable paginateWorkflowInstances(int pageNum, String status)(self, workflowInstId)
+ def paginateWorkflowInstances(self, pageNum, status):
+ return self._server.workflowmgr.paginateWorkflowInstances(pageNum, status)
+ #boolean pauseWorkflowInstance(String workflowInstId)
+ def pauseWorkflowInstance(self, workflowInstId):
+ return self._server.workflowmgr.pauseWorkflowInstance(workflowInstId)
+ #boolean resumeWorkflowInstance(String workflowInstId)
+ def resumeWorkflowInstance(self, workflowInstId):
+ return self._server.workflowmgr.resumeWorkflowInstance(workflowInstId)
+ #boolean setWorkflowInstanceCurrentTaskEndDateTime(String wInstId, String endDateTimeIsoStr)
+ def setWorkflowInstanceCurrentTaskEndDateTime(self, workflowInstId, endDateTimeIsoStr):
+ return self._server.workflowmgr.setWorkflowInstanceCurrentTaskEndDateTime(workflowInstId, endDateTimeIsoStr)
+ #boolean setWorkflowInstanceCurrentTaskStartDateTime(String wInstId, String startDateTimeIsoStr)
+ def setWorkflowInstanceCurrentTaskStartDateTime(self, workflowInstId, startDateTimeIsoStr):
+ return self._server.workflowmgr.setWorkflowInstanceCurrentTaskStartDateTime(workflowInstId, startDateTimeIsoStr)
+ #boolean stopWorkflowInstance(String workflowInstId)
+ def stopWorkflowInstance(self, workflowInstId):
+ return self._server.workflowmgr.stopWorkflowInstance(workflowInstId)
+ #boolean updateMetadataForWorkflow(String workflowInstId, Hashtable metadata)
+ def updateMetadataForWorkflow(self, workflowInstId, metadata):
+ return self._server.workflowmgr.updateMetadataForWorkflow(workflowInstId, metadata)
+ #boolean updateWorkflowInstance(Hashtable workflowInst)
+ def updateWorkflowInstance(workflowInst):
+ return self._server.workflowmgr.updateWorkflowInstance(workflowInst)
+ #boolean updateWorkflowInstanceStatus(String workflowInstanceId, String status)
+ def updateWorkflowInstanceStatus(self, workflowInstId, status):
+ return self._server.workflowmgr.updateWorkflowInstanceStatus(workflowInstId, status)
+
+
+
+def main():
+ # create instance
+ oodt = OODTWorkFlowWrapper("http://localhost:9200")
+ # get event info
+ events = oodt.getEventNames()
+ # create metadata object to invoke an event
+ met = MetaData()
+ met.addMetaData("hello", "world")
+ # print available events
+ print 'available events:', events
+
+ # oodt.startEvent(events[0], met.toXmlRpc())
+
+if __name__ == '__main__':
+ main()
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/filemgr/src/main/python/fm.conf
----------------------------------------------------------------------
diff --git a/filemgr/src/main/python/fm.conf b/filemgr/src/main/python/fm.conf
deleted file mode 100644
index dae6438..0000000
--- a/filemgr/src/main/python/fm.conf
+++ /dev/null
@@ -1,43 +0,0 @@
-# encoding: utf-8
-#/*
-# * Licensed to the Apache Software Foundation (ASF) under one or more
-# * contributor license agreements. See the NOTICE file distributed with
-# * this work for additional information regarding copyright ownership.
-# * The ASF licenses this file to You under the Apache License, Version 2.0
-# * (the "License"); you may not use this file except in compliance with
-# * the License. You may obtain a copy of the License at
-# *
-# * http://www.apache.org/licenses/LICENSE-2.0
-# *
-# * Unless required by applicable law or agreed to in writing, software
-# * distributed under the License is distributed on an "AS IS" BASIS,
-# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# * See the License for the specific language governing permissions and
-# * limitations under the License.
-# */
-
-[index]
-# The path gives the name of the directory where the Lucene search engine
-# will build its indexes. The path refers to a directory; it will be created
-# if it doesn't exist. If it does exist, it must refer to an
-# existing Lucene index with correct sub-files present. So, don't make this
-# directory manually, or you'll be confused by error messages!
-path = @INDEX_PATH@
-pageSize = 20
-
-[policies]
-# The 'repo' and 'validation' are comma-separated lists of URLs. Typically, they
-# contain a single "file:" style URL that names a directory with policy XML files
-# within. The 'user' identifies the user database. It'll be created if it
-# doesn't exist.
-repo = @REPO_POLICY_URLS@
-validation = @VALIDATION_POLICY_URLS@
-user = @USER_DB_PATH@
-
-[factories]
-# There's no need to change anything here unless you really know what you're doing,
-# and trust me, you don't. Don't take it personally. It's just a fact.
-catalog = org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory
-repository = org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory
-datatransfer = org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory
-validation =org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/filemgr/src/main/python/fm.py
----------------------------------------------------------------------
diff --git a/filemgr/src/main/python/fm.py b/filemgr/src/main/python/fm.py
deleted file mode 100644
index 5eed046..0000000
--- a/filemgr/src/main/python/fm.py
+++ /dev/null
@@ -1,554 +0,0 @@
-#!/usr/bin/env python
-# encoding: utf-8
-#/*
-# * Licensed to the Apache Software Foundation (ASF) under one or more
-# * contributor license agreements. See the NOTICE file distributed with
-# * this work for additional information regarding copyright ownership.
-# * The ASF licenses this file to You under the Apache License, Version 2.0
-# * (the "License"); you may not use this file except in compliance with
-# * the License. You may obtain a copy of the License at
-# *
-# * http://www.apache.org/licenses/LICENSE-2.0
-# *
-# * Unless required by applicable law or agreed to in writing, software
-# * distributed under the License is distributed on an "AS IS" BASIS,
-# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# * See the License for the specific language governing permissions and
-# * limitations under the License.
-# */
-'''
-CAS Filemgr Python Server
-
-This is the Python server API for an authenticated Catalog and Archive System. It uses the
-OODT Catalog and Archive System as its core, adding user/password-based authentication
-and role-based authorization.
-'''
-
-import sys, os, os.path
-import getopt, sha, pickle
-
-from ConfigParser import ConfigParser
-from org.apache.oodt.cas.filemgr.system.auth import SecureWebServer, Dispatcher, Result
-from org.apache.oodt.cas.filemgr.datatransfer import TransferStatusTracker
-from org.apache.oodt.cas.filemgr.structs import Product
-from org.apache.oodt.cas.filemgr.util import GenericFileManagerObjectFactory
-from org.apache.oodt.cas.filemgr.util import XmlRpcStructFactory as Structs
-from org.apache.oodt.cas.metadata import Metadata
-from java.lang import Boolean, Double, Integer
-from java.util import Hashtable, Vector
-
-# We choose these default factory classes because it minimizes our dependencies
-# on heavyweight external packages, like smelly old SQL databases.
-_defaultFactories = {
- 'catalog': 'org.apache.oodt.cas.filemgr.catalog.LuceneCatalogFactory',
- 'repository': 'org.apache.oodt.cas.filemgr.repository.XMLRepositoryManagerFactory',
- 'datatransfer': 'org.apache.oodt.cas.filemgr.datatransfer.LocalDataTransferFactory',
- 'validation': 'org.apache.oodt.cas.filemgr.validation.XMLValidationLayerFactory'
-}
-
-# All available permissions. By default, the "root" user will be in the "wheel"
-# group and will have these permissions.
-_allPerms = [
- 'filemgr.addMetadata',
- 'filemgr.addProductReferences',
- 'filemgr.addProductType',
- 'filemgr.catalogProduct',
- 'filemgr.getCurrentFileTransfer',
- 'filemgr.getCurrentFileTransfers',
- 'filemgr.getElementById',
- 'filemgr.getElementByName',
- 'filemgr.getElementsByProductType',
- 'filemgr.getFirstPage',
- 'filemgr.getLastPage',
- 'filemgr.getMetadata',
- 'filemgr.getNextPage',
- 'filemgr.getNumProducts',
- 'filemgr.getPrevPage',
- 'filemgr.getProductById',
- 'filemgr.getProductByName',
- 'filemgr.getProductPctTransferred',
- 'filemgr.getProductReferences',
- 'filemgr.getProductsByProductType',
- 'filemgr.getProductTypeById',
- 'filemgr.getProductTypeByName',
- 'filemgr.getProductTypes',
- 'filemgr.getRefPctTransferred',
- 'filemgr.getTopNProducts',
- 'filemgr.handleRequest',
- 'filemgr.hasProduct',
- 'filemgr.ingestProduct',
- 'filemgr.isTransferComplete',
- 'filemgr.pagedQuery',
- 'filemgr.query',
- 'filemgr.removeFile',
- 'filemgr.removeProductTransferStatus',
- 'filemgr.setProductTransferStatus',
- 'filemgr.transferFile',
- 'filemgr.transferringProduct',
- 'usermgr.addGroup',
- 'usermgr.addPermissionToGroup',
- 'usermgr.addUser',
- 'usermgr.addUserToGroup',
- 'usermgr.removeGroup',
- 'usermgr.removePermissionFromGroup',
- 'usermgr.removeUser',
- 'usermgr.removeUserFromGroup'
-]
-
-def _toJavaBoolean(truthiness):
- '''Convert a Python boolean into the string format that Java uses: true or false.
- '''
- if truthiness:
- return 'true'
- else:
- return 'false'
-
-
-def _encodePassword(pw):
- '''Encode a password using an SHA-1 digest.
- '''
- return sha.new(pw).digest()
-
-
-class User:
- '''A user of the CAS. Users don't have permissions directly; instead they receive
- them implicitly by being members of groups, which do have permissions.
- '''
- def __init__(self, userID, name, email, password, groups=[]):
- self.userID, self.name, self.email, self.password, self.groups = userID, name, email, password, groups
-
- def __cmp__(self, other):
- return cmp(self.userID, other.userID)
-
- def __hash__(self):
- return hash(self.userID)
-
- def __repr__(self):
- return 'User(userID=%s,name=%s,email=%s,password=%s,groups=%r)' % (
- self.userID, self.name, self.email, self.password, self.groups
- )
-
-
-class Group:
- '''A CAS group. The group contains a sequence of permissions, which are strings
- that name the XML-RPC methods that the group is allowed to call.
- '''
- def __init__(self, groupID, name, email, perms=[]):
- self.groupID, self.name, self.email, self.perms = groupID, name, email, perms
-
- def __cmp__(self, other):
- return cmp(self.groupID, other.groupID)
-
- def __hash__(self):
- return hash(self.groupID)
-
- def __repr__(self):
- return 'Group(groupdID=%s,name=%s,email=%s,perms=%r)' % (self.groupdID, self.name, self.email, self.perms)
-
-
-class UserDB:
- '''The user database records all the users and groups.
- '''
- def __init__(self, users, groups, filename):
- self.users, self.groups, self.filename = users, groups, filename
-
- def authenticate(self, name, password):
- '''Authenticate a user by checking the user name and password.
- Return true if the user's password matches the given one. The
- password given should be in SHA-1 digest format.
- '''
- user = self.users[name]
- return user.userID == name and user.password == password
-
- def authorize(self, name, perm):
- '''Authorize if the user has the given permission. Return true if
- the user can do it, false otherwise.
- '''
- user = self.users[name]
- for group in user.groups:
- if perm in group.perms:
- return True
- return False
-
- def save(self):
- '''Save the user database to disk.
- '''
- f = file(self.filename, 'wb')
- pickle.dump(self, f)
- f.close()
-
-
-class FileMgrDispatcher(Dispatcher):
- '''The file manager dispatcher handles all XML-RPC calls.
- '''
- def __init__(self, catalog, repo, xfer, userDB):
- self.catalog, self.repo, self.xfer, self.userDB = catalog, repo, xfer, userDB
- self.tracker = TransferStatusTracker(self.catalog)
-
- def handleRequest(self, methodSpecifier, params, user, password):
- '''Handle an XML-RPC request. First, authenticate the user. If the user's
- authentic (by dint of providing a correct user ID and password pair), then
- authorize if the method the user is trying to call is available.
- '''
- password = _encodePassword(password)
- if self.userDB.authenticate(user, password):
- if self.userDB.authorize(user, methodSpecifier):
- obj, method = methodSpecifier.split('.')
- if obj not in ('filemgr', 'usermgr'):
- raise ValueError('Unknown object')
- func = getattr(self, method)
- return func(params)
- raise ValueError('Not authorized for "%s"' % methodSpecifier)
- raise ValueError('Illegal user name "%s" and/or password' % user)
-
- def getProductTypeByName(self, params):
- return Result(None, Structs.getXmlRpcProductType(self.repo.getProductTypeByName(params[0])))
-
- def ingestProduct(self, params):
- productHash, metadata, clientXfer = params
- p = Structs.getProductFromXmlRpc(productHash)
- p.setTransferStatus(Product.STATUS_TRANSFER)
- self.catalog.addProduct(p)
-
- m = Metadata()
- m.addMetadata(metadata)
- self.catalog.addMetadata(m, p)
-
- if not clientXfer:
- versioner = GenericFileManagerObjectFactory.getVersionerFromClassName(p.getProductType().getVersioner())
- versioner.createDataStoreReferences(p, m)
- self.catalog.addProductReferences(p)
- self.xfer.transferProduct(p)
- p.setTransferStatus(Product.STATUS_RECEIVED)
- self.catalog.setProductTranfserStatus(p)
- return Result(None, p.getProductId())
-
- def addProductReferences(self, params):
- self.catalog.addProductReferences(Structs.getProductFromXmlRpc(params[0]))
- return Result(Boolean, 'true')
-
- def transferringProduct(self, params):
- self.tracker.transferringProduct(Structs.getProductFromXmlRpc(params[0]))
- return Result(Boolean, 'true')
-
- def removeProductTransferStatus(self, params):
- self.tracker.removeProductTransferStatus(Structs.getProductFromXmlRpc(params[0]))
- return Result(Boolean, 'true')
-
- def setProductTransferStatus(self, params):
- self.catalog.setProductTransferStatus(Structs.getProductFromXmlRpc(params[0]))
- return Result(Boolean, 'true')
-
- def getCurrentFileTransfer(self, params):
- status = self.tracker.getCurrentFileTransfer()
- if status is None:
- return Result(None, Hashtable())
- else:
- return Result(None, Structs.getXmlRpcFileTransferStatus(status))
-
- def getCurrentFileTransfers(self, params):
- xfers = self.tracker.getCurrentFileTransfers()
- if xfers is not None and len(xfers) > 0:
- return Result(None, Structs.getXmlRpcFileTransferStatuses(xfers))
- else:
- return Result(None, Vector())
-
- def getProductPctTransferred(self, params):
- return Result(Double, str(self.tracker.getPctTransferred((Structs.getProductFromXmlRpc(params[0])))))
-
- def getRefPctTransferred(self, params):
- pct = self.tracker.getPctTransferred(Structs.getReferenceFromXmlRpc(params[0]))
- return Result(Double, str(pct))
-
- def isTransferComplete(self, params):
- return Result(Boolean, _toJavaBoolean(self.tracker.isTransferComplete(Structs.getProductFromXmlRpc(params[0]))))
-
- def pagedQuery(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[1])
- query = Structs.getQueryFromXmlRpc(params[0])
- return Result(None, Structs.getXmlRpcProductPage(self.catalog.pagedQuery(query, ptype, params[2])))
-
- def getFirstPage(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- return Result(None, Structs.getXmlRpcProductPage(self.catalog.getFirstPage(ptype)))
-
- def getLastPage(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- return Result(None, Structs.getXmlRpcProductPage(self.catalog.getLastProductPage(ptype)))
-
- def getNextPage(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- page = Structs.getProductPageFromXmlRpc(params[1])
- return Result(None, Structs.getXmlRpcProductPage(self.catalog.getNextPage(ptype, page)))
-
- def getPrevPage(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- page = Structs.getProductPageFromXmlRpc(params[1])
- return Result(None, Structs.getXmlRpcProductPage(self.catalog.getPrevPage(ptype, page)))
-
- def addProductType(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- self.repo.addProductType(ptype)
- return Result(None, ptype.getProductTypeId())
-
- def getNumProducts(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- return Result(Integer, str(self.catalog.getNumProducts(ptype)))
-
- def getTopNProducts(self, params):
- if len(params) == 1:
- return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0])))
- ptype = Structs.getProductTypeFromXmlRpc(params[1])
- return Result(None, Structs.getXmlRpcProductList(self.catalog.getTopNProducts(params[0], ptype)))
-
- def hasProduct(self, params):
- p = self.catalog.getProductByName(params[0])
- return Result(Boolean, _toJavaBoolean(p is not None and p.transferStatus == Product.STATUS_RECEIVED))
-
- def getMetadata(self, params):
- return Result(None, self.catalog.getMetadata(Structs.getProductFromXmlRpc(params[0])).getHashtable())
-
- def getProductTypes(self, params):
- return Result(None, Structs.getXmlRpcProductList(self.repo.getProductTypes()))
-
- def getProductReferences(self, params):
- p = Structs.getProductFromXmlRpc(params[0])
- return Result(None, Structs.getXmlRpcReferences(self.catalog.getProductReferences(p)))
-
- def getProductById(self, params):
- return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductById(params[0])))
-
- def getProductByName(self, params):
- return Result(None, Structs.getXmlRpcProduct(self.catalog.getProductByName(params[0])))
-
- def getProductsByProductType(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- return Result(None, Structs.getXmlRpcProductList(self.catalog.getProductsByProductType(ptype)))
-
- def getElementsByProductType(self, params):
- ptype = Structs.getProductTypeFromXmlRpc(params[0])
- return Structs.getXmlRpcElementList(self.catalog.getValidationLayer().getElements(ptype))
-
- def getElementById(self, params):
- return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementById(params[0]))
-
- def getElementByName(self, params):
- return Structs.getXmlRpcElement(self.catalog.getValidationLayer().getElementByName(params[0]))
-
- def query(self, params):
- q = Structs.getQueryFromXmlRpc(params[0])
- ptype = Structs.getProductTypeFromXmlRpc(params[1])
- ids = self.catalog.query(q, ptype)
- if ids is not None and len(ids) > 0:
- return Result(None, [self.catalog.getProductById(i) for i in ids])
- return Result(None, Vector())
-
- def getProductTypeById(self, params):
- ptype = self.repo.getProductTypeById(params[0])
- return Result(None, Structs.getXmlRpcProductType(ptype))
-
- def catalogProduct(self, params):
- p = Structs.getProductFromXmlRpc(params[0])
- return Result(None, self.catalog.addProduct(p))
-
- def addMetadata(self, params):
- p = Structs.getProductFromXmlRpc(params[0])
- m = Metadata()
- m.addMetadata(params[1])
- self.catalog.addMetadata(m, p)
- return Result(Boolean, 'true')
-
- def transferFile(self, params):
- outFile, data, offset, numBytes = params
- if os.path.exists(outFile):
- out = file(outFile, 'ab')
- else:
- dirPath = os.dirname(outFile)
- os.makedirs(dirPath)
- out = file(outFile, 'wb')
- out.seek(offset)
- out.write(data)
- out.close()
- return Result(Boolean, 'true')
-
- def removeFile(self, params):
- os.remove(params[0])
- return Result(Boolean, 'true')
-
- def addUser(self, params):
- userID = params[0]
- user = User(userID, params[1], params[2], _encodePassword(params[3]), [])
- self.userDB.users[userID] = user
- self.userDB.save()
- return Result(Boolean, 'true')
-
- def removeUser(self, params):
- del self.userDB.users[params[0]]
- return Result(Boolean, 'true')
-
- def addGroup(self, params):
- groupID = params[0]
- group = Group(groupID, params[1], params[2])
- self.userDB.groups[groupID] = group
- self.userDB.save()
- return Result(Boolean, 'true')
-
- def removeGroup(self, params):
- groupID = params[0]
- del self.userDB.groups[groupID]
- for user in self.userDB.users.itervalues():
- indexes = []
- index = 0
- for group in user.groups:
- if group.groupID == groupID:
- indexes.append(index)
- index += 1
- indexes.reverse()
- for index in indexes:
- del user.groups[index]
- self.userDB.save()
- return Result(Boolean, 'true')
-
- def addUserToGroup(self, params):
- self.userDB.users[params[0]].groups.append(self.userDB.groups[params[1]])
- return Result(Boolean, 'true')
-
- def removeUserFromGroup(self, params):
- groupID = params[1]
- user = self.userDB.users[params[0]]
- indexes = []
- index = 0
- for group in user.group:
- if group.groupID == groupID:
- indexes.append(index)
- index += 1
- indexes.reverse()
- for index in indexes:
- del user.groups[index]
- self.userDB.save()
- return Result(Boolean, 'true')
-
- def addPermissionToGroup(self, params):
- self.userDB.groups[params[0]].perms.append(params[1])
- self.userDB.save()
- return Result(Boolean, 'true')
-
- def removePermissionFromGroup(self, params):
- permName = params[1]
- group = self.userDB.groups[params[0]]
- indexes = []
- index = 0
- for perm in group.perms:
- if perm == permName:
- indexes.append(index)
- indexes.reverse()
- for index in indexes:
- del group.perms[index]
- self.userDB.save()
- return Result(Boolean, 'true')
-
-
-def _usage():
- '''Show a usage message to the stderr and quit.
- '''
- print >>sys.stderr, 'Usage: %s [-c <configFile>]' % sys.argv[0]
- print >>sys.stderr, ' or: %s [--config=<configFile>]' % sys.argv[0]
- sys.exit(2)
-
-
-def _parseCommandLine():
- '''Parse the command line options. If any. The only option is -c (or --config)
- that names a configuration file to use. If none given, reasonable defaults are
- used. Well, mostly reasonable.
- '''
- try:
- opts, args = getopt.getopt(sys.argv[1:], 'c:', 'config=')
- except getopt.GetoptError:
- _usage()
- configFile = None
- for option, arg in opts:
- if option in ('-c', '--config'):
- configFile = arg
- if configFile is None or len(configFile) == 0:
- _usage()
- return configFile
-
-
-def _getConfig(configFile):
- '''Get the configuration. This populates a configuration with default values
- and then overwrites them with the configFile, which may be None (in which case,
- no overwriting happens).
- '''
- configParser = ConfigParser()
- configParser.add_section('factories')
- for key, val in _defaultFactories.iteritems():
- configParser.set('factories', key, val)
-
- configParser.add_section('index')
- configParser.set('index', 'path', 'index')
- configParser.set('index', 'pageSize', '20')
-
- current = '/'.join(os.path.split(os.getcwd()))
- configParser.add_section('policies')
- configParser.set('policies', 'repo', 'file:%s/policy' % current)
- configParser.set('policies', 'validation', 'file:%s/policy' % current)
- configParser.set('policies', 'user', '%s/user.db' % current)
-
- if configFile is not None:
- configParser.readfp(file(configFile))
- return configParser
-
-
-def _setJavaProperties(config):
- '''Set Java-based properties. The Java-based cas expects a whole bunch of
- system properties to be set, sort of like global variables. Woot! Global
- variables!
- '''
- from java.lang import System
- System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.idxPath', config.get('index', 'path'))
- System.setProperty('org.apache.oodt.cas.filemgr.catalog.lucene.pageSize', config.get('index', 'pageSize'))
- System.setProperty('org.apache.oodt.cas.filemgr.repositorymgr.dirs', config.get('policies', 'repo'))
- System.setProperty('org.apache.oodt.cas.filemgr.validation.dirs', config.get('policies', 'validation'))
- System.setProperty('org.apache.oodt.cas.filemgr.datatransfer.remote.chunkSize', '1024')
- System.setProperty('filemgr.repository.factory', config.get('factories', 'repository'))
- System.setProperty('filemgr.catalog.factory', config.get('factories', 'catalog'))
- System.setProperty('filemgr.datatransfer.factory', config.get('factories', 'datatransfer'))
- System.setProperty('filemgr.validationLayer.factory', config.get('factories', 'validation'))
-
-def _getUserDB(path):
- '''Get the user database, creating it if necessary.
- '''
- try:
- f = file(path, 'rb')
- db = pickle.load(f)
- f.close()
- db.filename = path
- db.save()
- except:
- wheel = Group('wheel', 'Administrators', 'teh.power@power.users', _allPerms)
- root = User('root', 'Super User', 'teh.root@power.users', _encodePassword('poipu'), [wheel])
- db = UserDB({'root': root}, {'wheel': wheel}, path)
- db.save()
- return db
-
-
-def main():
- '''Start the CAS Filemgr Backend.
- '''
- configFile = _parseCommandLine()
- configParser = _getConfig(configFile)
- _setJavaProperties(configParser)
-
- catalog = GenericFileManagerObjectFactory.getCatalogServiceFromFactory(configParser.get('factories', 'catalog'))
- repo = GenericFileManagerObjectFactory.getRepositoryManagerServiceFromFactory(configParser.get('factories', 'repository'))
- xfer = GenericFileManagerObjectFactory.getDataTransferServiceFromFactory(configParser.get('factories', 'datatransfer'))
- userDB = _getUserDB(configParser.get('policies', 'user'))
-
- ws = SecureWebServer(1999)
- ws.addDispatcher(FileMgrDispatcher(catalog, repo, xfer, userDB))
- ws.start()
-
-
-if __name__ == '__main__':
- main()
-
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/filemgr/src/main/python/setclasspath.sh
----------------------------------------------------------------------
diff --git a/filemgr/src/main/python/setclasspath.sh b/filemgr/src/main/python/setclasspath.sh
deleted file mode 100755
index 1fbfc48..0000000
--- a/filemgr/src/main/python/setclasspath.sh
+++ /dev/null
@@ -1,17 +0,0 @@
-#/*
-# * Licensed to the Apache Software Foundation (ASF) under one or more
-# * contributor license agreements. See the NOTICE file distributed with
-# * this work for additional information regarding copyright ownership.
-# * The ASF licenses this file to You under the Apache License, Version 2.0
-# * (the "License"); you may not use this file except in compliance with
-# * the License. You may obtain a copy of the License at
-# *
-# * http://www.apache.org/licenses/LICENSE-2.0
-# *
-# * Unless required by applicable law or agreed to in writing, software
-# * distributed under the License is distributed on an "AS IS" BASIS,
-# * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# * See the License for the specific language governing permissions and
-# * limitations under the License.
-# */
-setenv CLASSPATH ${HOME}/.m2/repository/org.apache.oodt.cas-filemgr/1.7.0-dev/cas-filemgr-1.7.0-dev.jar:${HOME}/.m2/repository/xmlrpc/xmlrpc/2.1-dev/xmlrpc-2.1-dev.jar:${HOME}/.m2/repository/org.apache.oodt.cas-metadata/1.5.0/cas-metadata-1.5.0.jar:${HOME}/.m2/repository/org/apache/lucene/lucene-core/2.0.0/lucene-core-2.0.0.jar:${HOME}/.m2/repository/jug/jug-asl/2.0.0/jug-asl-2.0.0.jar
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py
----------------------------------------------------------------------
diff --git a/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py b/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py
deleted file mode 100644
index 1f1a9a9..0000000
--- a/workflow/src/main/python/xmlrpc-wrapper/oodt_metadata.py
+++ /dev/null
@@ -1,29 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-
-class MetaData(object):
- def __init__(self):
- self.data = {}
-
- def addMetaData(self, key, value):
- if key not in self.data:
- self.data[key] = []
- self.data[key].append(value)
-
- def toXmlRpc(self):
- return self.data
http://git-wip-us.apache.org/repos/asf/oodt/blob/d329e1ab/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py
----------------------------------------------------------------------
diff --git a/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py b/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py
deleted file mode 100644
index ae83c23..0000000
--- a/workflow/src/main/python/xmlrpc-wrapper/oodt_workflow_xmlrpc.py
+++ /dev/null
@@ -1,134 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-
-import xmlrpclib
-from oodt_metadata import MetaData
-
-
-class OODTWorkFlowWrapper(object):
- def __init__(self, host):
- self._server = xmlrpclib.Server(host)
-
- #String executeDynamicWorkflow(Vector<String> taskIds, Hashtable metadata)
- def executeDynamicWorkflow(self, taskIds, metadata):
- return self._server.workflowmgr.executeDynamicWorkflow(taskIds, metadata)
- #Hashtable getConditionById(String conditionId)
- def getConditionById(self, conditionId):
- return self._server.workflowmgr.getConditionById(conditionId)
- #Hashtable getFirstPage()
- def getFirstPage(self):
- return self._server.workflowmgr.getFirstPage()
- #Hashtable getLastPage()
- def getLastPage(self):
- return self._server.workflowmgr.getLastPage()
- #Hashtable getNextPage(Hashtable currentPage)
- def getNextPage(self, currentPage):
- return self._server.workflowmgr.getNextPage()
- #int getNumWorkflowInstances()
- def getNumWorkflowInstances(self):
- return self._server.workflowmgr.getNumWorkflowInstances()
- #int getNumWorkflowInstancesByStatus(String status)
- def getNumWorkflowInstancesByStatus(self, status):
- return self._server.workflowmgr.getNumWorkflowInstancesByStatus(status)
- #Hashtable getPrevPage(Hashtable currentPage)
- def getPrevPage(self, currentPage):
- return self._server.workflowmgr.getPrevPage(currentPage)
- #Vector getRegisteredEvents()
- def getEventNames(self):
- return self._server.workflowmgr.getRegisteredEvents()
- #Hashtable getTaskById(String taskId)
- def getTaskById(self, taskId):
- return self._server.workflowmgr.getTaskById(taskId)
- #Hashtable getWorkflowById(String workflowId)
- def getWorkflowById(self, workflowId):
- return self._server.workflowmgr.getWorkflowById(workflowId)
- #double getWorkflowCurrentTaskWallClockMinutes(String workflowInstId)
- def getWorkflowCurrentTaskWallClockMinutes(self, workflowInstId):
- return self._server.workflowmgr.getWorkflowCurrentTaskWallClockMinutes(workflowInstId)
- #Hashtable getWorkflowInstanceById(String wInstId)
- def getWorkflowInstanceById(self,wInstId):
- return self._server.workflowmgr.getWorkflowInstanceById(wInstId)
- #Hashtable getWorkflowInstanceMetadata(String wInstId)
- def getWorkflowInstanceMetadata(self, wInstId):
- return self._server.workflowmgr.getWorkflowInstanceMetadata(wInstId)
- #Vector getWorkflowInstances()
- def getWorkflowInstances(self):
- return self._server.workflowmgr.getWorkflowInstances()
- #Vector getWorkflowInstancesByStatus(String status)
- def getWorkflowInstancesByStatus(self, status):
- return self._server.workflowmgr.getWorkflowInstancesByStatus(status)
- #Vector getWorkflows()
- def getWorkflows(self):
- return self._server.workflowmgr.getWorkflows()
- #Vector getWorkflowsByEvent(String eventName)
- def getWorkflowsByEvent(self, eventName):
- return self._server.workflowmgr.getWorkflowsByEvent(eventName)
- #double getWorkflowWallClockMinutes(String workflowInstId)
- def getWorkflowWallClockMinutes(self,workflowInstId):
- return getWorkflowWallClockMinutes(workflowInstId)
- #boolean handleEvent(String eventName, Hashtable metadata)
- def startEvent(self, eventName, metaData):
- return self._server.workflowmgr.handleEvent(eventName, metaData)
- #Hashtable paginateWorkflowInstances(int pageNum)
- def paginateWorkflowInstances(self, pageNum):
- return self._server.workflowmgr.paginateWorkflowInstances(pageNum)
- #Hashtable paginateWorkflowInstances(int pageNum, String status)(self, workflowInstId)
- def paginateWorkflowInstances(self, pageNum, status):
- return self._server.workflowmgr.paginateWorkflowInstances(pageNum, status)
- #boolean pauseWorkflowInstance(String workflowInstId)
- def pauseWorkflowInstance(self, workflowInstId):
- return self._server.workflowmgr.pauseWorkflowInstance(workflowInstId)
- #boolean resumeWorkflowInstance(String workflowInstId)
- def resumeWorkflowInstance(self, workflowInstId):
- return self._server.workflowmgr.resumeWorkflowInstance(workflowInstId)
- #boolean setWorkflowInstanceCurrentTaskEndDateTime(String wInstId, String endDateTimeIsoStr)
- def setWorkflowInstanceCurrentTaskEndDateTime(self, workflowInstId, endDateTimeIsoStr):
- return self._server.workflowmgr.setWorkflowInstanceCurrentTaskEndDateTime(workflowInstId, endDateTimeIsoStr)
- #boolean setWorkflowInstanceCurrentTaskStartDateTime(String wInstId, String startDateTimeIsoStr)
- def setWorkflowInstanceCurrentTaskStartDateTime(self, workflowInstId, startDateTimeIsoStr):
- return self._server.workflowmgr.setWorkflowInstanceCurrentTaskStartDateTime(workflowInstId, startDateTimeIsoStr)
- #boolean stopWorkflowInstance(String workflowInstId)
- def stopWorkflowInstance(self, workflowInstId):
- return self._server.workflowmgr.stopWorkflowInstance(workflowInstId)
- #boolean updateMetadataForWorkflow(String workflowInstId, Hashtable metadata)
- def updateMetadataForWorkflow(self, workflowInstId, metadata):
- return self._server.workflowmgr.updateMetadataForWorkflow(workflowInstId, metadata)
- #boolean updateWorkflowInstance(Hashtable workflowInst)
- def updateWorkflowInstance(workflowInst):
- return self._server.workflowmgr.updateWorkflowInstance(workflowInst)
- #boolean updateWorkflowInstanceStatus(String workflowInstanceId, String status)
- def updateWorkflowInstanceStatus(self, workflowInstId, status):
- return self._server.workflowmgr.updateWorkflowInstanceStatus(workflowInstId, status)
-
-
-
-def main():
- # create instance
- oodt = OODTWorkFlowWrapper("http://localhost:9200")
- # get event info
- events = oodt.getEventNames()
- # create metadata object to invoke an event
- met = MetaData()
- met.addMetaData("hello", "world")
- # print available events
- print 'available events:', events
-
- # oodt.startEvent(events[0], met.toXmlRpc())
-
-if __name__ == '__main__':
- main()