CRAB/python/DBinterface.py

from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
import common
import os, time, shutil
import traceback

from ProdCommon.BossLite.API.BossLiteAPI import BossLiteAPI
from ProdCommon.BossLite.Common.Exceptions import DbError
from ProdCommon.BossLite.Common.Exceptions import TaskError

from ProdCommon.BossLite.DbObjects.Job import Job
from ProdCommon.BossLite.DbObjects.Task import Task
from ProdCommon.BossLite.DbObjects.RunningJob import RunningJob


class DBinterface:
    def __init__(self, cfg_params):

        self.cfg_params = cfg_params

        self.db_type =  cfg_params.get("USER.use_db",'SQLite')
        return


    def configureDB(self):

        dbname = common.work_space.shareDir()+'crabDB'
        dbConfig = {'dbName':dbname
            }
        try: 
            common.bossSession = BossLiteAPI( self.db_type, dbConfig)
        except Exception, e :
            raise CrabException('Istantiate DB Session : '+str(e))

        try:
            common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')     
        except Exception, e :
            raise CrabException('DB Installation error : '+str(e))
        return 
 
    def loadDB(self):

        dbname = common.work_space.shareDir()+'crabDB'
        dbConfig = {'dbName':dbname
            }
        try:
            common.bossSession = BossLiteAPI( self.db_type, dbConfig)
        except Exception, e :
            raise CrabException('Istantiate DB Session : '+str(e))

        return
 
    def getTask(self, jobsList='all'): 
        """
        Return task with all/list of jobs 
        """
        try:
            task = common.bossSession.load(1,jobsList)[0]
        except Exception, e :
            common.logger.debug(3, "Error while getting task : " +str(traceback.format_exc()))
            raise CrabException('Error while getting task '+str(e))
        return task

    def getJob(self, n): 
        """
        Return a task with a single job 
        """ 
        try:
            task = common.bossSession.load(1,str(n))[0]
        except Exception, e :
            common.logger.debug(3, "Error while getting job : " +str(traceback.format_exc()))
            raise CrabException('Error while getting job '+str(e))
        return task


    def createTask_(self, optsToSave):       
        """
        Task declaration
        with the first coniguration stuff 
        """
        opt={}
        if optsToSave.get('server_mode',0) == 1: opt['serverName']=optsToSave['server_name'] 
        opt['name']=common.work_space.taskName()  
        task = Task( opt )
        try:
            common.bossSession.saveTask( task )
        except Exception, e :
           # common.logger.debug(3, "Error creating task : " +str(traceback.format_exc()))
           # raise CrabException('Error creating task '+str(e))
            raise CrabException('Error creating task '+str(traceback.format_exc()))
            
        return 

    def updateTask_(self,optsToSave):       
        """
        Update task fields   
        """
        task = self.getTask()
   
        for key in optsToSave.keys():
            task[key] = optsToSave[key]
        try:
            common.bossSession.updateDB( task )
        except Exception, e :
            raise CrabException('Error updating task '+str(traceback.format_exc()))

        return 

    def createJobs_(self, jobsL):
        """  
        Fill crab DB with  the jobs filed 
        """
        task = self.getTask()

        jobs = [] 
        for id in jobsL:
            parameters = {}
            parameters['jobId'] =  str(id)
            parameters['name'] = task['name'] + '_' + 'job' + str(id)
            job = Job(parameters)
            jobs.append(job)  
        task.addJobs(jobs)
        try:
            common.bossSession.updateDB( task )
        except Exception, e :
            raise CrabException('Error updating task '+str(traceback.format_exc()))

        return

    def updateJob_(self, jobsL, optsToSave):       
        """
        Update Job fields   
        """
        task = self.getTask(jobsL)
        id =0 
        for job in task.jobs:
            for key in optsToSave[id].keys():
                job[key] = optsToSave[id][key]
            id+=1
        try:
            common.bossSession.updateDB( task )
        except Exception, e :
            raise CrabException('Error updating task '+str(traceback.format_exc()))
        return 

    def updateRunJob_(self, jobsL, optsToSave):       
        """
        Update Running Job fields   
        """
        task = self.getTask(jobsL)

        id=0
        for job in task.jobs:
            common.bossSession.getRunningInstance(job)
            for key in optsToSave[id].keys():
                job.runningJob[key] = optsToSave[id][key]
            id+=1
        common.bossSession.updateDB( task )
        return 

    def nJobs(self,list=''):
        
        task = self.getTask()
        listId=[]
        if list == 'list':
            for job in task.jobs:listId.append(int(job['jobId']))  
            return listId
        else:
            return len(task.jobs) 

    def dump(self,jobs):
        """
         List a complete set of infos for a job/range of jobs   
        """
        task = self.getTask()

        njobs = len(jobs)
        lines=[] 
        header=''
     #   ##query the DB asking the right infos for runningJobs  TODO  DS
     #   for job in jobs:
     #       ## here the query over runngJobs  
     #       pass


     #   ##Define Header to show and Pass the query results,
     #   ##  header and format to displayReport()   TODO  DS
     #   if njobs == 1: plural = ''
     #   else:          plural = 's'
     #   header += 'Listing %d job%s:\n' % (njobs, plural) 
     #   header += ' :\n' % (---) ## TODO DS 

     #   displayReport(header, lines):
        return      

    def serializeTask(self, tmp_task = None):
        if tmp_task is None:
            tmp_task = self.getTask()
        return common.bossSession.serialize(tmp_task)   
 
    def queryID(self,server_mode=0):
        '''
        Return the taskId if serevr_mode =1 
        Return the joblistId if serevr_mode =0 
        '''     
        header=''
        lines=[]
        task = self.getTask()
        if server_mode == 1:
            header= "Task Id = %-40s " %(task['name'])
        else:
            for job in task.jobs: 
                toPrint=''
                common.bossSession.getRunningInstance(job)
                toPrint = "%-5s %-50s " % (job['id'],job.runningJob['schedulerId'])
                lines.append(toPrint)
            header+= "%-5s %-50s " % ('Job:','ID' ) 
        displayReport(self,header,lines)
        return   

    def queryTask(self,attr):
        '''
        Perform a query over a generic task attribute
        '''
        task = self.getTask()
        return task[attr]

    def queryJob(self, attr, jobsL):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        task = self.getTask(jobsL)
        for job in task.jobs:
            lines.append(eval(job[attr]))
        return lines

    def queryRunJob(self, attr, jobsL):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        task = self.getTask(jobsL)
        for job in task.jobs:
            common.bossSession.getRunningInstance(job)
            lines.append(job.runningJob[attr])
        return lines

    def queryDistJob(self, attr):
        '''
        Returns the list of distinct value for a given job attributes 
        '''
        distAttr=[]
        try:
            task = common.bossSession.loadJobDist( 1, attr ) 
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By distinct Attr '+str(e))

        for i in task: distAttr.append(eval(i[attr]))   
        return  distAttr

    def queryDistJob_Attr(self, attr_1, attr_2, list):
        '''
        Returns the list of distinct value for a given job attribute 
        '''
        distAttr=[]
        try:
            task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list ) 
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By distinct Attr '+str(e))

        for i in task: distAttr.append(eval(i[attr_1]))   
        return  distAttr

    def queryAttrJob(self, attr, field):
        '''
        Returns the list of jobs matching the given attribute
        '''
        matched=[]
        try:
            task = common.bossSession.loadJobsByAttr(attr ) 
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By Attr '+str(e))
        for i in task:
            matched.append(i[field])
        return  matched


    def queryAttrRunJob(self, attr,field):
        '''
        Returns the list of jobs matching the given attribute
        '''
        matched=[]
        try:
            task = common.bossSession.loadJobsByRunningAttr(attr)
        except Exception, e :
            common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
            raise CrabException('Error loading Jobs By Running Attr '+str(e))
        for i in task:
            matched.append(i[field])
        return matched 
Revision:	1.15
Committed:	Tue Apr 15 14:13:07 2008 UTC (17 years ago) by mcinquil
Content type:	text/x-python
Branch:	MAIN
Changes since 1.14:	+1 -1 lines
Log Message:	Added taskname in front of job name
#	Content
1	from crab_logger import Logger
2	from crab_exceptions import *
3	from crab_util import *
4	import common
5	import os, time, shutil
6	import traceback
7
8	from ProdCommon.BossLite.API.BossLiteAPI import BossLiteAPI
9	from ProdCommon.BossLite.Common.Exceptions import DbError
10	from ProdCommon.BossLite.Common.Exceptions import TaskError
11
12	from ProdCommon.BossLite.DbObjects.Job import Job
13	from ProdCommon.BossLite.DbObjects.Task import Task
14	from ProdCommon.BossLite.DbObjects.RunningJob import RunningJob
15
16
17	class DBinterface:
18	def __init__(self, cfg_params):
19
20	self.cfg_params = cfg_params
21
22	self.db_type = cfg_params.get("USER.use_db",'SQLite')
23	return
24
25
26	def configureDB(self):
27
28	dbname = common.work_space.shareDir()+'crabDB'
29	dbConfig = {'dbName':dbname
30	}
31	try:
32	common.bossSession = BossLiteAPI( self.db_type, dbConfig)
33	except Exception, e :
34	raise CrabException('Istantiate DB Session : '+str(e))
35
36	try:
37	common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
38	except Exception, e :
39	raise CrabException('DB Installation error : '+str(e))
40	return
41
42	def loadDB(self):
43
44	dbname = common.work_space.shareDir()+'crabDB'
45	dbConfig = {'dbName':dbname
46	}
47	try:
48	common.bossSession = BossLiteAPI( self.db_type, dbConfig)
49	except Exception, e :
50	raise CrabException('Istantiate DB Session : '+str(e))
51
52	return
53
54	def getTask(self, jobsList='all'):
55	"""
56	Return task with all/list of jobs
57	"""
58	try:
59	task = common.bossSession.load(1,jobsList)[0]
60	except Exception, e :
61	common.logger.debug(3, "Error while getting task : " +str(traceback.format_exc()))
62	raise CrabException('Error while getting task '+str(e))
63	return task
64
65	def getJob(self, n):
66	"""
67	Return a task with a single job
68	"""
69	try:
70	task = common.bossSession.load(1,str(n))[0]
71	except Exception, e :
72	common.logger.debug(3, "Error while getting job : " +str(traceback.format_exc()))
73	raise CrabException('Error while getting job '+str(e))
74	return task
75
76
77	def createTask_(self, optsToSave):
78	"""
79	Task declaration
80	with the first coniguration stuff
81	"""
82	opt={}
83	if optsToSave.get('server_mode',0) == 1: opt['serverName']=optsToSave['server_name']
84	opt['name']=common.work_space.taskName()
85	task = Task( opt )
86	try:
87	common.bossSession.saveTask( task )
88	except Exception, e :
89	# common.logger.debug(3, "Error creating task : " +str(traceback.format_exc()))
90	# raise CrabException('Error creating task '+str(e))
91	raise CrabException('Error creating task '+str(traceback.format_exc()))
92
93	return
94
95	def updateTask_(self,optsToSave):
96	"""
97	Update task fields
98	"""
99	task = self.getTask()
100
101	for key in optsToSave.keys():
102	task[key] = optsToSave[key]
103	try:
104	common.bossSession.updateDB( task )
105	except Exception, e :
106	raise CrabException('Error updating task '+str(traceback.format_exc()))
107
108	return
109
110	def createJobs_(self, jobsL):
111	"""
112	Fill crab DB with the jobs filed
113	"""
114	task = self.getTask()
115
116	jobs = []
117	for id in jobsL:
118	parameters = {}
119	parameters['jobId'] = str(id)
120	parameters['name'] = task['name'] + '_' + 'job' + str(id)
121	job = Job(parameters)
122	jobs.append(job)
123	task.addJobs(jobs)
124	try:
125	common.bossSession.updateDB( task )
126	except Exception, e :
127	raise CrabException('Error updating task '+str(traceback.format_exc()))
128
129	return
130
131	def updateJob_(self, jobsL, optsToSave):
132	"""
133	Update Job fields
134	"""
135	task = self.getTask(jobsL)
136	id =0
137	for job in task.jobs:
138	for key in optsToSave[id].keys():
139	job[key] = optsToSave[id][key]
140	id+=1
141	try:
142	common.bossSession.updateDB( task )
143	except Exception, e :
144	raise CrabException('Error updating task '+str(traceback.format_exc()))
145	return
146
147	def updateRunJob_(self, jobsL, optsToSave):
148	"""
149	Update Running Job fields
150	"""
151	task = self.getTask(jobsL)
152
153	id=0
154	for job in task.jobs:
155	common.bossSession.getRunningInstance(job)
156	for key in optsToSave[id].keys():
157	job.runningJob[key] = optsToSave[id][key]
158	id+=1
159	common.bossSession.updateDB( task )
160	return
161
162	def nJobs(self,list=''):
163
164	task = self.getTask()
165	listId=[]
166	if list == 'list':
167	for job in task.jobs:listId.append(int(job['jobId']))
168	return listId
169	else:
170	return len(task.jobs)
171
172	def dump(self,jobs):
173	"""
174	List a complete set of infos for a job/range of jobs
175	"""
176	task = self.getTask()
177
178	njobs = len(jobs)
179	lines=[]
180	header=''
181	# ##query the DB asking the right infos for runningJobs TODO DS
182	# for job in jobs:
183	# ## here the query over runngJobs
184	# pass
185
186
187	# ##Define Header to show and Pass the query results,
188	# ## header and format to displayReport() TODO DS
189	# if njobs == 1: plural = ''
190	# else: plural = 's'
191	# header += 'Listing %d job%s:\n' % (njobs, plural)
192	# header += ' :\n' % (---) ## TODO DS
193
194	# displayReport(header, lines):
195	return
196
197	def serializeTask(self, tmp_task = None):
198	if tmp_task is None:
199	tmp_task = self.getTask()
200	return common.bossSession.serialize(tmp_task)
201
202	def queryID(self,server_mode=0):
203	'''
204	Return the taskId if serevr_mode =1
205	Return the joblistId if serevr_mode =0
206	'''
207	header=''
208	lines=[]
209	task = self.getTask()
210	if server_mode == 1:
211	header= "Task Id = %-40s " %(task['name'])
212	else:
213	for job in task.jobs:
214	toPrint=''
215	common.bossSession.getRunningInstance(job)
216	toPrint = "%-5s %-50s " % (job['id'],job.runningJob['schedulerId'])
217	lines.append(toPrint)
218	header+= "%-5s %-50s " % ('Job:','ID' )
219	displayReport(self,header,lines)
220	return
221
222	def queryTask(self,attr):
223	'''
224	Perform a query over a generic task attribute
225	'''
226	task = self.getTask()
227	return task[attr]
228
229	def queryJob(self, attr, jobsL):
230	'''
231	Perform a query for a range/all/single job
232	over a generic job attribute
233	'''
234	lines=[]
235	task = self.getTask(jobsL)
236	for job in task.jobs:
237	lines.append(eval(job[attr]))
238	return lines
239
240	def queryRunJob(self, attr, jobsL):
241	'''
242	Perform a query for a range/all/single job
243	over a generic job attribute
244	'''
245	lines=[]
246	task = self.getTask(jobsL)
247	for job in task.jobs:
248	common.bossSession.getRunningInstance(job)
249	lines.append(job.runningJob[attr])
250	return lines
251
252	def queryDistJob(self, attr):
253	'''
254	Returns the list of distinct value for a given job attributes
255	'''
256	distAttr=[]
257	try:
258	task = common.bossSession.loadJobDist( 1, attr )
259	except Exception, e :
260	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
261	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
262
263	for i in task: distAttr.append(eval(i[attr]))
264	return distAttr
265
266	def queryDistJob_Attr(self, attr_1, attr_2, list):
267	'''
268	Returns the list of distinct value for a given job attribute
269	'''
270	distAttr=[]
271	try:
272	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
273	except Exception, e :
274	common.logger.debug(3, "Error loading Jobs By distinct Attr : " +str(traceback.format_exc()))
275	raise CrabException('Error loading Jobs By distinct Attr '+str(e))
276
277	for i in task: distAttr.append(eval(i[attr_1]))
278	return distAttr
279
280	def queryAttrJob(self, attr, field):
281	'''
282	Returns the list of jobs matching the given attribute
283	'''
284	matched=[]
285	try:
286	task = common.bossSession.loadJobsByAttr(attr )
287	except Exception, e :
288	common.logger.debug(3, "Error loading Jobs By Attr : " +str(traceback.format_exc()))
289	raise CrabException('Error loading Jobs By Attr '+str(e))
290	for i in task:
291	matched.append(i[field])
292	return matched
293
294
295	def queryAttrRunJob(self, attr,field):
296	'''
297	Returns the list of jobs matching the given attribute
298	'''
299	matched=[]
300	try:
301	task = common.bossSession.loadJobsByRunningAttr(attr)
302	except Exception, e :
303	common.logger.debug(3, "Error loading Jobs By Running Attr : " +str(traceback.format_exc()))
304	raise CrabException('Error loading Jobs By Running Attr '+str(e))
305	for i in task:
306	matched.append(i[field])
307	return matched