def __init__(self, *args, **kwargs):
pass
+ def __eq__(self, val):
+ if isinstance(val, str):
+ return (self.arch_string== val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
+ def __ne__(self, val):
+ if isinstance(val, str):
+ return (self.arch_string != val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
def __repr__(self):
return '<Architecture %s>' % self.arch_string
__all__.append('DBBinary')
+def get_suites_binary_in(package, session=None):
+ """
+ Returns list of Suite objects which given C{package} name is in
+
+ @type source: str
+ @param source: DBBinary package name to search for
+
+ @rtype: list
+ @return: list of Suite objects for the given package
+ """
+
+ if session is None:
+ session = DBConn().session()
+
+ return session.query(Suite).join(BinAssociation).join(DBBinary).filter_by(package=package).all()
+
+__all__.append('get_suites_binary_in')
+
def get_binary_from_id(id, session=None):
"""
Returns DBBinary object for given C{id}
__all__.append('get_binary_from_id')
-def get_binaries_from_name(package, session=None):
+def get_binaries_from_name(package, version=None, architecture=None, session=None):
"""
Returns list of DBBinary objects for given C{package} name
@type package: str
@param package: DBBinary package name to search for
+ @type version: str or None
+ @param version: Version to search for (or None)
+
+ @type package: str, list or None
+ @param package: Architectures to limit to (or None if no limit)
+
@type session: Session
@param session: Optional SQL session object (a temporary one will be
generated if not supplied)
"""
if session is None:
session = DBConn().session()
- return session.query(DBBinary).filter_by(package=package).all()
+
+ q = session.query(DBBinary).filter_by(package=package)
+
+ if version is not None:
+ q = q.filter_by(version=version)
+
+ if architecture is not None:
+ if not isinstance(architecture, list):
+ architecture = [architecture]
+ q = q.join(Architecture).filter(Architecture.arch_string.in_(architecture))
+
+ return q.all()
__all__.append('get_binaries_from_name')
+def get_binaries_from_source_id(source_id, session=None):
+ """
+ Returns list of DBBinary objects for given C{source_id}
+
+ @type source_id: int
+ @param source_id: source_id to search for
+
+ @type session: Session
+ @param session: Optional SQL session object (a temporary one will be
+ generated if not supplied)
+
+ @rtype: list
+ @return: list of DBBinary objects for the given name (may be empty)
+ """
+ if session is None:
+ session = DBConn().session()
+ return session.query(DBBinary).filter_by(source_id=source_id).all()
+
+__all__.append('get_binaries_from_source_id')
+
+
def get_binary_from_name_suite(package, suitename, session=None):
### For dak examine-package
### XXX: Doesn't use object API yet
return session.execute(query, vals)
__all__.append('get_binary_components')
+
################################################################################
class Component(object):
def __init__(self, *args, **kwargs):
pass
+ def __eq__(self, val):
+ if isinstance(val, str):
+ return (self.component_name == val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
+ def __ne__(self, val):
+ if isinstance(val, str):
+ return (self.component_name != val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
def __repr__(self):
return '<Component %s>' % self.component_name
__all__.append('PoolFile')
+def check_poolfile(filename, filesize, md5sum, location_id, session=None):
+ """
+ Returns a tuple:
+ (ValidFileFound [boolean or None], PoolFile object or None)
+
+ @type filename: string
+ @param filename: the filename of the file to check against the DB
+
+ @type filesize: int
+ @param filesize: the size of the file to check against the DB
+
+ @type md5sum: string
+ @param md5sum: the md5sum of the file to check against the DB
+
+ @type location_id: int
+ @param location_id: the id of the location to look in
+
+ @rtype: tuple
+ @return: Tuple of length 2.
+ If more than one file found with that name:
+ (None, None)
+ If valid pool file found: (True, PoolFile object)
+ If valid pool file not found:
+ (False, None) if no file found
+ (False, PoolFile object) if file found with size/md5sum mismatch
+ """
+
+ if session is None:
+ session = DBConn().session()
+
+ q = session.query(PoolFile).filter_by(filename=filename)
+ q = q.join(Location).filter_by(location_id=location_id)
+
+ if q.count() > 1:
+ return (None, None)
+ if q.count() < 1:
+ return (False, None)
+
+ obj = q.one()
+ if obj.md5sum != md5sum or obj.filesize != filesize:
+ return (False, obj)
+
+ return (True, obj)
+
+__all__.append('check_poolfile')
+
+
def get_poolfile_by_name(filename, location_id=None, session=None):
"""
Returns an array of PoolFile objects for the given filename and
@return: array of PoolFile objects
"""
- if session is not None:
+ if session is None:
session = DBConn().session()
q = session.query(PoolFile).filter_by(filename=filename)
@return: array of PoolFile objects
"""
- if session is not None:
+ if session is None:
session = DBConn().session()
# TODO: There must be a way of properly using bind parameters with %FOO%
def __init__(self, *args, **kwargs):
pass
+ def __eq__(self, val):
+ if isinstance(val, str):
+ return (self.priority == val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
+ def __ne__(self, val):
+ if isinstance(val, str):
+ return (self.priority != val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
def __repr__(self):
return '<Priority %s (%s)>' % (self.priority, self.priority_id)
def __init__(self, *args, **kwargs):
pass
+ def __eq__(self, val):
+ if isinstance(val, str):
+ return (self.section == val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
+ def __ne__(self, val):
+ if isinstance(val, str):
+ return (self.section != val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
def __repr__(self):
return '<Section %s>' % self.section
__all__.append('source_exists')
-def get_sources_from_name(source, dm_upload_allowed=None, session=None):
+def get_suites_source_in(source, session=None):
"""
- Returns list of DBSource objects for given C{source} name
+ Returns list of Suite objects which given C{source} name is in
@type source: str
@param source: DBSource package name to search for
+ @rtype: list
+ @return: list of Suite objects for the given source
+ """
+
+ if session is None:
+ session = DBConn().session()
+
+ return session.query(Suite).join(SrcAssociation).join(DBSource).filter_by(source=source).all()
+
+__all__.append('get_suites_source_in')
+
+def get_sources_from_name(source, version=None, dm_upload_allowed=None, session=None):
+ """
+ Returns list of DBSource objects for given C{source} name and other parameters
+
+ @type source: str
+ @param source: DBSource package name to search for
+
+ @type source: str or None
+ @param source: DBSource version name to search for or None if not applicable
+
@type dm_upload_allowed: bool
@param dm_upload_allowed: If None, no effect. If True or False, only
return packages with that dm_upload_allowed setting
session = DBConn().session()
q = session.query(DBSource).filter_by(source=source)
+
+ if version is not None:
+ q = q.filter_by(version=version)
+
if dm_upload_allowed is not None:
q = q.filter_by(dm_upload_allowed=dm_upload_allowed)
def __repr__(self):
return '<Suite %s>' % self.suite_name
+ def __eq__(self, val):
+ if isinstance(val, str):
+ return (self.suite_name == val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
+ def __ne__(self, val):
+ if isinstance(val, str):
+ return (self.suite_name != val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
def details(self):
ret = []
for disp, field in SUITE_FIELDS:
def __init__(self, *args, **kwargs):
pass
+ def __eq__(self, val):
+ if isinstance(val, str):
+ return (self.uid == val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
+ def __ne__(self, val):
+ if isinstance(val, str):
+ return (self.uid != val)
+ # This signals to use the normal comparison operator
+ return NotImplemented
+
def __repr__(self):
return '<Uid %s (%s)>' % (self.uid, self.name)