X-Git-Url: https://git.decadent.org.uk/gitweb/?a=blobdiff_plain;f=daklib%2Fdbconn.py;h=64437561c67a67e31646bc2b96d20097149af727;hb=e81d92555e88dbf275b465a306e8461e9f3991aa;hp=47e2b130645fd582df973432bf33cb4c6d7bb04d;hpb=4bd036e1e2c3663991a9685a8d09ed58f293c140;p=dak.git diff --git a/daklib/dbconn.py b/daklib/dbconn.py index 47e2b130..64437561 100755 --- a/daklib/dbconn.py +++ b/daklib/dbconn.py @@ -33,12 +33,14 @@ ################################################################################ +import apt_pkg import os from os.path import normpath import re import psycopg2 import traceback import commands +import signal try: # python >= 2.6 @@ -72,7 +74,7 @@ from sqlalchemy.orm.exc import NoResultFound # in the database from config import Config from textutils import fix_maintainer -from dak_exceptions import DBUpdateError, NoSourceFieldError +from dak_exceptions import DBUpdateError, NoSourceFieldError, FileExistsError # suppress some deprecation warnings in squeeze related to sqlalchemy import warnings @@ -487,6 +489,11 @@ __all__.append('BinContents') ################################################################################ +def subprocess_setup(): + # Python installs a SIGPIPE handler by default. This is usually not what + # non-Python subprocesses expect. + signal.signal(signal.SIGPIPE, signal.SIG_DFL) + class DBBinary(ORMObject): def __init__(self, package = None, source = None, version = None, \ maintainer = None, architecture = None, poolfile = None, \ @@ -525,7 +532,8 @@ class DBBinary(ORMObject): package does not contain any regular file. ''' fullpath = self.poolfile.fullpath - dpkg = Popen(['dpkg-deb', '--fsys-tarfile', fullpath], stdout = PIPE) + dpkg = Popen(['dpkg-deb', '--fsys-tarfile', fullpath], stdout = PIPE, + preexec_fn = subprocess_setup) tar = TarFile.open(fileobj = dpkg.stdout, mode = 'r|') for member in tar.getmembers(): if not member.isdir(): @@ -703,6 +711,7 @@ class BuildQueue(object): try: # Grab files we want to include newer = session.query(BuildQueueFile).filter_by(build_queue_id = self.queue_id).filter(BuildQueueFile.lastused + timedelta(seconds=self.stay_of_execution) > starttime).all() + newer += session.query(BuildQueuePolicyFile).filter_by(build_queue_id = self.queue_id).filter(BuildQueuePolicyFile.lastused + timedelta(seconds=self.stay_of_execution) > starttime).all() # Write file list with newer files (fl_fd, fl_name) = mkstemp() for n in newer: @@ -795,6 +804,7 @@ class BuildQueue(object): # Grab files older than our execution time older = session.query(BuildQueueFile).filter_by(build_queue_id = self.queue_id).filter(BuildQueueFile.lastused + timedelta(seconds=self.stay_of_execution) <= starttime).all() + older += session.query(BuildQueuePolicyFile).filter_by(build_queue_id = self.queue_id).filter(BuildQueuePolicyFile.lastused + timedelta(seconds=self.stay_of_execution) <= starttime).all() for o in older: killdb = False @@ -822,9 +832,7 @@ class BuildQueue(object): if f.startswith('Packages') or f.startswith('Source') or f.startswith('Release') or f.startswith('advisory'): continue - try: - r = session.query(BuildQueueFile).filter_by(build_queue_id = self.queue_id).filter_by(filename = f).one() - except NoResultFound: + if not self.contains_filename(f): fp = os.path.join(self.path, f) if dryrun: Logger.log(["I: Would remove unused link %s" % fp]) @@ -835,6 +843,18 @@ class BuildQueue(object): except OSError: Logger.log(["E: Failed to unlink unreferenced file %s" % r.fullpath]) + def contains_filename(self, filename): + """ + @rtype Boolean + @returns True if filename is supposed to be in the queue; False otherwise + """ + session = DBConn().session().object_session(self) + if session.query(BuildQueueFile).filter_by(build_queue_id = self.queue_id, filename = filename).count() > 0: + return True + elif session.query(BuildQueuePolicyFile).filter_by(build_queue = self, filename = filename).count() > 0: + return True + return False + def add_file_from_pool(self, poolfile): """Copies a file into the pool. Assumes that the PoolFile object is attached to the same SQLAlchemy session as the Queue object is. @@ -844,8 +864,8 @@ class BuildQueue(object): # Check if we have a file of this name or this ID already for f in self.queuefiles: - if f.fileid is not None and f.fileid == poolfile.file_id or \ - f.poolfile.filename == poolfile_basename: + if (f.fileid is not None and f.fileid == poolfile.file_id) or \ + (f.poolfile is not None and f.poolfile.filename == poolfile_basename): # In this case, update the BuildQueueFile entry so we # don't remove it too early f.lastused = datetime.now() @@ -871,6 +891,9 @@ class BuildQueue(object): else: os.symlink(targetpath, queuepath) qf.fileid = poolfile.file_id + except FileExistsError: + if not poolfile.identical_to(queuepath): + raise except OSError: return None @@ -879,6 +902,64 @@ class BuildQueue(object): return qf + def add_changes_from_policy_queue(self, policyqueue, changes): + """ + Copies a changes from a policy queue together with its poolfiles. + + @type policyqueue: PolicyQueue + @param policyqueue: policy queue to copy the changes from + + @type changes: DBChange + @param changes: changes to copy to this build queue + """ + for policyqueuefile in changes.files: + self.add_file_from_policy_queue(policyqueue, policyqueuefile) + for poolfile in changes.poolfiles: + self.add_file_from_pool(poolfile) + + def add_file_from_policy_queue(self, policyqueue, policyqueuefile): + """ + Copies a file from a policy queue. + Assumes that the policyqueuefile is attached to the same SQLAlchemy + session as the Queue object is. The caller is responsible for + committing after calling this function. + + @type policyqueue: PolicyQueue + @param policyqueue: policy queue to copy the file from + + @type policyqueuefile: ChangePendingFile + @param policyqueuefile: file to be added to the build queue + """ + session = DBConn().session().object_session(policyqueuefile) + + # Is the file already there? + try: + f = session.query(BuildQueuePolicyFile).filter_by(build_queue=self, file=policyqueuefile).one() + f.lastused = datetime.now() + return f + except NoResultFound: + pass # continue below + + # We have to add the file. + f = BuildQueuePolicyFile() + f.build_queue = self + f.file = policyqueuefile + f.filename = policyqueuefile.filename + + source = os.path.join(policyqueue.path, policyqueuefile.filename) + target = f.fullpath + try: + # Always copy files from policy queues as they might move around. + import utils + utils.copy(source, target) + except FileExistsError: + if not policyqueuefile.identical_to(target): + raise + except OSError: + return None + + session.add(f) + return f __all__.append('BuildQueue') @@ -911,6 +992,10 @@ __all__.append('get_build_queue') ################################################################################ class BuildQueueFile(object): + """ + BuildQueueFile represents a file in a build queue coming from a pool. + """ + def __init__(self, *args, **kwargs): pass @@ -926,6 +1011,27 @@ __all__.append('BuildQueueFile') ################################################################################ +class BuildQueuePolicyFile(object): + """ + BuildQueuePolicyFile represents a file in a build queue that comes from a + policy queue (and not a pool). + """ + + def __init__(self, *args, **kwargs): + pass + + #@property + #def filename(self): + # return self.file.filename + + @property + def fullpath(self): + return os.path.join(self.build_queue.path, self.filename) + +__all__.append('BuildQueuePolicyFile') + +################################################################################ + class ChangePendingBinary(object): def __init__(self, *args, **kwargs): pass @@ -944,6 +1050,24 @@ class ChangePendingFile(object): def __repr__(self): return '' % self.change_pending_file_id + def identical_to(self, filename): + """ + compare size and hash with the given file + + @rtype: bool + @return: true if the given file has the same size and hash as this object; false otherwise + """ + st = os.stat(filename) + if self.size != st.st_size: + return False + + f = open(filename, "r") + sha256sum = apt_pkg.sha256sum(f) + if sha256sum != self.sha256sum: + return False + + return True + __all__.append('ChangePendingFile') ################################################################################ @@ -1260,6 +1384,17 @@ __all__.append('get_dscfiles') ################################################################################ +class ExternalOverride(ORMObject): + def __init__(self, *args, **kwargs): + pass + + def __repr__(self): + return '' % (self.package, self.key, self.value) + +__all__.append('ExternalOverride') + +################################################################################ + class PoolFile(ORMObject): def __init__(self, filename = None, location = None, filesize = -1, \ md5sum = None): @@ -1282,6 +1417,24 @@ class PoolFile(ORMObject): def not_null_constraints(self): return ['filename', 'md5sum', 'location'] + def identical_to(self, filename): + """ + compare size and hash with the given file + + @rtype: bool + @return: true if the given file has the same size and hash as this object; false otherwise + """ + st = os.stat(filename) + if self.filesize != st.st_size: + return False + + f = open(filename, "r") + sha256sum = apt_pkg.sha256sum(f) + if sha256sum != self.sha256sum: + return False + + return True + __all__.append('PoolFile') @session_wrapper @@ -2354,20 +2507,13 @@ def source_exists(source, source_version, suites = ["any"], session=None): q = session.query(DBSource).filter_by(source=source). \ filter(DBSource.version.in_([source_version, orig_source_version])) if suite != "any": - # source must exist in suite X, or in some other suite that's - # mapped to X, recursively... silent-maps are counted too, - # unreleased-maps aren't. - maps = cnf.ValueList("SuiteMappings")[:] - maps.reverse() - maps = [ m.split() for m in maps ] - maps = [ (x[1], x[2]) for x in maps - if x[0] == "map" or x[0] == "silent-map" ] - s = [suite] - for (from_, to) in maps: - if from_ in s and to not in s: - s.append(to) - - q = q.filter(DBSource.suites.any(Suite.suite_name.in_(s))) + # source must exist in 'suite' or a suite that is enhanced by 'suite' + s = get_suite(suite, session) + enhances_vcs = session.query(VersionCheck).filter(VersionCheck.suite==s).filter_by(check='Enhances') + considered_suites = [ vc.reference for vc in enhances_vcs ] + considered_suites.append(s) + + q = q.filter(DBSource.suites.any(Suite.suite_id.in_([s.suite_id for s in considered_suites]))) if q.count() > 0: continue @@ -2489,6 +2635,15 @@ __all__.append('import_metadata_into_db') ################################################################################ +def split_uploaders(uploaders_list): + ''' + Split the Uploaders field into the individual uploaders and yield each of + them. Beware: email addresses might contain commas. + ''' + import re + for uploader in re.sub(">[ ]*,", ">\t", uploaders_list).split("\t"): + yield uploader.strip() + @session_wrapper def add_dsc_to_db(u, filename, session=None): entry = u.pkg.files[filename] @@ -2498,7 +2653,11 @@ def add_dsc_to_db(u, filename, session=None): source.source = u.pkg.dsc["source"] source.version = u.pkg.dsc["version"] # NB: not files[file]["version"], that has no epoch source.maintainer_id = get_or_set_maintainer(u.pkg.dsc["maintainer"], session).maintainer_id - source.changedby_id = get_or_set_maintainer(u.pkg.changes["changed-by"], session).maintainer_id + # If Changed-By isn't available, fall back to maintainer + if u.pkg.changes.has_key("changed-by"): + source.changedby_id = get_or_set_maintainer(u.pkg.changes["changed-by"], session).maintainer_id + else: + source.changedby_id = get_or_set_maintainer(u.pkg.dsc["maintainer"], session).maintainer_id source.fingerprint_id = get_or_set_fingerprint(u.pkg.changes["fingerprint"], session).fingerprint_id source.install_date = datetime.now().date() @@ -2571,10 +2730,11 @@ def add_dsc_to_db(u, filename, session=None): session.add(df) # Add the src_uploaders to the DB + session.flush() + session.refresh(source) source.uploaders = [source.maintainer] if u.pkg.dsc.has_key("uploaders"): - for up in u.pkg.dsc["uploaders"].replace(">, ", ">\t").split("\t"): - up = up.strip() + for up in split_uploaders(u.pkg.dsc["uploaders"]): source.uploaders.append(get_or_set_maintainer(up, session)) session.flush() @@ -2769,6 +2929,12 @@ class Suite(ORMObject): return session.query(DBSource).filter_by(source = source). \ with_parent(self) + def get_overridesuite(self): + if self.overridesuite is None: + return self + else: + return object_session(self).query(Suite).filter_by(suite_name=self.overridesuite).one() + __all__.append('Suite') @session_wrapper @@ -3027,6 +3193,35 @@ __all__.append('SourceMetadata') ################################################################################ +class VersionCheck(ORMObject): + def __init__(self, *args, **kwargs): + pass + + def properties(self): + #return ['suite_id', 'check', 'reference_id'] + return ['check'] + + def not_null_constraints(self): + return ['suite', 'check', 'reference'] + +__all__.append('VersionCheck') + +@session_wrapper +def get_version_checks(suite_name, check = None, session = None): + suite = get_suite(suite_name, session) + if not suite: + # Make sure that what we return is iterable so that list comprehensions + # involving this don't cause a traceback + return [] + q = session.query(VersionCheck).filter_by(suite=suite) + if check: + q = q.filter_by(check=check) + return q.all() + +__all__.append('get_version_checks') + +################################################################################ + class DBConn(object): """ database module init. @@ -3053,6 +3248,7 @@ class DBConn(object): 'binary_acl_map', 'build_queue', 'build_queue_files', + 'build_queue_policy_files', 'changelogs_text', 'changes', 'component', @@ -3064,6 +3260,7 @@ class DBConn(object): 'changes_pending_source_files', 'changes_pool_files', 'dsc_files', + 'external_overrides', 'extra_src_references', 'files', 'fingerprint', @@ -3092,13 +3289,13 @@ class DBConn(object): 'suite_src_formats', 'uid', 'upload_blocks', + 'version_check', ) views = ( 'almost_obsolete_all_associations', 'almost_obsolete_src_associations', 'any_associations_source', - 'bin_assoc_by_arch', 'bin_associations_binaries', 'binaries_suite_arch', 'binfiles_suite_component_arch', @@ -3146,6 +3343,11 @@ class DBConn(object): properties = dict(buildqueue = relation(BuildQueue, backref='queuefiles'), poolfile = relation(PoolFile, backref='buildqueueinstances'))) + mapper(BuildQueuePolicyFile, self.tbl_build_queue_policy_files, + properties = dict( + build_queue = relation(BuildQueue, backref='policy_queue_files'), + file = relation(ChangePendingFile, lazy='joined'))) + mapper(DBBinary, self.tbl_binaries, properties = dict(binary_id = self.tbl_binaries.c.id, package = self.tbl_binaries.c.package, @@ -3193,6 +3395,13 @@ class DBConn(object): poolfile_id = self.tbl_dsc_files.c.file, poolfile = relation(PoolFile))) + mapper(ExternalOverride, self.tbl_external_overrides, + properties = dict( + suite_id = self.tbl_external_overrides.c.suite, + suite = relation(Suite), + component_id = self.tbl_external_overrides.c.component, + component = relation(Component))) + mapper(PoolFile, self.tbl_files, properties = dict(file_id = self.tbl_files.c.id, filesize = self.tbl_files.c.size, @@ -3406,6 +3615,13 @@ class DBConn(object): key = relation(MetadataKey), value = self.tbl_source_metadata.c.value)) + mapper(VersionCheck, self.tbl_version_check, + properties = dict( + suite_id = self.tbl_version_check.c.suite, + suite = relation(Suite, primaryjoin=self.tbl_version_check.c.suite==self.tbl_suite.c.id), + reference_id = self.tbl_version_check.c.reference, + reference = relation(Suite, primaryjoin=self.tbl_version_check.c.reference==self.tbl_suite.c.id, lazy='joined'))) + ## Connection functions def __createconn(self): from config import Config @@ -3458,12 +3674,21 @@ class DBConn(object): self.__setupmappers() self.pid = os.getpid() - def session(self): + def session(self, work_mem = 0): + ''' + Returns a new session object. If a work_mem parameter is provided a new + transaction is started and the work_mem parameter is set for this + transaction. The work_mem parameter is measured in MB. A default value + will be used if the parameter is not set. + ''' # reinitialize DBConn in new processes if self.pid != os.getpid(): clear_mappers() self.__createconn() - return self.db_smaker() + session = self.db_smaker() + if work_mem > 0: + session.execute("SET LOCAL work_mem TO '%d MB'" % work_mem) + return session __all__.append('DBConn')