CRAB/python/DBinterface.py

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

from ProdCommon.BossLite.API.BossLiteAPI import BossLiteAPI


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
            }
 
        common.bossSession = BossLiteAPI( self.db_type, dbConfig)
        common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')     
        
        return

    def loadDB(self):

        dbname = common.work_space.shareDir()+'crabDB'
        dbConfig = {'dbName':dbname
            }
        common.bossSession = BossLiteAPI( self.db_type, dbConfig)
        self.task = common.bossSession.load(1)[0]
        return
 
    def getTask(self, jobsList='all'): #, cfg_params):
        """
        Return task with all/list of jobs 
        """

        if jobsList == 'all':
            task = common.bossSession.load(1)[0]
        else: 
            if len(jobsList)>1: str_jobs=string.join(map(str,jobsList),",") 
            else: str_jobs=str(jobsList)  
            task = common.bossSession.load(1,jobsList)[0]
        return task

    def getJob(self, n): 
        """
        Return a task with a single job 
        """ 
        task = common.bossSession.load(1,str(n))[0]
        return task


    def createTask_(self, optsToSave):       
        """
        Task declaration
        with the first coniguration stuff 
        """
        opt={}
        opt['serverName']=optsToSave['server_name'] 
        opt[ 'name']=common.work_space.taskName()  
        task = Task( opt )
      
        common.bossSession.saveTask( task )
        return 

    def updateTask_(self,optsToSave):       
        """
        Update task fields   
        """
        task = common.bossSession.load(1)[0]
         
        for key in optsToSave.keys():
            task[key] = optsToSave[key]
        common.bossSession.updateDB( task )
        return 

    def createJobs_(self, jobsL):
        """  
        Fill crab DB with  the jobs filed 
        """
        task = common.bossSession.loadTask(1) 
        jobs = [] 
        for id in jobsL:
            parameters = {}
            parameters['jobId'] =  str(id)
            parameters['name'] = 'job' + str(id)
            job = Job(parameters)
            jobs.append(job)  
        task.addJobs(jobs)
        common.bossSession.updateDB( task )
        return

    def updateJob_(self, jobsL, optsToSave):       
        """
        Update Job fields   
        """
        if len(jobsL)>1: str_jobs=string.join(map(str,jobsL),",")
        else: str_jobs=str(jobsL)
        task = common.bossSession.load(1,jobsL)[0]
        id =0 
        for job in task.jobs:
            for key in optsToSave[id].keys():
                job[key] = optsToSave[id][key]
            id+=1
        common.bossSession.updateDB( task )
        return 

    def updateRunJob_(self, jobsL, optsToSave):       
        """
        Update Running Job fields   
        """
        if len(jobsL)>1: str_jobs=string.join(map(str,jobsL),",")
        else: str_jobs=str(jobsL)
        task = common.bossSession.load(1,jobsL)[0]
        for job in task.jobs:
            common.bossSession.getRunningInstance(job)
            for key in optsToSave.keys():
                job.runningJob[key] = optsToSave[key]
        common.bossSession.updateDB( task )
        return 

    def nJobs(self,list=''):
        
        task = common.bossSession.load(1)[0]
        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 = common.bossSession.load(1)[0]

        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 = common.bossSession.loadTaskByID(1)
            tmp_task = common.bossSession.load(1)[0]
        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 = common.bossSession.load(1)[0]
        if server_mode == 1:
            header= "Task Id = %-40s " %(task['name'])
        else:
            for i in range(len(task.job)): 
                common.bossSession.getRunningInstance(task.jobs[i])
                lines.append(task.jobs[i].runningJob['schedulerId'])
           
            header+= "Job: %-5s Id = %-40s: \n" 
        displayReport(header,lines)
        return   

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

    def queryJob(self, attr, jobs):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        str_jobs=string.join(map(str,jobs),",") 
        task = common.bossSession.load(1,str_jobs)[0]
        for job in task.jobs:
            lines.append(eval(job[attr]))
        return lines

    def queryRunJob(self, attr, jobs):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        str_jobs=string.join(map(str,jobs),",") 
        task = common.bossSession.load(1,str_jobs)[0]
        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=[]
        task = common.bossSession.loadJobDist( 1, attr ) 
        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=[]
        task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list ) 
        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=[]
        task = common.bossSession.loadJobsByAttr(attr ) 
        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=[]
        task = common.bossSession.loadJobsByRunningAttr(attr)
        for i in task:
            matched.append(i[field])
        return matched 
Revision:	1.10
Committed:	Tue Mar 25 16:29:12 2008 UTC (17 years, 1 month ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.9:	+13 -7 lines
Log Message:	still another fix on the job index plus bosslite integration optimization.
#	User	Rev	Content
1	spiga	1.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
7			from ProdCommon.BossLite.API.BossLiteAPI import BossLiteAPI
8
9
10			from ProdCommon.BossLite.DbObjects.Job import Job
11			from ProdCommon.BossLite.DbObjects.Task import Task
12			from ProdCommon.BossLite.DbObjects.RunningJob import RunningJob
13
14
15			class DBinterface:
16	spiga	1.2	def __init__(self, cfg_params):
17	spiga	1.1
18	spiga	1.2	self.cfg_params = cfg_params
19
20			self.db_type = cfg_params.get("USER.use_db",'SQLite')
21	spiga	1.1	return
22
23
24	spiga	1.2	def configureDB(self):
25	spiga	1.1
26			dbname = common.work_space.shareDir()+'crabDB'
27			dbConfig = {'dbName':dbname
28			}
29
30			common.bossSession = BossLiteAPI( self.db_type, dbConfig)
31			common.bossSession.installDB('$CRABPRODCOMMONPYTHON/ProdCommon/BossLite/DbObjects/setupDatabase-sqlite.sql')
32
33			return
34
35	spiga	1.2	def loadDB(self):
36
37			dbname = common.work_space.shareDir()+'crabDB'
38			dbConfig = {'dbName':dbname
39			}
40			common.bossSession = BossLiteAPI( self.db_type, dbConfig)
41	spiga	1.9	self.task = common.bossSession.load(1)[0]
42	spiga	1.2	return
43	spiga	1.4
44	spiga	1.7	def getTask(self, jobsList='all'): #, cfg_params):
45	spiga	1.9	"""
46			Return task with all/list of jobs
47			"""
48	spiga	1.4
49	spiga	1.9	if jobsList == 'all':
50			task = common.bossSession.load(1)[0]
51			else:
52			if len(jobsList)>1: str_jobs=string.join(map(str,jobsList),",")
53			else: str_jobs=str(jobsList)
54			task = common.bossSession.load(1,jobsList)[0]
55			return task
56	spiga	1.2
57	slacapra	1.6	def getJob(self, n):
58	spiga	1.9	"""
59			Return a task with a single job
60			"""
61			task = common.bossSession.load(1,str(n))[0]
62			return task
63	spiga	1.2
64	spiga	1.1
65			def createTask_(self, optsToSave):
66			"""
67			Task declaration
68			with the first coniguration stuff
69			"""
70			opt={}
71	spiga	1.9	opt['serverName']=optsToSave['server_name']
72	spiga	1.1	opt[ 'name']=common.work_space.taskName()
73			task = Task( opt )
74
75			common.bossSession.saveTask( task )
76			return
77
78			def updateTask_(self,optsToSave):
79			"""
80			Update task fields
81			"""
82	spiga	1.9	task = common.bossSession.load(1)[0]
83	spiga	1.1
84			for key in optsToSave.keys():
85			task[key] = optsToSave[key]
86	spiga	1.9	common.bossSession.updateDB( task )
87	spiga	1.1	return
88
89	spiga	1.9	def createJobs_(self, jobsL):
90	spiga	1.1	"""
91			Fill crab DB with the jobs filed
92			"""
93	spiga	1.9	task = common.bossSession.loadTask(1)
94	spiga	1.1	jobs = []
95	spiga	1.9	for id in jobsL:
96	spiga	1.1	parameters = {}
97	spiga	1.9	parameters['jobId'] = str(id)
98	spiga	1.7	parameters['name'] = 'job' + str(id)
99	spiga	1.1	job = Job(parameters)
100	spiga	1.9	jobs.append(job)
101	spiga	1.1	task.addJobs(jobs)
102			common.bossSession.updateDB( task )
103			return
104
105	spiga	1.9	def updateJob_(self, jobsL, optsToSave):
106	spiga	1.1	"""
107			Update Job fields
108			"""
109	spiga	1.10	if len(jobsL)>1: str_jobs=string.join(map(str,jobsL),",")
110			else: str_jobs=str(jobsL)
111			task = common.bossSession.load(1,jobsL)[0]
112			id =0
113			for job in task.jobs:
114	spiga	1.9	for key in optsToSave[id].keys():
115	spiga	1.10	job[key] = optsToSave[id][key]
116			id+=1
117	spiga	1.7	common.bossSession.updateDB( task )
118	spiga	1.1	return
119
120	spiga	1.9	def updateRunJob_(self, jobsL, optsToSave):
121	spiga	1.1	"""
122			Update Running Job fields
123			"""
124	spiga	1.10	if len(jobsL)>1: str_jobs=string.join(map(str,jobsL),",")
125			else: str_jobs=str(jobsL)
126			task = common.bossSession.load(1,jobsL)[0]
127			for job in task.jobs:
128			common.bossSession.getRunningInstance(job)
129	spiga	1.7	for key in optsToSave.keys():
130	spiga	1.10	job.runningJob[key] = optsToSave[key]
131	spiga	1.1	common.bossSession.updateDB( task )
132			return
133
134	spiga	1.9	def nJobs(self,list=''):
135	spiga	1.2
136	spiga	1.9	task = common.bossSession.load(1)[0]
137			listId=[]
138			if list == 'list':
139			for job in task.jobs:listId.append(int(job['jobId']))
140			return listId
141			else:
142			return len(task.jobs)
143	spiga	1.1
144			def dump(self,jobs):
145			"""
146			List a complete set of infos for a job/range of jobs
147			"""
148	spiga	1.9	task = common.bossSession.load(1)[0]
149	spiga	1.1
150			njobs = len(jobs)
151			lines=[]
152			header=''
153			# ##query the DB asking the right infos for runningJobs TODO DS
154			# for job in jobs:
155			# ## here the query over runngJobs
156			# pass
157
158
159			# ##Define Header to show and Pass the query results,
160			# ## header and format to displayReport() TODO DS
161			# if njobs == 1: plural = ''
162			# else: plural = 's'
163			# header += 'Listing %d job%s:\n' % (njobs, plural)
164			# header += ' :\n' % (---) ## TODO DS
165
166			# displayReport(header, lines):
167			return
168	farinafa	1.8
169			def serializeTask(self, tmp_task = None):
170			if tmp_task is None:
171	spiga	1.9	#tmp_task = common.bossSession.loadTaskByID(1)
172			tmp_task = common.bossSession.load(1)[0]
173	farinafa	1.8	return common.bossSession.serialize(tmp_task)
174	spiga	1.1
175			def queryID(self,server_mode=0):
176			'''
177			Return the taskId if serevr_mode =1
178			Return the joblistId if serevr_mode =0
179			'''
180			header=''
181			lines=[]
182	spiga	1.9	task = common.bossSession.load(1)[0]
183	spiga	1.1	if server_mode == 1:
184	spiga	1.2	header= "Task Id = %-40s " %(task['name'])
185	spiga	1.1	else:
186			for i in range(len(task.job)):
187			common.bossSession.getRunningInstance(task.jobs[i])
188			lines.append(task.jobs[i].runningJob['schedulerId'])
189
190			header+= "Job: %-5s Id = %-40s: \n"
191			displayReport(header,lines)
192			return
193
194			def queryTask(self,attr):
195			'''
196			Perform a query over a generic task attribute
197			'''
198	spiga	1.9	task = common.bossSession.loadTask(1)
199	spiga	1.1	return task[attr]
200
201	spiga	1.9	def queryJob(self, attr, jobs):
202	spiga	1.1	'''
203			Perform a query for a range/all/single job
204			over a generic job attribute
205			'''
206			lines=[]
207	spiga	1.9	str_jobs=string.join(map(str,jobs),",")
208			task = common.bossSession.load(1,str_jobs)[0]
209			for job in task.jobs:
210			lines.append(eval(job[attr]))
211	spiga	1.1	return lines
212
213			def queryRunJob(self, attr, jobs):
214			'''
215			Perform a query for a range/all/single job
216			over a generic job attribute
217			'''
218			lines=[]
219	spiga	1.9	str_jobs=string.join(map(str,jobs),",")
220			task = common.bossSession.load(1,str_jobs)[0]
221			for job in task.jobs:
222			common.bossSession.getRunningInstance(job)
223			lines.append(job.runningJob[attr])
224	spiga	1.1	return lines
225	spiga	1.3
226			def queryDistJob(self, attr):
227			'''
228			Returns the list of distinct value for a given job attributes
229			'''
230			distAttr=[]
231	spiga	1.5	task = common.bossSession.loadJobDist( 1, attr )
232	spiga	1.9	for i in task: distAttr.append(eval(i[attr]))
233	spiga	1.3	return distAttr
234
235	spiga	1.5	def queryDistJob_Attr(self, attr_1, attr_2, list):
236	spiga	1.4	'''
237	spiga	1.9	Returns the list of distinct value for a given job attribute
238	spiga	1.4	'''
239			distAttr=[]
240	slacapra	1.6	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
241	spiga	1.9	for i in task: distAttr.append(eval(i[attr_1]))
242	spiga	1.4	return distAttr
243	slacapra	1.6
244	spiga	1.3	def queryAttrJob(self, attr, field):
245			'''
246			Returns the list of jobs matching the given attribute
247			'''
248			matched=[]
249			task = common.bossSession.loadJobsByAttr(attr )
250			for i in task:
251			matched.append(i[field])
252			return matched
253
254	spiga	1.7
255			def queryAttrRunJob(self, attr,field):
256			'''
257			Returns the list of jobs matching the given attribute
258			'''
259			matched=[]
260			task = common.bossSession.loadJobsByRunningAttr(attr)
261			for i in task:
262			matched.append(i[field])
263			return matched