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)
        
        return
 
    def getTask(self): 

        self.task = common.bossSession.loadTaskByID(1)
        return self.task


    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['jobType']=optsToSave['jobtype']  
        opt[ 'name']=common.work_space.taskName()  
        task = Task( opt )
      
        common.bossSession.saveTask( task )
        #common.bossSession.updateDB( 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())
        jobs = common.bossSession.loadJob(task['id'],nj+1)
        for key in optsToSave.keys():
            jobs[key] = optsToSave[key]
            common.bossSession.updateDB( jobs )
        return 

    def updateRunJob_(self, nj, optsToSave):       
        """
        Update Running Job fields   
        """
        task = common.bossSession.loadTaskByID(1)
        #task = common.bossSession.loadTaskByName( common.work_space.taskName())
        common.bossSession.getRunningInstance(task.jobs[nj])
        for key in optsToSave.keys():
            task.jobs[nj].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 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])
            lines.append(task.jobs[i].runningJob[attr])
        return lines

    def queryDistJob(self, attr):
        '''
        Returns the list of distinct value for a given job attributes 
        '''
        distAttr=[]
        task = common.bossSession.loadJobDistAttr( 1, attr ) 
        for i in task: distAttr.append(i[attr])   
        return  distAttr

    def queryDistJobHadr(self, attr_1, attr_2, list):
        '''
        Returns the list of distinct value for a given job attributes 
        '''
        distAttr=[]
        task = common.bossSession.loadJobDistAttrHadr( 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

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