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.loadTaskByID(1)
        return
 
    def getTask(self, jobsList='all'): #, cfg_params):

        #if jobsList == 'all':
        #    self.task = common.bossSession.loadTaskByID(1)
        #else: 
        #self.task = common.bossSession.load('1','5')  
        #return self.task[0]
        self.task = common.bossSession.loadTaskByID(1)
        return self.task

    def getJob(self, n): 

        self.job = common.bossSession.loadJobByID(1,n)
        return self.job


    def createTask_(self, optsToSave):       
        """
        Task declaration
        with the first coniguration stuff 
        {'server_name': 'crabas.lnl.infn.it/data1/cms/', '-scheduler': 'glite', '-jobtype': 'cmssw', '-server_mode': '0'}

        """
        opt={}
        if optsToSave['server_mode'] == 1: 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.loadTaskByName(common.work_space.taskName() )
        task = common.bossSession.loadTaskByID(1)
         
        for key in optsToSave.keys():
            task[key] = optsToSave[key]
            common.bossSession.updateDB( task )
        return 

    def createJobs_(self, nj):
        """  
        Fill crab DB with  the jobs filed 
        """
        #task = common.bossSession.loadTaskByName(common.work_space.taskName())
        task = common.bossSession.loadTaskByID(1)
        jobs = [] 
        for id in range(nj):
            parameters = {}
            parameters['name'] = 'job' + str(id)
            job = Job(parameters)
            jobs.append(job)    
        task.addJobs(jobs)
        common.bossSession.updateDB( task )
        return

    def updateJob_(self, nj, optsToSave):       
        """
        Update Job fields   
        """
        task = common.bossSession.loadTaskByID(1)
        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
        for i in range(len(nj)):
           # jobs = common.bossSession.loadJob(task['id'],i)
            for key in optsToSave[i].keys():
                task.jobs[i][key] = optsToSave[i][key]
        common.bossSession.updateDB( task )
        return 

    def updateRunJob_(self, nj, optsToSave):       
        """
        Update Running Job fields   
        """
        task = common.bossSession.loadTaskByID(1)
        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
        for i in nj:
            common.bossSession.getRunningInstance(task.jobs[i])
            for key in optsToSave.keys():
                task.jobs[i].runningJob[key] = optsToSave[key]
        common.bossSession.updateDB( task )
        return 

    def nJobs(self):
        
        task = common.bossSession.loadTaskByID(1)
        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
        return len(task.jobs) 

    def dump(self,jobs):
        """
         List a complete set of infos for a job/range of jobs   
        """
        task = common.bossSession.loadTaskByID(1)
        #task = common.bossSession.loadTaskByName( common.work_space.taskName())

        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)
        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.loadTaskByID(1)
        if server_mode == 1:
            header= "Task Id = %-40s " %(task['name'])
        else:
         #   task = common.bossSession.loadTaskByName(common.work_space.taskName() )
            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.loadTaskByID(1)
        return task[attr]

    def queryJob(self, attr, njobs):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        task = common.bossSession.loadTaskByID(1)
        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
        for i in njobs:
            jobs = common.bossSession.loadJob(task['id'],i+1)
            lines.append(task.jobs[i][attr])
        return lines

    def queryRunJob(self, attr, jobs):
        '''
        Perform a query for a range/all/single job 
        over a generic job attribute 
        '''
        lines=[]
        task = common.bossSession.loadTaskByID(1)
       # task = common.bossSession.loadTaskByName( common.work_space.taskName() )
        for i in jobs:
            common.bossSession.getRunningInstance(task.jobs[i-1])
            lines.append(task.jobs[i-1].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(i[attr])   
        return  distAttr

    def queryDistJob_Attr(self, attr_1, attr_2, list):
        '''
        Returns the list of distinct value for a given job attributes 
        '''
        distAttr=[]
        task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list ) 
        for i in task: distAttr.append(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.8
Committed:	Mon Mar 17 17:42:53 2008 UTC (17 years, 1 month ago) by farinafa
Content type:	text/x-python
Branch:	MAIN
Changes since 1.7:	+5 -0 lines
Log Message:	A small fix on the usage of the DBinterface facade pattern
#	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.7	self.task = common.bossSession.loadTaskByID(1)
42	spiga	1.2	return
43	spiga	1.4
44	spiga	1.7	def getTask(self, jobsList='all'): #, cfg_params):
45	spiga	1.4
46	spiga	1.7	#if jobsList == 'all':
47			# self.task = common.bossSession.loadTaskByID(1)
48			#else:
49			#self.task = common.bossSession.load('1','5')
50			#return self.task[0]
51	spiga	1.4	self.task = common.bossSession.loadTaskByID(1)
52			return self.task
53	spiga	1.2
54	slacapra	1.6	def getJob(self, n):
55
56			self.job = common.bossSession.loadJobByID(1,n)
57			return self.job
58	spiga	1.2
59	spiga	1.1
60			def createTask_(self, optsToSave):
61			"""
62			Task declaration
63			with the first coniguration stuff
64	slacapra	1.6	{'server_name': 'crabas.lnl.infn.it/data1/cms/', '-scheduler': 'glite', '-jobtype': 'cmssw', '-server_mode': '0'}
65	spiga	1.1
66			"""
67			opt={}
68	spiga	1.7	if optsToSave['server_mode'] == 1: opt['serverName']=optsToSave['server_name']
69	spiga	1.1	opt[ 'name']=common.work_space.taskName()
70			task = Task( opt )
71
72			common.bossSession.saveTask( task )
73			return
74
75			def updateTask_(self,optsToSave):
76			"""
77			Update task fields
78			"""
79	spiga	1.2	#task = common.bossSession.loadTaskByName(common.work_space.taskName() )
80			task = common.bossSession.loadTaskByID(1)
81	spiga	1.1
82			for key in optsToSave.keys():
83			task[key] = optsToSave[key]
84			common.bossSession.updateDB( task )
85			return
86
87			def createJobs_(self, nj):
88			"""
89			Fill crab DB with the jobs filed
90			"""
91	spiga	1.2	#task = common.bossSession.loadTaskByName(common.work_space.taskName())
92			task = common.bossSession.loadTaskByID(1)
93	spiga	1.1	jobs = []
94			for id in range(nj):
95			parameters = {}
96	spiga	1.7	parameters['name'] = 'job' + str(id)
97	spiga	1.1	job = Job(parameters)
98			jobs.append(job)
99			task.addJobs(jobs)
100			common.bossSession.updateDB( task )
101			return
102
103			def updateJob_(self, nj, optsToSave):
104			"""
105			Update Job fields
106			"""
107	spiga	1.2	task = common.bossSession.loadTaskByID(1)
108			#task = common.bossSession.loadTaskByName( common.work_space.taskName())
109	spiga	1.7	for i in range(len(nj)):
110			# jobs = common.bossSession.loadJob(task['id'],i)
111			for key in optsToSave[i].keys():
112			task.jobs[i][key] = optsToSave[i][key]
113			common.bossSession.updateDB( task )
114	spiga	1.1	return
115
116			def updateRunJob_(self, nj, optsToSave):
117			"""
118			Update Running Job fields
119			"""
120	spiga	1.2	task = common.bossSession.loadTaskByID(1)
121			#task = common.bossSession.loadTaskByName( common.work_space.taskName())
122	spiga	1.7	for i in nj:
123			common.bossSession.getRunningInstance(task.jobs[i])
124			for key in optsToSave.keys():
125			task.jobs[i].runningJob[key] = optsToSave[key]
126	spiga	1.1	common.bossSession.updateDB( task )
127			return
128
129			def nJobs(self):
130	spiga	1.2
131			task = common.bossSession.loadTaskByID(1)
132			#task = common.bossSession.loadTaskByName( common.work_space.taskName())
133	spiga	1.1	return len(task.jobs)
134
135			def dump(self,jobs):
136			"""
137			List a complete set of infos for a job/range of jobs
138			"""
139	spiga	1.2	task = common.bossSession.loadTaskByID(1)
140			#task = common.bossSession.loadTaskByName( common.work_space.taskName())
141	spiga	1.1
142			njobs = len(jobs)
143			lines=[]
144			header=''
145			# ##query the DB asking the right infos for runningJobs TODO DS
146			# for job in jobs:
147			# ## here the query over runngJobs
148			# pass
149
150
151			# ##Define Header to show and Pass the query results,
152			# ## header and format to displayReport() TODO DS
153			# if njobs == 1: plural = ''
154			# else: plural = 's'
155			# header += 'Listing %d job%s:\n' % (njobs, plural)
156			# header += ' :\n' % (---) ## TODO DS
157
158			# displayReport(header, lines):
159			return
160	farinafa	1.8
161			def serializeTask(self, tmp_task = None):
162			if tmp_task is None:
163			tmp_task = common.bossSession.loadTaskByID(1)
164			return common.bossSession.serialize(tmp_task)
165	spiga	1.1
166			def queryID(self,server_mode=0):
167			'''
168			Return the taskId if serevr_mode =1
169			Return the joblistId if serevr_mode =0
170			'''
171			header=''
172			lines=[]
173	spiga	1.2	task = common.bossSession.loadTaskByID(1)
174	spiga	1.1	if server_mode == 1:
175	spiga	1.2	header= "Task Id = %-40s " %(task['name'])
176	spiga	1.1	else:
177	spiga	1.2	# task = common.bossSession.loadTaskByName(common.work_space.taskName() )
178	spiga	1.1	for i in range(len(task.job)):
179			common.bossSession.getRunningInstance(task.jobs[i])
180			lines.append(task.jobs[i].runningJob['schedulerId'])
181
182			header+= "Job: %-5s Id = %-40s: \n"
183			displayReport(header,lines)
184			return
185
186			def queryTask(self,attr):
187			'''
188			Perform a query over a generic task attribute
189			'''
190	spiga	1.2	task = common.bossSession.loadTaskByID(1)
191	spiga	1.1	return task[attr]
192
193			def queryJob(self, attr, njobs):
194			'''
195			Perform a query for a range/all/single job
196			over a generic job attribute
197			'''
198			lines=[]
199	spiga	1.2	task = common.bossSession.loadTaskByID(1)
200			#task = common.bossSession.loadTaskByName( common.work_space.taskName())
201	spiga	1.1	for i in njobs:
202			jobs = common.bossSession.loadJob(task['id'],i+1)
203			lines.append(task.jobs[i][attr])
204			return lines
205
206			def queryRunJob(self, attr, jobs):
207			'''
208			Perform a query for a range/all/single job
209			over a generic job attribute
210			'''
211			lines=[]
212	spiga	1.2	task = common.bossSession.loadTaskByID(1)
213			# task = common.bossSession.loadTaskByName( common.work_space.taskName() )
214	spiga	1.1	for i in jobs:
215	spiga	1.5	common.bossSession.getRunningInstance(task.jobs[i-1])
216			lines.append(task.jobs[i-1].runningJob[attr])
217	spiga	1.1	return lines
218	spiga	1.3
219			def queryDistJob(self, attr):
220			'''
221			Returns the list of distinct value for a given job attributes
222			'''
223			distAttr=[]
224	spiga	1.5	task = common.bossSession.loadJobDist( 1, attr )
225	spiga	1.3	for i in task: distAttr.append(i[attr])
226			return distAttr
227
228	spiga	1.5	def queryDistJob_Attr(self, attr_1, attr_2, list):
229	spiga	1.4	'''
230			Returns the list of distinct value for a given job attributes
231			'''
232			distAttr=[]
233	slacapra	1.6	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
234	spiga	1.4	for i in task: distAttr.append(i[attr_1])
235			return distAttr
236	slacapra	1.6
237	spiga	1.3	def queryAttrJob(self, attr, field):
238			'''
239			Returns the list of jobs matching the given attribute
240			'''
241			matched=[]
242			task = common.bossSession.loadJobsByAttr(attr )
243			for i in task:
244			matched.append(i[field])
245			return matched
246
247	spiga	1.7
248			def queryAttrRunJob(self, attr,field):
249			'''
250			Returns the list of jobs matching the given attribute
251			'''
252			matched=[]
253			task = common.bossSession.loadJobsByRunningAttr(attr)
254			for i in task:
255			matched.append(i[field])
256			return matched