Returns a writer object.
'''
values = {
+ 'archive': self.suite.archive.path,
'suite': self.suite.suite_name,
'component': self.component.component_name,
'debtype': self.overridetype.overridetype,
insert into newest_sources (id, source)
select distinct on (source) s.id, s.source from source s
- join files f on f.id = s.file
- join location l on l.id = f.location
+ join files_archive_map af on s.file = af.file_id
where s.id in (select source from src_associations where suite = :suite_id)
- and l.component = :component_id
+ and af.component_id = :component_id
order by source, version desc;'''
self.session.execute(sql_create_temp, params=params)
Returns a writer object.
'''
values = {
+ 'archive': self.suite.archive.path,
'suite': self.suite.suite_name,
'component': self.component.component_name
}
overridetype.overridetype, component.component_name]
contents_writer = BinaryContentsWriter(suite, architecture, overridetype, component)
contents_writer.write_file()
+ session.close()
return log_message
def source_helper(suite_id, component_id):
log_message = [suite.suite_name, 'source', component.component_name]
contents_writer = SourceContentsWriter(suite, component)
contents_writer.write_file()
+ session.close()
return log_message
class ContentsWriter(object):
class_.logger.log(result)
@classmethod
- def write_all(class_, logger, suite_names = [], component_names = [], force = False):
+ def write_all(class_, logger, archive_names = [], suite_names = [], component_names = [], force = False):
'''
Writes all Contents files for suites in list suite_names which defaults
to all 'touchable' suites if not specified explicitely. Untouchable
class_.logger = logger
session = DBConn().session()
suite_query = session.query(Suite)
+ if len(archive_names) > 0:
+ suite_query = suite_query.join(Suite.archive).filter(Archive.archive_name.in_(archive_names))
if len(suite_names) > 0:
suite_query = suite_query.filter(Suite.suite_name.in_(suite_names))
component_query = session.query(Component)
if len(component_names) > 0:
component_query = component_query.filter(Component.component_name.in_(component_names))
if not force:
- suite_query = suite_query.filter_by(untouchable = False)
+ suite_query = suite_query.filter(Suite.untouchable == False)
deb_id = get_override_type('deb', session).overridetype_id
udeb_id = get_override_type('udeb', session).overridetype_id
pool = Pool()
UnpackedSource extracts a source package into a temporary location and
gives you some convinient function for accessing it.
'''
- def __init__(self, dscfilename):
+ def __init__(self, dscfilename, tmpbasedir=None):
'''
The dscfilename is a name of a DSC file that will be extracted.
'''
- temp_directory = mkdtemp(dir = Config()['Dir::TempPath'])
+ basedir = tmpbasedir if tmpbasedir else Config()['Dir::TempPath']
+ temp_directory = mkdtemp(dir = basedir)
self.root_directory = os.path.join(temp_directory, 'root')
command = ('dpkg-source', '--no-copy', '--no-check', '-q', '-x',
dscfilename, self.root_directory)