import sqlalchemy
from sqlalchemy import create_engine, Table, MetaData, Column, Integer
-from sqlalchemy.orm import sessionmaker, mapper, relation
+from sqlalchemy.orm import sessionmaker, mapper, relation, object_session
from sqlalchemy import types as sqltypes
# Don't remove this, we re-export the exceptions to scripts which import us
# Patch in support for the debversion field type so that it works during
# reflection
-class DebVersion(sqltypes.Text):
- """
- Support the debversion type
- """
+try:
+ # that is for sqlalchemy 0.6
+ UserDefinedType = sqltypes.UserDefinedType
+except:
+ # this one for sqlalchemy 0.5
+ UserDefinedType = sqltypes.TypeEngine
+class DebVersion(UserDefinedType):
def get_col_spec(self):
return "DEBVERSION"
+ def bind_processor(self, dialect):
+ return None
+
+ # ' = None' is needed for sqlalchemy 0.5:
+ def result_processor(self, dialect, coltype = None):
+ return None
+
sa_major_version = sqlalchemy.__version__[0:3]
if sa_major_version in ["0.5", "0.6"]:
from sqlalchemy.databases import postgres
################################################################################
-__all__ = ['IntegrityError', 'SQLAlchemyError']
+__all__ = ['IntegrityError', 'SQLAlchemyError', 'DebVersion']
################################################################################
################################################################################
class Architecture(object):
- def __init__(self, *args, **kwargs):
- pass
+ def __init__(self, arch_string = None, description = None):
+ self.arch_string = arch_string
+ self.description = description
def __eq__(self, val):
if isinstance(val, str):
__all__.append('get_architecture')
+# TODO: should be removed because the implementation is too trivial
@session_wrapper
def get_architecture_suites(architecture, session=None):
"""
@return: list of Suite objects for the given name (may be empty)
"""
- q = session.query(Suite)
- q = q.join(SuiteArchitecture)
- q = q.join(Architecture).filter_by(arch_string=architecture).order_by('suite_name')
-
- ret = q.all()
-
- return ret
+ return get_architecture(architecture, session).suites
__all__.append('get_architecture_suites')
################################################################################
class Fingerprint(object):
- def __init__(self, *args, **kwargs):
- pass
+ def __init__(self, fingerprint = None):
+ self.fingerprint = fingerprint
def __repr__(self):
return '<Fingerprint %s>' % self.fingerprint
('OverrideSuite', 'overridesuite')]
class Suite(object):
- def __init__(self, *args, **kwargs):
- pass
+ def __init__(self, suite_name = None, version = None):
+ self.suite_name = suite_name
+ self.version = version
def __repr__(self):
return '<Suite %s>' % self.suite_name
return "\n".join(ret)
+ def get_architectures(self, skipsrc=False, skipall=False):
+ """
+ Returns list of Architecture objects
+
+ @type skipsrc: boolean
+ @param skipsrc: Whether to skip returning the 'source' architecture entry
+ (Default False)
+
+ @type skipall: boolean
+ @param skipall: Whether to skip returning the 'all' architecture entry
+ (Default False)
+
+ @rtype: list
+ @return: list of Architecture objects for the given name (may be empty)
+ """
+
+ q = object_session(self).query(Architecture). \
+ filter(Architecture.suites.contains(self))
+ if skipsrc:
+ q = q.filter(Architecture.arch_string != 'source')
+ if skipall:
+ q = q.filter(Architecture.arch_string != 'all')
+ return q.order_by(Architecture.arch_string).all()
+
__all__.append('Suite')
@session_wrapper
################################################################################
+# TODO: remove SuiteArchitecture class
class SuiteArchitecture(object):
def __init__(self, *args, **kwargs):
pass
__all__.append('SuiteArchitecture')
+# TODO: should be removed because the implementation is too trivial
@session_wrapper
def get_suite_architectures(suite, skipsrc=False, skipall=False, session=None):
"""
@return: list of Architecture objects for the given name (may be empty)
"""
- q = session.query(Architecture)
- q = q.join(SuiteArchitecture)
- q = q.join(Suite).filter_by(suite_name=suite)
-
- if skipsrc:
- q = q.filter(Architecture.arch_string != 'source')
-
- if skipall:
- q = q.filter(Architecture.arch_string != 'all')
-
- q = q.order_by('arch_string')
-
- return q.all()
+ return get_suite(suite, session).get_architectures(skipsrc, skipall)
__all__.append('get_suite_architectures')
################################################################################
class Uid(object):
- def __init__(self, *args, **kwargs):
- pass
+ def __init__(self, uid = None, name = None):
+ self.uid = uid
+ self.name = name
def __eq__(self, val):
if isinstance(val, str):
'binary_acl',
'binary_acl_map',
'build_queue',
- 'build_queue_files',
'changelogs_text',
'component',
'config',
'files',
'fingerprint',
'keyrings',
- 'changes',
'keyring_acl_map',
'location',
'maintainer',
'suite',
'uid',
'upload_blocks',
+ # The following tables have primary keys but sqlalchemy
+ # version 0.5 fails to reflect them correctly with database
+ # versions before upgrade #41.
+ #'changes',
+ #'build_queue_files',
)
tables_no_primary = (
'suite_src_formats',
'suite_build_queue_copy',
'udeb_contents',
+ # see the comment above
+ 'changes',
+ 'build_queue_files',
+ )
+
+ 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',
+ 'changelogs',
+ 'file_arch_suite',
+ 'newest_all_associations',
+ 'newest_any_associations',
+ 'newest_source',
+ 'newest_src_association',
+ 'obsolete_all_associations',
+ 'obsolete_any_associations',
+ 'obsolete_any_by_all_associations',
+ 'obsolete_src_associations',
+ 'source_suite',
+ 'src_associations_bin',
+ 'src_associations_src',
+ 'suite_arch_by_name',
)
- # Sqlalchemy fails to reflect the SERIAL type correctly and that
- # is why we have to use a workaround. It can be removed as soon
- # as we switch to version 0.6.
+ # Sqlalchemy version 0.5 fails to reflect the SERIAL type
+ # correctly and that is why we have to use a workaround. It can
+ # be removed as soon as we switch to version 0.6.
for table_name in tables_with_primary:
table = Table(table_name, self.db_meta, \
Column('id', Integer, primary_key = True), \
table = Table(table_name, self.db_meta, autoload=True)
setattr(self, 'tbl_%s' % table_name, table)
+ for view_name in views:
+ view = Table(view_name, self.db_meta, autoload=True)
+ setattr(self, 'view_%s' % view_name, view)
+
def __setupmappers(self):
mapper(Architecture, self.tbl_architecture,
- properties = dict(arch_id = self.tbl_architecture.c.id))
+ properties = dict(arch_id = self.tbl_architecture.c.id,
+ suites = relation(Suite, secondary=self.tbl_suite_architectures, backref='architectures', order_by='suite_name')))
mapper(Archive, self.tbl_archive,
properties = dict(archive_id = self.tbl_archive.c.id,
projects / dak.git / blobdiff