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 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 = 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.11
Committed:	Wed Mar 26 14:05:52 2008 UTC (17 years, 1 month ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.10:	+7 -6 lines
Log Message:	restored -printId functionality using bosslite
#	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	self.task = common.bossSession.load(1)[0]
42	return
43
44	def getTask(self, jobsList='all'): #, cfg_params):
45	"""
46	Return task with all/list of jobs
47	"""
48
49	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
57	def getJob(self, n):
58	"""
59	Return a task with a single job
60	"""
61	task = common.bossSession.load(1,str(n))[0]
62	return task
63
64
65	def createTask_(self, optsToSave):
66	"""
67	Task declaration
68	with the first coniguration stuff
69	"""
70	opt={}
71	opt['serverName']=optsToSave['server_name']
72	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	task = common.bossSession.load(1)[0]
83
84	for key in optsToSave.keys():
85	task[key] = optsToSave[key]
86	common.bossSession.updateDB( task )
87	return
88
89	def createJobs_(self, jobsL):
90	"""
91	Fill crab DB with the jobs filed
92	"""
93	task = common.bossSession.loadTask(1)
94	jobs = []
95	for id in jobsL:
96	parameters = {}
97	parameters['jobId'] = str(id)
98	parameters['name'] = 'job' + str(id)
99	job = Job(parameters)
100	jobs.append(job)
101	task.addJobs(jobs)
102	common.bossSession.updateDB( task )
103	return
104
105	def updateJob_(self, jobsL, optsToSave):
106	"""
107	Update Job fields
108	"""
109	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	for key in optsToSave[id].keys():
115	job[key] = optsToSave[id][key]
116	id+=1
117	common.bossSession.updateDB( task )
118	return
119
120	def updateRunJob_(self, jobsL, optsToSave):
121	"""
122	Update Running Job fields
123	"""
124	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	for key in optsToSave.keys():
130	job.runningJob[key] = optsToSave[key]
131	common.bossSession.updateDB( task )
132	return
133
134	def nJobs(self,list=''):
135
136	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
144	def dump(self,jobs):
145	"""
146	List a complete set of infos for a job/range of jobs
147	"""
148	task = common.bossSession.load(1)[0]
149
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
169	def serializeTask(self, tmp_task = None):
170	if tmp_task is None:
171	#tmp_task = common.bossSession.loadTaskByID(1)
172	tmp_task = common.bossSession.load(1)[0]
173	return common.bossSession.serialize(tmp_task)
174
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	task = common.bossSession.load(1)[0]
183	if server_mode == 1:
184	header= "Task Id = %-40s " %(task['name'])
185	else:
186	for job in task.jobs:
187	toPrint=''
188	common.bossSession.getRunningInstance(job)
189	toPrint = "%-5s %-50s " % (job['id'],job.runningJob['schedulerId'])
190	lines.append(toPrint)
191	header+= "%-5s %-50s " % ('Job:','ID' )
192	displayReport(self,header,lines)
193	return
194
195	def queryTask(self,attr):
196	'''
197	Perform a query over a generic task attribute
198	'''
199	task = common.bossSession.loadTask(1)
200	return task[attr]
201
202	def queryJob(self, attr, jobs):
203	'''
204	Perform a query for a range/all/single job
205	over a generic job attribute
206	'''
207	lines=[]
208	str_jobs=string.join(map(str,jobs),",")
209	task = common.bossSession.load(1,str_jobs)[0]
210	for job in task.jobs:
211	lines.append(eval(job[attr]))
212	return lines
213
214	def queryRunJob(self, attr, jobs):
215	'''
216	Perform a query for a range/all/single job
217	over a generic job attribute
218	'''
219	lines=[]
220	str_jobs=string.join(map(str,jobs),",")
221	task = common.bossSession.load(1,str_jobs)[0]
222	for job in task.jobs:
223	common.bossSession.getRunningInstance(job)
224	lines.append(job.runningJob[attr])
225	return lines
226
227	def queryDistJob(self, attr):
228	'''
229	Returns the list of distinct value for a given job attributes
230	'''
231	distAttr=[]
232	task = common.bossSession.loadJobDist( 1, attr )
233	for i in task: distAttr.append(eval(i[attr]))
234	return distAttr
235
236	def queryDistJob_Attr(self, attr_1, attr_2, list):
237	'''
238	Returns the list of distinct value for a given job attribute
239	'''
240	distAttr=[]
241	task = common.bossSession.loadJobDistAttr( 1, attr_1, attr_2, list )
242	for i in task: distAttr.append(eval(i[attr_1]))
243	return distAttr
244
245	def queryAttrJob(self, attr, field):
246	'''
247	Returns the list of jobs matching the given attribute
248	'''
249	matched=[]
250	task = common.bossSession.loadJobsByAttr(attr )
251	for i in task:
252	matched.append(i[field])
253	return matched
254
255
256	def queryAttrRunJob(self, attr,field):
257	'''
258	Returns the list of jobs matching the given attribute
259	'''
260	matched=[]
261	task = common.bossSession.loadJobsByRunningAttr(attr)
262	for i in task:
263	matched.append(i[field])
264	return matched