python.TaskManager.TaskManager Class Reference

Collaboration diagram for python.TaskManager.TaskManager:

Public Member Functions
	parseConnectionInfo (self, connstring='')
	__init__ (self, connstring='', createDatabase=False)
	__enter__ (self)
	__exit__ (self, exc_type, exc_value, traceback)
	execute (self, statementParts, doCommit=False, limit=None)
	deleteTask (self, dsName, taskName)
	addTask (self, dsName, taskName, template, release, njobs, taskpostprocsteps='', status=StatusCodes['SUBMITTED'], onDisk=OnDiskCodes['ALLONDISK'], createdTime=None, createdUser=None, createdHost=None, comment='')
	getStatus (self, dsName, taskName)
	setStatus (self, dsName, taskName, status, oldStatus=None)
	setDiskStatus (self, dsName, taskName, onDisk)
	updateStatus (self, dsName, taskName, status, nResultFiles, nJobs, nJobsSubmitted, nJobsRunning, nJobsPostProc, nJobsFailed, nJobsCompleted)
	setValue (self, dsName, taskName, fieldName, value)
	getValue (self, what, qual=())
	getCount (self, what, qual=())
	getNTasks (self, qual=())
	getTaskValue (self, dsName, taskName, what)
	taskIter (self, what=' *', qual=('order by UPDATED',))
	taskIterDict (self, what=' *', qual=('order by UPDATED',), limit=999999999)
	getTaskDict (self, dsname, taskname, what=' *', qual=())
	getDSNames (self, dsname)
	getTaskNames (self, dsname, taskname=None, addWildCards=True)

Public Attributes
bool	debug = False
str	paramstyle = None
bool	is_managing_context = False
str	dbtype
	dbname = self.__class__.parseConnectionInfo(connstring)
	dbcon = sqlite3.connect(self.dbname)

Static Public Attributes
dict	StatusCodes
dict	OnDiskCodes

Protected Member Functions
	_createSQLiteSchema (self)
	_createOracleSchema (self)

Detailed Description

TaskManager is tool for keeping track of JobRunner jobs.

Definition at line 120 of file TaskManager.py.

Constructor & Destructor Documentation

__init__()

python.TaskManager.TaskManager.__init__	(		self,
			connstring = '',
			createDatabase = False )

Constructor. connstring is a connection string specifying either a SQLite file
   ("sqlite_file:..."), an Oracle database ("oracle://..."), or an authorization file
   ("auth_file:...") with connection information. If connstring is empty, the connection
   information will be taken from TASKDB (if set), or otherwise defaults to
   'sqlite_file:taskdata.db'.

Definition at line 166 of file TaskManager.py.

    def __init__(self, connstring='', createDatabase=False):
        """Constructor. connstring is a connection string specifying either a SQLite file
           ("sqlite_file:..."), an Oracle database ("oracle://..."), or an authorization file
           ("auth_file:...") with connection information. If connstring is empty, the connection
           information will be taken from TASKDB (if set), or otherwise defaults to
           'sqlite_file:taskdata.db'."""
 
        self.debug = False
        self.paramstyle = None
        self.is_managing_context = False
 
        self.dbtype, self.dbname = self.__class__.parseConnectionInfo(connstring)
 
        if self.dbtype == 'sqlite_file':
            import sqlite3
            self.paramstyle = 'qmark'
            dbexists = os.access(self.dbname, os.F_OK)
            if dbexists and createDatabase:
                raise ValueError ('SQLite file {} exists already - remove before recreating'.format(self.dbname))
            if not (dbexists or createDatabase):
                raise ValueError ('TaskManager database not found (SQLite file {})'.format(self.dbname))
            self.dbcon = sqlite3.connect(self.dbname)
            if createDatabase:
                self._createSQLiteSchema()
        elif self.dbtype == 'oracle':
            import cx_Oracle
            self.paramstyle = 'named'
            try:
                self.dbcon = cx_Oracle.connect(self.dbname)
            except Exception:
                print ('ERROR: First connection attempt to Beam Spot Oracle database failed; will retry in 10s ...')
                time.sleep(10)
                self.dbcon = cx_Oracle.connect(self.dbname)
            if createDatabase:
                self._createOracleSchema()
        else:
            raise ValueError ('Unknown database type: {}'.format(self.dbtype))
 

Member Function Documentation

__enter__()

python.TaskManager.TaskManager.__enter__

(

self

)

Remember that we're inside a 'with' statement so we can warn otherwise: 

Definition at line 204 of file TaskManager.py.

    def __enter__(self):
        ''' Remember that we're inside a 'with' statement so we can warn otherwise: '''
        self.is_managing_context = True
        return self
 

__exit__()

python.TaskManager.TaskManager.__exit__	(		self,
			exc_type,
			exc_value,
			traceback )

Close the database connection at the end of the 'with' statement. 

Definition at line 209 of file TaskManager.py.

    def __exit__(self, exc_type, exc_value, traceback):
        ''' Close the database connection at the end of the 'with' statement. '''
        try:
            self.dbcon.close()
        except Exception:
            print ('ERROR: Unable to close database connection')
 

_createOracleSchema()

python.TaskManager.TaskManager._createOracleSchema ( self )

protected

Create the database schema for an Oracle database.

Definition at line 253 of file TaskManager.py.

    def _createOracleSchema(self):
        """Create the database schema for an Oracle database."""
        try:
            self.dbcon.cursor().execute('drop sequence TASKS_SEQ')
            self.dbcon.cursor().execute('drop trigger TASKS_ID_TRIGGER')
            self.dbcon.cursor().execute('drop table TASKS')
        except Exception as e:
            print (e)
        self.dbcon.cursor().execute("""
            create table tasks (
                TASKID number(10,0),
                DSNAME varchar2(256),
                RUNNR number(10,0),
                TASKNAME varchar2(256),
                TEMPLATE varchar2(256),
                TASKPOSTPROCSTEPS varchar2(256),
                ATLREL varchar2(256),
                CREATED number,
                CREATED_USER varchar2(256),
                CREATED_HOST varchar2(256),
                UPDATED number,
                STATUS number(5,0),
                ONDISK number(5,0),
                LOGFILES varchar2(256),
                COOLTAGS varchar2(256),
                NRESULTFILES number(10,0) default 0,
                NJOBS number(10,0) default 0,
                NJOBS_SUBMITTED number(10,0) default -1,
                NJOBS_RUNNING number(10,0) default -1,
                NJOBS_POSTPROC number(10,0) default -1,
                NJOBS_FAILED number(10,0) default -1,
                NJOBS_COMPLETED number(10,0) default -1,
                RESULTFILES varchar2(4000),
                RESULTLINKS varchar2(4000),
                TASKCOMMENT varchar2(4000),
                STATUS_POSTPROC number(5,0) default -1,
                BASEPATH varchar2(256),
                AUXDATA varchar2(4000),
                constraint BEAMSPOT_TASKS_PK primary key (TASKID)
              )
            """)
        self.dbcon.cursor().execute('create index TASKS_DSNAME_INDX on TASKS(DSNAME)')
        self.dbcon.cursor().execute('create index TASKS_RUNNR_INDX on TASKS(RUNNR)')
        self.dbcon.cursor().execute('alter table TASKS add constraint TASKS_TNAME_DSNAME_UQ  unique (TASKNAME, DSNAME)')
        self.dbcon.cursor().execute('create sequence TASKS_SEQ start with 1 increment by 1 nomaxvalue')
        # The following statement must not have whitespace at the beginning of the line, otherwise
        # the trigger function won't be parsed correctly:
        self.dbcon.cursor().execute("""
create or replace trigger TASKS_ID_TRIGGER
before insert on TASKS
for each row
begin
   select TASKS_SEQ.nextval into :new.TASKID from dual;
end;
""")
        self.dbcon.commit()
 
 

_createSQLiteSchema()

python.TaskManager.TaskManager._createSQLiteSchema ( self )

protected

Create the database schema for a SQLite3 database.

Definition at line 216 of file TaskManager.py.

    def _createSQLiteSchema(self):
        """Create the database schema for a SQLite3 database."""
        self.dbcon.cursor().executescript("""
            create table tasks (
                TASKID integer primary key autoincrement,
                DSNAME text,
                RUNNR integer,
                TASKNAME text,
                TEMPLATE text,
                TASKPOSTPROCSTEPS text,
                ATLREL text,
                CREATED real,
                CREATED_USER text,
                CREATED_HOST text,
                UPDATED real,
                STATUS integer,
                ONDISK integer,
                LOGFILES text,
                COOLTAGS text,
                NRESULTFILES integer default 0,
                NJOBS integer default 0,
                NJOBS_SUBMITTED integer default -1,
                NJOBS_RUNNING integer default -1,
                NJOBS_POSTPROC integer default -1,
                NJOBS_FAILED integer default -1,
                NJOBS_COMPLETED integer default -1,
                RESULTFILES text,
                RESULTLINKS text,
                TASKCOMMENT text,
                STATUS_POSTPROC integer,
                BASEPATH text,
                AUXDATA text
             );
         """)
        self.dbcon.commit()
 
 

addTask()

python.TaskManager.TaskManager.addTask	(		self,
			dsName,
			taskName,
			template,
			release,
			njobs,
			taskpostprocsteps = '',
			status = StatusCodes['SUBMITTED'],
			onDisk = OnDiskCodes['ALLONDISK'],
			createdTime = None,
			createdUser = None,
			createdHost = None,
			comment = '' )

Add an entry for a new task if the task doesn't exist already. If the task exists,
   its UPDATED, NJOBS, STATUS and ONDISK fields will be updated.

Definition at line 395 of file TaskManager.py.

                comment=''):
        """Add an entry for a new task if the task doesn't exist already. If the task exists,
           its UPDATED, NJOBS, STATUS and ONDISK fields will be updated."""
        tstamp = time.time()
        nDefined = self.getNTasks(['where DSNAME =',DbParam(dsName),'and TASKNAME=',DbParam(taskName)])
        updateStatus = self.getNTasks(['where DSNAME = ',DbParam(dsName),'and TASKNAME=',DbParam(taskName),'and STATUS < %i' % TaskManager.StatusCodes['POSTPROCRUNNING']])
 
        if nDefined:
            # Task entry exists already; only update UPDATED, NJOBS, STATUS, ONDISK and possibly ATLREL
            task = self.getTaskDict(dsName,taskName)
            if task['TEMPLATE']!=template:
                print ('ERROR: Must not update task with different template: DSNAME = %s, TASKNAME = %s, templates = %s vs %s' % (dsName,taskName,task['TEMPLATE'],template))
            else:
                updateStr = ['update TASKS set UPDATED =',DbParam(tstamp),
                             ', NJOBS = ',DbParam(task['NJOBS']+njobs),
                             ', ONDISK = ',DbParam(onDisk)]
 
                if release not in task['ATLREL']:
                    print ('WARNING: Updating task using different release: DSNAME = %s, TASKNAME = %s, release = = %s vs %s' % (dsName,taskName,task['ATLREL'],release))
                    release = '; '.join([task['ATLREL'],release])
                    updateStr += [', ATLREL = ',DbParam(release)]
 
                if updateStatus:
                    updateStr += [', STATUS = ',DbParam(status)]
 
                updateStr += ['where DSNAME = ',DbParam(dsName),'and TASKNAME = ',DbParam(taskName)]
 
                self.execute( updateStr)
 
        else:
            # New task entry
            if not createdTime:
                createdTime = tstamp
            if not createdUser:
                createdUser = getUserName()
            if not createdHost:
                createdHost = os.uname()[1]
            runnr = getRunFromName(dsName,None,True)
            self.execute( ['insert into TASKS (DSNAME,RUNNR,TASKNAME,TEMPLATE,TASKPOSTPROCSTEPS,ATLREL,CREATED,CREATED_USER,CREATED_HOST,UPDATED,STATUS,ONDISK,NJOBS,TASKCOMMENT) values (',
                           DbParam(dsName),',',
                           DbParam(runnr),',',
                           DbParam(taskName),',',
                           DbParam(template),',',
                           DbParam(taskpostprocsteps),',',
                           DbParam(release),',',
                           DbParam(createdTime),',',
                           DbParam(createdUser),',',
                           DbParam(createdHost),',',
                           DbParam(tstamp),',',
                           DbParam(status),',',
                           DbParam(onDisk),',',
                           DbParam(njobs),',',
                           DbParam(comment),')'],
                           True)
 
 

deleteTask()

python.TaskManager.TaskManager.deleteTask	(		self,
			dsName,
			taskName )

Delete a task entry from the taskmanager database.

Definition at line 388 of file TaskManager.py.

    def deleteTask(self,dsName,taskName):
        """Delete a task entry from the taskmanager database."""
        q = [ 'delete from TASKS where DSNAME =',DbParam(dsName),'and TASKNAME =',DbParam(taskName) ]
        cursor = self.execute(q,True)
        return cursor.rowcount
 
 

execute()

python.TaskManager.TaskManager.execute	(		self,
			statementParts,
			doCommit = False,
			limit = None )

Execute a SQL statement passed as a list or tuple of statement parts, where each
   part is either a partial SQL string, or an object of type DbParam specifying a parameter
   for the SQL statement. The actual SQL statement is assembled (using the parameter style
   of the current database) and executed, and the resulting cursor is returned.
   Loosely follows the method discussed in the Python Cookbook.
   WARNING: At present, limit doesn't work when ordering rows for Oracle!

Definition at line 311 of file TaskManager.py.

    def execute(self,statementParts, doCommit=False, limit=None):
        """Execute a SQL statement passed as a list or tuple of statement parts, where each
           part is either a partial SQL string, or an object of type DbParam specifying a parameter
           for the SQL statement. The actual SQL statement is assembled (using the parameter style
           of the current database) and executed, and the resulting cursor is returned.
           Loosely follows the method discussed in the Python Cookbook.
           WARNING: At present, limit doesn't work when ordering rows for Oracle!"""
        if not self.is_managing_context:
            print ('**WARN**  TaskManager will keep the database connection open until it is deleted.')
            print ('**INFO**  TaskManager should generally only be used inside a with statement:')
            print ('**INFO**      with TaskManager(...) as taskman:')
            print ('**INFO**        # do something ...')
 
        if not statementParts:
            return None
        if isinstance(statementParts,str):
            raise TypeError ('Must pass list or tuple to TaskManager.execute')
        for p in statementParts:
            if not (isinstance(p,DbParam) or isinstance(p,str)):
                raise ValueError ('Can only pass SQL string fragments and DbParam objects in list'
                                   'to TaskManager.execute, found %s with "%s"' % (type(p),str(p)))
        sqlParts = None
        params = None
 
        if self.paramstyle=='qmark':
            sqlParts = []
            params = []
            for p in statementParts:
                if isinstance(p,DbParam):
                    sqlParts.append('?')
                    params.append(p.value)
                else:
                    sqlParts.append(p)
 
        if self.paramstyle=='named':
            sqlParts = []
            params = {}
            for p in statementParts:
                if isinstance(p,DbParam):
                    name = 'p%d' % len(params)
                    sqlParts.append(':%s' % name)
                    params[name] = p.value
                else:
                    sqlParts.append(p)
 
        if sqlParts is None:
            raise ValueError ('Unknown SQL parameter style %s' % self.paramstyle)
            return None
        sql = ' '.join(sqlParts)
 
        # Handle database-dependent limit clause (PRELIMINARY)
        if limit:
            if self.dbtype=='oracle':
                # WARNING: the following doesn't work always when ordering rows!
                if 'order by' in sql:   # WARNING: works only if order by is in lowercase!
                    sqlParts2 = sql.split('order by')
                    sql = '%s and ROWNUM <= %i order by %s' % (sqlParts2[0],limit,sqlParts2[1])
                else:
                    sql += ' and ROWNUM <= %i' % limit
            else:
                sql += ' limit %i' % limit
 
        if self.debug:
            print ('\nExecuting SQL statement:  ',sql)
            print ('        with parameters:  ',params)
 
        cursor = self.dbcon.cursor()
        try:
            cursor.execute(sql,params)
            if doCommit:
                self.dbcon.commit()
        except Exception as e:
            msg = '\nDatabase error executing SQL statement\n  %s\nusing parameters\n  %s\n%s' % (sql,params,e)
            raise TaskManagerDatabaseError (msg)
        return cursor
 
 

getCount()

python.TaskManager.TaskManager.getCount	(		self,
			what,
			qual = () )

Definition at line 517 of file TaskManager.py.

    def getCount(self, what, qual=()):
        # Except when using count(distinct(...)) which is not supported by earlier sqlite versions,
        # better use getNTasks
        q = [ 'select %s from TASKS' % what ]
        q.extend(qual)
        return len(self.execute(q).fetchall())
 
 

getDSNames()

python.TaskManager.TaskManager.getDSNames	(		self,
			dsname )

Get all DSNAMEs for a given (partial) data set name dsname.

Definition at line 582 of file TaskManager.py.

    def getDSNames(self,dsname):
        """Get all DSNAMEs for a given (partial) data set name dsname."""
        #q = [ "select distinct(DSNAME) from TASKS where DSNAME like '%%%s%%' order by UPDATED" % dsname ]
        q = [ "select distinct(DSNAME) from TASKS where DSNAME like '%%%s%%' order by DSNAME" % dsname ]
        return self.execute(q).fetchall()
 
 

getNTasks()

python.TaskManager.TaskManager.getNTasks	(		self,
			qual = () )

Definition at line 525 of file TaskManager.py.

    def getNTasks(self, qual=()):
        q = [ 'select count(*) from TASKS' ]
        q.extend(qual)
        return self.execute(q).fetchone()[0]
 

getStatus()

python.TaskManager.TaskManager.getStatus	(		self,
			dsName,
			taskName )

Get status of a task.

Definition at line 458 of file TaskManager.py.

    def getStatus(self,dsName,taskName):
        """Get status of a task."""
        cursor = self.execute( ['select STATUS from TASKS where DSNAME =', DbParam(dsName), 'and TASKNAME =', DbParam(taskName) ] )
        return cursor.fetchone()[0]
 
 

getTaskDict()

python.TaskManager.TaskManager.getTaskDict	(		self,
			dsname,
			taskname,
			what = '*',
			qual = () )

Definition at line 561 of file TaskManager.py.

    def getTaskDict(self, dsname, taskname, what='*', qual=()):
        # NOTE: We could implement this more elegantly using row_factory (sqlite3)
        #       or rowfactory (Oracle), but this would come at the expense of a
        #       direct database dependence.
        if qual:
            q = [ 'select %s from TASKS where DSNAME =' % what, DbParam(dsname), 'and TASKNAME =', DbParam(taskname) ]
            q.extend(qual)
        else:
            if self.dbtype=='oracle':
                # WARNING: DOES THIS ALWAYS WORK? There anyway should only be a single task here?
                q = [ 'select %s from TASKS where DSNAME =' % what, DbParam(dsname), 'and TASKNAME = ', DbParam(taskname), 'and ROWNUM <= 1 order by UPDATED' ]
            else:
                q = [ 'select %s from TASKS where DSNAME =' % what, DbParam(dsname), 'and TASKNAME = ', DbParam(taskname), 'order by UPDATED limit 1' ]
        cursor = self.execute(q)
        row = cursor.fetchone()
        if row is None:
            return {}
        else:
            return dictFactory(cursor,row)
 
 

getTaskNames()

python.TaskManager.TaskManager.getTaskNames	(		self,
			dsname,
			taskname = None,
			addWildCards = True )

Find all matching tasks and return list of full (DSNAME,TASKNAME) pairs. dsname and taskname
   may be any partial dataset and task name, respectively.

Definition at line 589 of file TaskManager.py.

    def getTaskNames(self,dsname,taskname=None,addWildCards=True):
        """Find all matching tasks and return list of full (DSNAME,TASKNAME) pairs. dsname and taskname
           may be any partial dataset and task name, respectively."""
        if addWildCards:
            if taskname:
                q = [ "select DSNAME,TASKNAME from TASKS where DSNAME like '%%%s%%' and TASKNAME like '%%%s%%' order by DSNAME,TASKNAME" % (dsname,taskname) ]
            else:
                q = [ "select DSNAME,TASKNAME from TASKS where DSNAME like '%%%s%%' order by DSNAME,TASKNAME" % (dsname) ]
        else:
            if taskname:
                q = [ "select DSNAME,TASKNAME from TASKS where DSNAME like '%s' and TASKNAME like '%s' order by DSNAME,TASKNAME" % (dsname,taskname) ]
            else:
                q = [ "select DSNAME,TASKNAME from TASKS where DSNAME like '%s' order by DSNAME,TASKNAME" % (dsname) ]
        return self.execute(q).fetchall()
 
 

getTaskValue()

python.TaskManager.TaskManager.getTaskValue	(		self,
			dsName,
			taskName,
			what )

Get a single value from the task database. If the query results in more than one value, only
   the first value is returned.

Definition at line 530 of file TaskManager.py.

    def getTaskValue(self, dsName,taskName, what):
        """Get a single value from the task database. If the query results in more than one value, only
           the first value is returned."""
        q = [ 'select %s from TASKS where DSNAME =' % what , DbParam(dsName), 'and TASKNAME =', DbParam(taskName)]
        return self.execute(q).fetchone()[0]
 

getValue()

python.TaskManager.TaskManager.getValue	(		self,
			what,
			qual = () )

Get a single value from the task database. If the query results in more than one value, only
   the first value is returned.

Definition at line 509 of file TaskManager.py.

    def getValue(self, what, qual=()):
        """Get a single value from the task database. If the query results in more than one value, only
           the first value is returned."""
        q = [ 'select %s from TASKS' % what ]
        q.extend(qual)
        return self.execute(q).fetchone()[0]
 
 

parseConnectionInfo()

python.TaskManager.TaskManager.parseConnectionInfo	(		self,
			connstring = '' )

Definition at line 144 of file TaskManager.py.

    def parseConnectionInfo(self, connstring=''):
        if not connstring:
            connstring = os.environ.get('TASKDB', 'sqlite_file:taskdata.db')
 
        try:
            dbtype, dbname = connstring.split(':', 1)
        except Exception:
            raise ValueError ('Illegal database connection string {}'.format(connstring))
 
        if dbtype == 'auth_file':
            # dbname is a file with the actual connection information
            authfile = dbname
            try:
                with open(authfile) as af:
                    connstring = af.read().strip()
                dbtype, dbname = connstring.split(':', 1)
            except Exception:
                raise ValueError ('Invalid authorization file {} (not readable or invalid format)'.format(authfile))
 
        return dbtype, dbname
 
 

setDiskStatus()

python.TaskManager.TaskManager.setDiskStatus	(		self,
			dsName,
			taskName,
			onDisk )

Update the onDisk-status of a task.

Definition at line 476 of file TaskManager.py.

    def setDiskStatus(self,dsName,taskName,onDisk):
        """Update the onDisk-status of a task."""
        tstamp = time.time()
        cursor = self.execute( ['update TASKS set UPDATED =',DbParam(tstamp),', ONDISK =',DbParam(onDisk),
                                'where DSNAME =', DbParam(dsName), 'and TASKNAME =', DbParam(taskName)], True)
        return cursor.rowcount
 
 

setStatus()

python.TaskManager.TaskManager.setStatus	(		self,
			dsName,
			taskName,
			status,
			oldStatus = None )

Set the status of a task. If oldStatus is set, the task must be in the designated status.

Definition at line 464 of file TaskManager.py.

    def setStatus(self,dsName,taskName,status,oldStatus=None):
        """Set the status of a task. If oldStatus is set, the task must be in the designated status."""
        tstamp = time.time()
        if oldStatus:
            cursor = self.execute( ['update TASKS set UPDATED =',DbParam(tstamp),', STATUS =',DbParam(status),
                                    'where DSNAME =', DbParam(dsName), 'and TASKNAME =', DbParam(taskName),'and STATUS =',DbParam(oldStatus)], True)
        else:
            cursor = self.execute( ['update TASKS set UPDATED =',DbParam(tstamp),', STATUS =',DbParam(status),
                                    'where DSNAME =', DbParam(dsName), 'and TASKNAME =', DbParam(taskName)], True)
        return cursor.rowcount
 
 

setValue()

python.TaskManager.TaskManager.setValue	(		self,
			dsName,
			taskName,
			fieldName,
			value )

Update a field of a given task.

Definition at line 501 of file TaskManager.py.

    def setValue(self,dsName,taskName,fieldName,value):
        """Update a field of a given task."""
        tstamp = time.time()
        cursor = self.execute( ['update TASKS set UPDATED =',DbParam(tstamp),', %s =' % fieldName,DbParam(value),
                                'where DSNAME =', DbParam(dsName), 'and TASKNAME =', DbParam(taskName)], True)
        return cursor.rowcount
 
 

taskIter()

python.TaskManager.TaskManager.taskIter	(		self,
			what = '*',
			qual = ('order by UPDATED',) )

Definition at line 536 of file TaskManager.py.

    def taskIter(self, what='*', qual=('order by UPDATED',)):
        q = [ 'select %s from TASKS' % what ]
        q.extend(qual)
        cursor = self.execute(q)
        return iter(cursor.fetchone,None)
 
 

taskIterDict()

python.TaskManager.TaskManager.taskIterDict	(		self,
			what = '*',
			qual = ('order by UPDATED',),
			limit = 999999999 )

Definition at line 543 of file TaskManager.py.

    def taskIterDict(self, what='*', qual=('order by UPDATED',), limit=999999999):
        # NOTE: We could implement this more elegantly using row_factory (sqlite3)
        #       or rowfactory (Oracle), but this would come at the expense of a
        #       direct database dependence. Also, the primitive limit on the
        #       number of items returned by the generator avoids database-
        #       dependent SQL limit statements (until limit in TaskManager.execute
        #       is fully implemented also for Oracle.
        q = [ 'select %s from TASKS' % what ]
        q.extend(qual)
        cursor = self.execute(q)
        n = 0
        while n<limit:
            row = cursor.fetchone()
            if row is None: break
            n += 1
            yield dictFactory(cursor,row)
 
 

updateStatus()

python.TaskManager.TaskManager.updateStatus	(		self,
			dsName,
			taskName,
			status,
			nResultFiles,
			nJobs,
			nJobsSubmitted,
			nJobsRunning,
			nJobsPostProc,
			nJobsFailed,
			nJobsCompleted )

Update task status information including number of jobs in different states.

Definition at line 484 of file TaskManager.py.

                     status,nResultFiles,nJobs,nJobsSubmitted,nJobsRunning,nJobsPostProc,nJobsFailed,nJobsCompleted):
        """Update task status information including number of jobs in different states."""
        tstamp = time.time()
        cursor = self.execute( ['update TASKS set UPDATED =',DbParam(tstamp),
                                ', STATUS =',DbParam(status),
                                ', NRESULTFILES =',DbParam(nResultFiles),
                                ', NJOBS =',DbParam(nJobs),
                                ', NJOBS_SUBMITTED =',DbParam(nJobsSubmitted),
                                ', NJOBS_RUNNING =',DbParam(nJobsRunning),
                                ', NJOBS_POSTPROC =',DbParam(nJobsPostProc),
                                ', NJOBS_FAILED =',DbParam(nJobsFailed),
                                ', NJOBS_COMPLETED =',DbParam(nJobsCompleted),
                                'where DSNAME =', DbParam(dsName), 'and TASKNAME =', DbParam(taskName)], True)
        return cursor.rowcount
 
 

Member Data Documentation

dbcon

python.TaskManager.TaskManager.dbcon = sqlite3.connect(self.dbname)

Definition at line 187 of file TaskManager.py.

dbname

python.TaskManager.TaskManager.dbname = self.__class__.parseConnectionInfo(connstring)

Definition at line 177 of file TaskManager.py.

dbtype

python.TaskManager.TaskManager.dbtype

Definition at line 177 of file TaskManager.py.

debug

bool python.TaskManager.TaskManager.debug = False

Definition at line 173 of file TaskManager.py.

is_managing_context

bool python.TaskManager.TaskManager.is_managing_context = False

Definition at line 175 of file TaskManager.py.

OnDiskCodes

dict python.TaskManager.TaskManager.OnDiskCodes

static

Initial value:

=  { 'UNKNOWN': -1,
                    'ALLONDISK': 0,
                    'RESULTSONDISK': 1,
                    # Codes above are for jobs whose task directory is on disk
                    'ARCHIVED': 10,
                    'DELETED': 11 }

Definition at line 136 of file TaskManager.py.

paramstyle

python.TaskManager.TaskManager.paramstyle = None

Definition at line 174 of file TaskManager.py.

StatusCodes

dict python.TaskManager.TaskManager.StatusCodes

static

Initial value:

=  { 'UNKNOWN': -1,
                    'CONFIGURED': 0,
                    'SUBMITTED': 1,
                    'RUNNING':   2,
                    'PARTIALFAILED': 3,
                    # Codes above are considered tasks whose status needs to
                    # be monitored and updated
                    'POSTPROCESSING': 5,
                    'POSTPROCRUNNING': 6,
                    'POSTPROCFAILED': 7,
                    'COMPLETED': 10,
                    'FAILED': 11}

Definition at line 123 of file TaskManager.py.

---

The documentation for this class was generated from the following file:

• TaskManager.py