daklib/contents.py

   1 #!/usr/bin/env python
   2 """
   3 Helper code for contents generation.
   4
   5 @contact: Debian FTPMaster <ftpmaster@debian.org>
   6 @copyright: 2011 Torsten Werner <twerner@debian.org>
   7 @license: GNU General Public License version 2 or later
   8 """
   9
  10 ################################################################################
  11
  12 # This program is free software; you can redistribute it and/or modify
  13 # it under the terms of the GNU General Public License as published by
  14 # the Free Software Foundation; either version 2 of the License, or
  15 # (at your option) any later version.
  16
  17 # This program is distributed in the hope that it will be useful,
  18 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  19 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  20 # GNU General Public License for more details.
  21
  22 # You should have received a copy of the GNU General Public License
  23 # along with this program; if not, write to the Free Software
  24 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  25
  26 ################################################################################
  27
  28 from daklib.dbconn import *
  29 from daklib.config import Config
  30 from daklib.threadpool import ThreadPool
  31
  32 from sqlalchemy import desc, or_
  33 from subprocess import Popen, PIPE
  34
  35 class ContentsWriter(object):
  36     '''
  37     ContentsWriter writes the Contents-$arch.gz files.
  38     '''
  39     def __init__(self, suite, architecture, overridetype, component = None):
  40         '''
  41         The constructor clones its arguments into a new session object to make
  42         sure that the new ContentsWriter object can be executed in a different
  43         thread.
  44         '''
  45         self.suite = suite.clone()
  46         self.session = self.suite.session()
  47         self.architecture = architecture.clone(self.session)
  48         self.overridetype = overridetype.clone(self.session)
  49         if component is not None:
  50             self.component = component.clone(self.session)
  51         else:
  52             self.component = None
  53
  54     def query(self):
  55         '''
  56         Returns a query object that is doing most of the work.
  57         '''
  58         params = {
  59             'suite':    self.suite.suite_id,
  60             'arch_all': get_architecture('all', self.session).arch_id,
  61             'arch':     self.architecture.arch_id,
  62             'type_id':  self.overridetype.overridetype_id,
  63             'type':     self.overridetype.overridetype,
  64         }
  65
  66         if self.component is not None:
  67             params['component'] = component.component_id
  68             sql = '''
  69 create temp table newest_binaries (
  70     id integer primary key,
  71     package text);
  72
  73 create index newest_binaries_by_package on newest_binaries (package);
  74
  75 insert into newest_binaries (id, package)
  76     select distinct on (package) id, package from binaries
  77         where type = :type and
  78             (architecture = :arch_all or architecture = :arch) and
  79             id in (select bin from bin_associations where suite = :suite)
  80         order by package, version desc;
  81
  82 with
  83
  84 unique_override as
  85     (select o.package, s.section
  86         from override o, section s
  87         where o.suite = :suite and o.type = :type_id and o.section = s.id and
  88         o.component = :component)
  89
  90 select bc.file, substring(o.section from position('/' in o.section) + 1) || '/' || b.package as package
  91     from newest_binaries b, bin_contents bc, unique_override o
  92     where b.id = bc.binary_id and o.package = b.package
  93     order by bc.file, b.package'''
  94
  95         else:
  96             sql = '''
  97 create temp table newest_binaries (
  98     id integer primary key,
  99     package text);
 100
 101 create index newest_binaries_by_package on newest_binaries (package);
 102
 103 insert into newest_binaries (id, package)
 104     select distinct on (package) id, package from binaries
 105         where type = :type and
 106             (architecture = :arch_all or architecture = :arch) and
 107             id in (select bin from bin_associations where suite = :suite)
 108         order by package, version desc;
 109
 110 with
 111
 112 unique_override as
 113     (select distinct on (o.package, s.section) o.package, s.section
 114         from override o, section s
 115         where o.suite = :suite and o.type = :type_id and o.section = s.id
 116         order by o.package, s.section, o.modified desc)
 117
 118 select bc.file, substring(o.section from position('/' in o.section) + 1) || '/' || b.package as package
 119     from newest_binaries b, bin_contents bc, unique_override o
 120     where b.id = bc.binary_id and o.package = b.package
 121     order by bc.file, b.package'''
 122
 123         return self.session.query("file", "package").from_statement(sql). \
 124             params(params)
 125
 126     def formatline(self, filename, package_list):
 127         '''
 128         Returns a formatted string for the filename argument.
 129         '''
 130         package_list = ','.join(package_list)
 131         return "%-60s%s\n" % (filename, package_list)
 132
 133     def fetch(self):
 134         '''
 135         Yields a new line of the Contents-$arch.gz file in filename order.
 136         '''
 137         last_filename = None
 138         package_list = []
 139         for filename, package in self.query().yield_per(100):
 140             if filename != last_filename:
 141                 if last_filename is not None:
 142                     yield self.formatline(last_filename, package_list)
 143                 last_filename = filename
 144                 package_list = []
 145             package_list.append(package)
 146         yield self.formatline(last_filename, package_list)
 147         # end transaction to return connection to pool
 148         self.session.rollback()
 149
 150     def get_list(self):
 151         '''
 152         Returns a list of lines for the Contents-$arch.gz file.
 153         '''
 154         return [item for item in self.fetch()]
 155
 156     def output_filename(self):
 157         '''
 158         Returns the name of the output file.
 159         '''
 160         values = {
 161             'root': Config()['Dir::Root'],
 162             'suite': self.suite.suite_name,
 163             'architecture': self.architecture.arch_string
 164         }
 165         if self.component is None:
 166             return "%(root)s%(suite)s/Contents-%(architecture)s.gz" % values
 167         values['component'] = self.component.component_name
 168         return "%(root)s%(suite)s/%(component)s/Contents-%(architecture)s.gz" % values
 169
 170     def write_file(self):
 171         '''
 172         Write the output file.
 173         '''
 174         command = ['gzip', '--rsyncable']
 175         output_file = open(self.output_filename(), 'w')
 176         pipe = Popen(command, stdin = PIPE, stdout = output_file).stdin
 177         for item in self.fetch():
 178             pipe.write(item)
 179         pipe.close()
 180         output_file.close()
 181
 182
 183 class ContentsScanner(object):
 184     '''
 185     ContentsScanner provides a threadsafe method scan() to scan the contents of
 186     a DBBinary object.
 187     '''
 188     def __init__(self, binary):
 189         '''
 190         The argument binary is the actual DBBinary object that should be
 191         scanned.
 192         '''
 193         self.binary_id = binary.binary_id
 194
 195     def scan(self, dummy_arg = None):
 196         '''
 197         This method does the actual scan and fills in the associated BinContents
 198         property. It commits any changes to the database. The argument dummy_arg
 199         is ignored but needed by our threadpool implementation.
 200         '''
 201         session = DBConn().session()
 202         binary = session.query(DBBinary).get(self.binary_id)
 203         for filename in binary.scan_contents():
 204             binary.contents.append(BinContents(file = filename))
 205         session.commit()
 206         session.close()
 207
 208     @classmethod
 209     def scan_all(class_, limit = None):
 210         '''
 211         The class method scan_all() scans all binaries using multiple threads.
 212         The number of binaries to be scanned can be limited with the limit
 213         argument. Returns the number of processed and remaining packages as a
 214         dict.
 215         '''
 216         session = DBConn().session()
 217         query = session.query(DBBinary).filter(DBBinary.contents == None)
 218         remaining = query.count
 219         if limit is not None:
 220             query = query.limit(limit)
 221         processed = query.count()
 222         threadpool = ThreadPool()
 223         for binary in query.yield_per(100):
 224             threadpool.queueTask(ContentsScanner(binary).scan)
 225         threadpool.joinAll()
 226         remaining = remaining()
 227         session.close()
 228         return { 'processed': processed, 'remaining': remaining }