Package entropy :: Package services :: Package repository :: Module interfaces

Source Code for Module entropy.services.repository.interfaces

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