Source Code for Module entropy.services.repository.interfaces

  1  # -*- coding: utf-8 -*- 
  2  ''' 
  3      # DESCRIPTION: 
  4      # Entropy Object Oriented Interface 
  5   
  6      Copyright (C) 2007-2009 Fabio Erculiani 
  7   
  8      This program is free software; you can redistribute it and/or modify 
  9      it under the terms of the GNU General Public License as published by 
 10      the Free Software Foundation; either version 2 of the License, or 
 11      (at your option) any later version. 
 12   
 13      This program is distributed in the hope that it will be useful, 
 14      but WITHOUT ANY WARRANTY; without even the implied warranty of 
 15      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 16      GNU General Public License for more details. 
 17   
 18      You should have received a copy of the GNU General Public License 
 19      along with this program; if not, write to the Free Software 
 20      Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA 
 21  ''' 
 22   
 23  from __future__ import with_statement 
 24  import os 
 25  import shutil 
 26  from entropy.services.interfaces import SocketHost 
 27  from entropy.output import TextInterface, blue, brown, darkred, darkgreen 
 28  from entropy.const import etpConst, etpCache 
 29  from entropy.misc import TimeScheduled 
 30  from entropy.i18n import _ 
 31   
 32 -class Server(SocketHost): 
 33   
 34 -    class ServiceInterface(TextInterface): 
 35 -        def __init__(self, *args, **kwargs): 
 36              pass 
 37   
 38      import entropy.tools as entropyTools 
 39      import entropy.dump as dumpTools 
 40 -    def __init__(self, repositories, do_ssl = False, stdout_logging = True, **kwargs): 
 41   
 42          # instantiate critical constants 
 43          etpConst['socket_service']['max_connections'] = 5000 
 44   
 45          from entropy.services.repository.commands import Repository 
 46          self.RepositoryCommands = Repository 
 47          from entropy.db import dbapi2 
 48          self.dbapi2 = dbapi2 
 49          from entropy.client.interfaces import Client 
 50          self.Entropy = Client(noclientdb = 2) 
 51          self.do_ssl = do_ssl 
 52          self.LockScanner = None 
 53          self.syscache = { 
 54              'db': {}, 
 55              'db_trashed': set(), 
 56              'dbs_not_available': set(), 
 57          } 
 58   
 59          # setup System Settings 
 60          from entropy.core import SystemSettings 
 61          self.SystemSettings = SystemSettings() 
 62          self.SystemSettings['socket_service']['max_connections'] = 5000 
 63   
 64          etpConst['socketloglevel'] = 1 
 65          if not kwargs.has_key('external_cmd_classes'): 
 66              kwargs['external_cmd_classes'] = [] 
 67          kwargs['external_cmd_classes'].insert(0,self.RepositoryCommands) 
 68          SocketHost.__init__( 
 69              self, 
 70              self.ServiceInterface, 
 71              noclientdb = 2, 
 72              sock_output = self.Entropy, 
 73              ssl = do_ssl, 
 74              **kwargs 
 75          ) 
 76          self.stdout_logging = stdout_logging 
 77          self.repositories = repositories.copy() 
 78          self.expand_repositories() 
 79          # start timed lock file scanning 
 80          self.start_repository_lock_scanner() 
 81   
 82 -    def killall(self): 
 83          SocketHost.killall(self) 
 84          if self.LockScanner != None: 
 85              self.LockScanner.kill() 
 86   
 87 -    def start_repository_lock_scanner(self): 
 88          self.LockScanner = TimeScheduled(0.5, self.lock_scan) 
 89          self.LockScanner.start() 
 90   
 91 -    def set_repository_db_availability(self, repo_tuple): 
 92          self.repositories[repo_tuple]['enabled'] = False 
 93          mydbpath = os.path.join(self.repositories[repo_tuple]['dbpath'],etpConst['etpdatabasefile']) 
 94          if os.path.isfile(mydbpath) and os.access(mydbpath, os.W_OK): 
 95              self.syscache['dbs_not_available'].discard(repo_tuple) 
 96              self.repositories[repo_tuple]['enabled'] = True 
 97   
 98 -    def is_repository_available(self, repo_tuple): 
 99   
100          if repo_tuple not in self.repositories: 
101              return None 
102          # is repository being updated 
103          if self.repositories[repo_tuple]['locked']: 
104              return False 
105          # repository database does not exist 
106          if not self.repositories[repo_tuple]['enabled']: 
107              return False 
108   
109          return True 
110   
111 -    def lock_scan(self): 
112          do_clear = set() 
113          for repository, arch, product, branch in self.repositories: 
114              x = (repository,arch,product,branch,) 
115              self.set_repository_db_availability(x) 
116              if not self.repositories[x]['enabled']: 
117                  if x in self.syscache['dbs_not_available']: 
118                      continue 
119                  self.syscache['dbs_not_available'].add(x) 
120                  mytxt = blue("%s.") % (_("database does not exist. Locking services for it"),) 
121                  self.updateProgress( 
122                      "[%s] %s" % ( 
123                              brown(str(x)), 
124                              mytxt, 
125                      ), 
126                      importance = 1, 
127                      type = "info" 
128                  ) 
129                  do_clear.add(repository) 
130                  continue 
131              if os.path.isfile(self.repositories[x]['download_lock']) and \ 
132                  not self.repositories[x]['locked']: 
133                      self.repositories[x]['locked'] = True 
134                      mydbpath = os.path.join(self.repositories[x]['dbpath'],etpConst['etpdatabasefile']) 
135                      self.close_db(mydbpath) 
136                      self.eapi3_lock_repo(*x) 
137                      do_clear.add(repository) 
138                      mytxt = blue("%s.") % (_("database got locked. Locking services for it"),) 
139                      self.updateProgress( 
140                          "[%s] %s" % ( 
141                                  brown(str(x)), 
142                                  mytxt, 
143                          ), 
144                          importance = 1, 
145                          type = "info" 
146                      ) 
147              elif not os.path.isfile(self.repositories[x]['download_lock']) and \ 
148                  self.repositories[x]['locked']: 
149   
150                  # setup variables 
151                  dbpath = self.repositories[x]['dbpath'] 
152                  cmethod = self.repositories[x]['cmethod'] 
153                  cmethod_data = etpConst['etpdatabasecompressclasses'].get(cmethod) 
154                  unpack_method = cmethod_data[1] 
155                  compressed_dbfile = etpConst[cmethod_data[2]] 
156                  compressed_dbpath = os.path.join(dbpath, compressed_dbfile) 
157   
158                  if not os.access(compressed_dbpath, os.R_OK | os.F_OK | os.W_OK): 
159                      mytxt = darkred("%s: %s !!") % ( 
160                          _("cannot unlock database, compressed file not found"), 
161                          compressed_dbpath, 
162                      ) 
163                      self.updateProgress( 
164                          "[%s] %s" % ( 
165                                  brown(str(x)), 
166                                  mytxt, 
167                          ), 
168                          importance = 1, 
169                          type = "warning" 
170                      ) 
171                      self.syscache['dbs_not_available'].add(x) 
172                      do_clear.add(repository) 
173                      continue 
174   
175                  # make sure this db is closed 
176                  mydbpath = os.path.join(dbpath, etpConst['etpdatabasefile']) 
177                  self.close_db(mydbpath) 
178   
179                  mytxt = blue("%s: %s") % ( 
180                      _("unpacking compressed database"), 
181                      compressed_dbpath, 
182                  ) 
183                  self.updateProgress( 
184                      "[%s] %s" % ( 
185                              brown(str(x)), 
186                              mytxt, 
187                      ), 
188                      importance = 1, 
189                      type = "info" 
190                  ) 
191   
192                  # now unpack compressed db in place 
193                  unpack_func = getattr(self.entropyTools, unpack_method) 
194                  generated_outpath = unpack_func(compressed_dbpath) 
195                  if mydbpath != generated_outpath: 
196                      try: 
197                          os.rename(generated_outpath, mydbpath) 
198                      except OSError: 
199                          shutil.move(generated_outpath, mydbpath) 
200   
201                  mytxt = blue("%s. %s:") % ( 
202                      _("unlocking and indexing database"), 
203                      _("hash"), 
204                  ) 
205                  self.updateProgress( 
206                      "[%s] %s" % ( 
207                              brown(str(x)), 
208                              mytxt, 
209                      ), 
210                      importance = 1, 
211                      type = "info" 
212                  ) 
213                  # woohoo, got unlocked eventually 
214                  mydb = self.open_db(mydbpath, docache = False) 
215                  mydb.createAllIndexes() 
216                  self.updateProgress( 
217                      darkgreen(str(mydb.database_checksum(do_order = True, strict = False, strings = True))), 
218                      importance = 1, 
219                      type = "info" 
220                  ) 
221                  mydb.closeDB() 
222                  self.Entropy.clear_dump_cache(etpCache['repository_server']+"/"+repository+"/") 
223                  self.repositories[x]['locked'] = False 
224                  self.eapi3_unlock_repo(*x) 
225   
226          for repo in do_clear: 
227              self.Entropy.clear_dump_cache(etpCache['repository_server']+"/"+repo+"/") 
228   
229 -    def eapi3_lock_repo(self, repository, arch, product, branch): 
230          lock_file = os.path.join(self.repositories[(repository, arch, product, branch,)]['dbpath'],etpConst['etpdatabaseeapi3lockfile']) 
231          if not os.path.lexists(lock_file): 
232              f = open(lock_file,"w") 
233              f.write("this repository is EAPI3 locked") 
234              f.flush() 
235              f.close() 
236   
237 -    def eapi3_unlock_repo(self, repository, arch, product, branch): 
238          lock_file = os.path.join(self.repositories[(repository, arch, product, branch,)]['dbpath'],etpConst['etpdatabaseeapi3lockfile']) 
239          if os.path.isfile(lock_file): 
240              os.remove(lock_file) 
241   
242 -    def get_dcache(self, item, repo = '_norepo_'): 
243          return self.dumpTools.loadobj(etpCache['repository_server']+"/"+repo+"/"+str(hash(item))) 
244   
245 -    def set_dcache(self, item, data, repo = '_norepo_'): 
246          self.dumpTools.dumpobj(etpCache['repository_server']+"/"+repo+"/"+str(hash(item)),data) 
247   
248 -    def close_db(self, dbpath): 
249          try: 
250              dbc = self.syscache['db'].pop(dbpath) 
251              dbc.closeDB() 
252          except KeyError: 
253              pass 
254          except self.dbapi2.ProgrammingError: 
255              # they've been opened by the Commands Processor 
256              self.syscache['db_trashed'].add(dbc) 
257   
258 -    def open_db(self, dbpath, docache = True): 
259          if docache: 
260              cached = self.syscache['db'].get(dbpath) 
261              if cached != None: 
262                  return cached 
263          dbc = self.Entropy.open_generic_database( 
264              dbpath, 
265              xcache = False, 
266              readOnly = True, 
267              skipChecks = True 
268          ) 
269          if docache: 
270              self.syscache['db'][dbpath] = dbc 
271          return dbc 
272   
273 -    def expand_repositories(self): 
274   
275          for repository, arch, product, branch in self.repositories: 
276              x = (repository,arch,product,branch,) 
277              self.repositories[x]['locked'] = True # loading locked 
278              self.set_repository_db_availability(x) 
279              mydbpath = self.repositories[x]['dbpath'] 
280              myrevfile = os.path.join(mydbpath,etpConst['etpdatabaserevisionfile']) 
281              myrev = '0' 
282              if os.path.isfile(myrevfile): 
283                  while 1: 
284                      try: 
285                          f = open(myrevfile) 
286                          myrev = f.readline().strip() 
287                          f.close() 
288                      except IOError: # should never happen but who knows 
289                          continue 
290                      break 
291              self.repositories[x]['dbrevision'] = myrev 
292              self.repositories[x]['download_lock'] = os.path.join( 
293                  mydbpath, 
294                  etpConst['etpdatabasedownloadlockfile'] 
295              ) 
296              if not self.repositories[x].has_key('cmethod'): 
297                  raise AttributeError("cmethod not specified for: %s" % (x,)) 
298              if self.repositories[x]['cmethod'] not in etpConst['etpdatabasesupportedcformats']: 
299                  raise AttributeError("wrong cmethod for: %s" % (x,)) 
300