"""Run the importer as a service Separate from importing and pushing individual packages, an importer service maintains a set of repositories by repeatedly importing them when required. """ import collections import datetime import enum import functools import pprint import sqlite3 import time # The 'time' attribute is the publication date, as a timestamp, of the SPPHR # corresponding to the URL stored in the 'link' attribute. Timestamp = collections.namedtuple('Timestamp', ['time', 'link']) class ImportRequest( collections.namedtuple('ImportRequest', ['package', 'level']) ): """Representation of an import request from the importer service :ivar str package: the name of the source package whose import is requested :ivar Level level: whether the request is for an incremental or full reimport """ __slots__ = () class Level(enum.Enum): # These numbers are hardcoded because they must match what we use in # our database schema. INCREMENTAL = 0 REIMPORT = 1 def is_database_locked_exception(e): """Determine if a given exception is sqlite3's "database locked" exception :param Exception e: the exception to examine :rtype: bool :returns: True if the exception is sqlite3's "database is locked" exception, or False otherwise """ # There doesn't seem to be any better way to detect the sqlite "locked" # exception than this. return ( isinstance(e, sqlite3.OperationalError) and e.args[0] == 'database is locked' ) def _read_lock(method): """Method decorator to wrap database reads into a transaction To permit deadlock-free concurrency, all sqlite3 database operations need to be within explicit transactions. Transactions must not be left open as this will block other instances. Since the transactions implemented here cannot be reentrant, nothing inside the wrapped method can call a similarly decorated function. To prevent accidents, this is asserted using a "lock". The class that uses methods decorated by this function must initialize self._locked to False and provide a self.db attribute of type sqlite3.Connection that is the open sqlite3 connection constructed with sqlite3.connect(..., isolation_level=None). This decorator may retry the wrapped method multiple times in case of database locking timeouts. The wrapped method must accept this without error - for example by avoiding side effects or by being idempotent. :param callable method: the method to decorate. :raises AssertionError: (at wrapped method runtime) if a decorated method has already begun a transaction. :returns: the wrapped method """ @functools.wraps(method) def wrapper(self, *args, **kwargs): # Detect accidental reentrancy: a _read_lock or _write_lock # -decorated method must only appear once in the call stack. assert not self._locked self._locked = True try: # From https://sqlite.org/lang_transaction.html: "Internally, the # BEGIN DEFERRED statement merely sets a flag on the database # connection that turns off the automatic commit that would # normally occur when the last statement finishes." # This command should therefore never fail and so doesn't need # retrying on lock acquisition failure. We do explicitly call # BEGIN/ROLLBACK though in order to be robust against being wrapped # around a database modification by accident. self.db.execute('BEGIN DEFERRED') while True: try: # We may get a "Database is locked" against the database # during times of heavy system I/O when we called the # wrapped function. In this case, retry the entire query. return method(self, *args, **kwargs) except sqlite3.OperationalError as e: if is_database_locked_exception(e): continue raise e finally: self.db.execute('ROLLBACK') finally: self._locked = False return wrapper def _write_lock(method): """Method decorator to wrap database writes into a transaction To permit deadlock-free concurrency, all sqlite3 database access need to be within explicit transactions and those transactions completed when not used. Further, transactions that write must begin with 'BEGIN IMMEDIATE'. This decorator makes this simple to do for methods in this class. Since the transactions implemented here cannot be reentrant, nothing inside the wrapped method can call a similarly decorated function. To prevent accidents, this is asserted using a "lock". The class that uses methods decorated by this function must initialize self._locked to False and provide a self.db attribute of type sqlite3.Connection that is the open sqlite3 connection constructed with sqlite3.connect(..., isolation_level=None). :param callable method: the method to decorate. :raises AssertionError: (at wrapped method runtime) if a decorated method has already begun a transaction. :returns: the wrapped method """ @functools.wraps(method) def wrapper(self, *args, **kwargs): # Detect accidental reentrancy: a _read_lock or _write_lock # -decorated method must only appear once in the call stack. assert not self._locked self._locked = True try: while True: try: self.db.execute('BEGIN IMMEDIATE') break except sqlite3.OperationalError as e: if is_database_locked_exception(e): continue raise e try: result = method(self, *args, **kwargs) except: self.db.execute('ROLLBACK') raise else: # We call 'COMMIT' explicitly instead of using # self.db.commit(). The latter is integrated with the # sqlite3's built-in transaction handling, which we've # disabled. How it behaves is therefore not precisely # defined, but what we do know is that it's safe for us to # take over all transaction-related commands explicitly # ourselves. self.db.execute('COMMIT') return result finally: self._locked = False return wrapper class State(object): """Hold all persisted importer service state State is kept in an sqlite3 database. The public interface provides a data model as follows: A Timestamp (a namedtuple defined above) is maintained for each of ['debian', 'ubuntu'] which lists where we are up to in Launchpad's ever-growing list of source package publications for these distributions. read_timestamps() and write_timestamps() are used to read and write the timestamps. A list of outstanding "package import requests" is maintained. Requests are added using request_imports(). Each request is identified by a (source_package_name, timestamp) tuple. The timestamp nominally identifies the time the request was made. The request is treated as fulfilled when a package import has been completed and pushed to Launchpad and that import began after the request timestamp; this event is notified to this class using mark_import_outcomes(). If a request is made while an import is in progress, the request timestamp will be after the time the import began. Completion of the in progress import will not fulfil the request; an additional import run will be needed. Requests can be regular import requests, or reimport import requests. Reimport requests are mostly treated as separate requests, except that they can supersede regular import requests and fulfilling a reimport request also fulfils regular import requests (subject to timing of the fulfilment). Each package import request additionally has a status and a priority. The status is one of ['needed', 'done', 'error', 'superseded', 'running']. 'done' means done successfully. If an import fails, the status will change to 'error'. 'done' and 'superseded' are not expected to be seen outside the implementation since 'superseded' and 'done' entries are removed automatically. These statuses exist to make the logic easier to follow during development, debugging and testing. A package import request is considered 'superseded' if a request with a higher timestamp exists for the same source package name. This includes running requests, which means that it is possible for a package import request to be running but have vanished from the database because it has been superseded by a newer one. The priority is an integer (low value means high priority) that can be used to order requests so that manual requests can be made that are prioritised over regular ones. """ # Implementation details: # The request table is kept normalized before a commit with private method # State._normalize(), which detects and removes 'superseded' and 'done' # requests. If the database is committed without normalization then query # method results are undefined. Responsibility for keeping the database # committed normalized rests with this class. It is a bug in this class if # there exists any set of public method calls that results in changes # having been committed to the database without normalization. # # isolation_level=None, _read_lock() and _write_lock() are used to ensure # that concurrent database access (ie. multiple instances of this class; # one for the poller and one for the broker in the IPC case) works # deadlock-free. All database accesses must be wrapped in an explicit # transaction. See https://www.sqlite.org/lockingv3.html, # https://www.sqlite.org/isolation.html (particularly the "Isolation and # Concurrency" section) and https://sqlite.org/lang_transaction.html for # details. The transaction wrapping decorators are used on all public # methods. They are never used for internal methods, which assume that # transactions are already in progress as necessary. Tests may call # internal methods directly without transaction wrapping if they aren't # testing the transaction wrapping. LAST_SEEN_DISTROS = frozenset(['debian', 'ubuntu']) def __init__(self, db_path): # isolation_level=None, together with use of our _read_lock() and # _write_lock() decorators, is required to make concurrency work # deadlock-free self.db = sqlite3.connect(db_path, isolation_level=None) self._init_db(self.db) self._locked = False @classmethod def _init_db(cls, db): """Make sure the database has the schema defined :param sqlite3.Connection db: sqlite3 connection to use """ # Instead of using the _write_lock() decorator, we wrap the transaction # manually in this method for this special case. self would be # unavailable anyway because this is a classmethod. db.execute('BEGIN IMMEDIATE') db.execute('''CREATE TABLE IF NOT EXISTS last_seen_publication ( distro TEXT PRIMARY KEY, timestamp REAL NOT NULL, link TEXT ) ''') db.execute('''CREATE TABLE IF NOT EXISTS request ( srcpkg TEXT NOT NULL, timestamp REAL NOT NULL, status TEXT NOT NULL DEFAULT 'needed', priority INTEGER NOT NULL, reimport INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (srcpkg, timestamp) )''') db.execute('''CREATE TABLE IF NOT EXISTS ipc_worker ( identity TEXT NOT NULL, package TEXT NOT NULL, reimport INTEGER NOT NULL, reckoning_time INTEGER NOT NULL, PRIMARY KEY (identity) )''') # Handle schema migration. This can be dropped after all instances can # be expected to have migrated. cls._ensure_reimport_column(db) db.execute('COMMIT') @staticmethod def _ensure_reimport_column(db): """Add the reimport column to the request table if it doesn't exist The reimport column was added later so this method provides the schema migration. It can be dropped after all instances can be expected to have migrated. :param sqlite3.Connection db: the database to act upon :returns: None """ cursor = db.cursor() cursor.execute('PRAGMA table_info(request)') # Determine which column number contains the column name col_name_index = { col[0]: i for i, col in enumerate(cursor.description) }['name'] # Determine the set of column names present in the table columns = {row[col_name_index] for row in cursor.fetchall()} # Add the reimport column if it is not already present if 'reimport' not in columns: db.execute('''ALTER TABLE request ADD COLUMN reimport INTEGER NOT NULL DEFAULT 0 ''') @_read_lock def _dump(self, table_name): """Return the contents of a table as a frozenset of tuples This is useful for testing. :returns: a set of tuples where the tuple is ordered the same as the request table definition. :rtype: depends on the table """ cursor = self.db.cursor() # I wouldn't normally use * in case the column definition changes, but # in this case this method is defined to be identical to the table # definition and we will detect any mismatch by tests failing, so I # think in this case it makes sense. cursor.execute('SELECT * FROM %s' % table_name) return frozenset([tuple(row) for row in cursor.fetchall()]) def _dump_request(self): """Return the contents of the request table as a frozenset of tuples This is useful for testing. :returns: a set of tuples where the tuple is ordered the same as the request table definition. :rtype: set(tuple(str, float, str, int, int)) """ return self._dump('request') def _dump_ipc_worker(self): """Return the contents of the request table as a frozenset of tuples This is useful for testing. :returns: a set of tuples where the tuple is ordered the same as the request table definition. :rtype: set(tuple(str, str, int, float)) """ return self._dump('ipc_worker') @staticmethod def _mark_superseded(db, assert_no_action=False): """Mark superseded request table rows as status='superseded' :param sqlite3.Connection db: sqlite3 connection to use :param bool assert_no_action: if True, assert that no changes were needed """ cursor = db.cursor() # This SQL statement is a little obtuse but does exactly what we want: # mark superseded requests as superseded. For the purposes of this # statement, we're treating a request for a regular import for a # package and a request for a reimport for the same package as two # distinct requests. # # The meat is in the outer join. By joining to ourselves on # srcpkg=srcpkg, reimport=reimport and timestamp < timestamp, we end up # with every element of request on the left and something on the right # only when there exists a higher timestamp. By WHERE NOT NULL on that, # we eliminate rows where the request (on the left) has a higher # timestamp (ie. it isn't superseded). What's left (on the left) are # requests that have been superseded. This result is then fed to INSERT # OR REPLACE but with the status changed to 'superseded'. In practice # every attempted "INSERT" will conflict and be resolved with "REPLACE" # (thanks to the primary key). This has the effect of changing all the # rows that are superseded to having status set to 'superseded'. cursor.execute(""" INSERT OR REPLACE INTO request (srcpkg, timestamp, status, priority, reimport) SELECT req_left.srcpkg, req_left.timestamp, 'superseded', req_left.priority, req_left.reimport FROM request AS req_left LEFT OUTER JOIN request AS req_right ON req_left.srcpkg = req_right.srcpkg AND req_left.reimport = req_right.reimport AND req_left.timestamp < req_right.timestamp WHERE req_right.timestamp IS NOT NULL AND req_left.status != 'running' """) assert not assert_no_action or not cursor.rowcount # If a regular import request for a package precedes (based on # timestamp) a reimport request for the same package, we can treat the # former request as superseded. This SQL statement marks such requests # as superseded using the same technique as the statement above. cursor.execute(""" INSERT OR REPLACE INTO request (srcpkg, timestamp, status, priority, reimport) SELECT req_left.srcpkg, req_left.timestamp, 'superseded', req_left.priority, req_left.reimport FROM request AS req_left LEFT OUTER JOIN request AS req_right ON req_left.srcpkg = req_right.srcpkg AND req_left.timestamp < req_right.timestamp WHERE req_right.timestamp IS NOT NULL AND req_left.status != 'running' AND req_left.reimport = 0 AND req_right.reimport = 1 """) assert not assert_no_action or not cursor.rowcount @staticmethod def _delete_superseded(db, assert_no_action=False): """Delete request table rows marked status='superseded' :param sqlite3.Connection db: sqlite3 connection to use :param bool assert_no_action: if True, assert that no changes were needed """ cursor = db.cursor() cursor.execute("DELETE FROM request WHERE status='superseded'") assert not assert_no_action or not cursor.rowcount @staticmethod def _delete_done(db, assert_no_action=False): """Delete request table rows marked status='done' status='done' applies only to successfully completed requests; requests that failed are marked status='error'. :param sqlite3.Connection db: sqlite3 connection to use :param bool assert_no_action: if True, assert that no changes were needed """ cursor = db.cursor() cursor.execute("DELETE FROM request WHERE status='done'") assert not assert_no_action or not cursor.rowcount @classmethod def _normalize(cls, db, assert_no_action=False): """Normalize the database according to the defined normalization rules Normalization rules are defined in the docstring for this class. This method does not commit the changes. A database commit after all required changes are made is the caller's responsibility. :param sqlite3.Connection db: sqlite3 connection to use :param bool assert_no_action: if True, assert that no changes were needed """ cls._mark_superseded(db=db, assert_no_action=assert_no_action) cls._delete_superseded(db=db, assert_no_action=assert_no_action) cls._delete_done(db=db, assert_no_action=assert_no_action) def _write_timestamps(self, timestamps): """Write timestamp values to data_directory This method is symmetrical to read_timestamps(). :param dict(str, Timestamp) timestamps: the dictionary must contain 'debian' and 'ubuntu' keys, each of which is a Timestamp namedtuple that corresponds to where we are up to in Launchpad against that distribution's publication history """ assert set(timestamps.keys()) == self.LAST_SEEN_DISTROS for distro in self.LAST_SEEN_DISTROS: self.db.execute( '''INSERT OR REPLACE INTO last_seen_publication (distro, timestamp, link) VALUES (?, ?, ?)''', [distro, timestamps[distro].time, timestamps[distro].link], ) @_read_lock def read_timestamps(self, num_days_ago=1, current_time=None): """Read saved timestamp values from data_directory This method is symmetrical to write_timestamps(). :param: int num_days_ago: number of days ago to start iterating Launchpad publishes :param: float current_time: the "current" time to use for synthesis if no timestamps were previously stored. If None, then time.time() is used. This option is useful for testing. :returns: the timestamp dictionary previously stored with write_timestamps(), or, if timestamps were never previously stored, a timestamp dictionary synthesized based on current_time and num_days_ago and with links set to None :rtype: dict(str, Timestamp) """ current_time = current_time or time.time() # default: now cursor = self.db.cursor() cursor.execute( 'SELECT distro, timestamp, link FROM last_seen_publication', ) timestamps = {} for distro, _time, link in cursor: timestamps[distro] = Timestamp(time=_time, link=link) if not timestamps: _start = current_time - (num_days_ago * 24 * 60 * 60) return { distro: Timestamp(time=_start, link=None) for distro in self.LAST_SEEN_DISTROS } return timestamps @_write_lock def request_imports( self, pkgnames, timestamps=None, priority=20, request_time=None ): """Add package import requests to the needed import list :param iterable(str) pkgnames: source package names to import :param dict(str, Timestamp) timestamps: timestamps with which to update the last_seen_publication table :param int priority: the request priority (lower integer = higher priority) :param float request_time: the reckoning time of the request If timestamps is provided, then a call to write_timestamps is made and committed atomically at the same time as the request table update. This is useful when catching up new package import requests against new Launchpad publications. It is not expected to be used for ad-hoc package import requests. If request_time is provided then the time provided is noted as the time of the request instead of acquiring it using time.time(). This is useful for testing. """ request_time = request_time or time.time() # default: now self.db.executemany( '''INSERT OR REPLACE INTO request (srcpkg, priority, timestamp) VALUES (?, ?, ?)''', [(pkg, priority, request_time) for pkg in pkgnames] ) self._normalize(self.db) if timestamps: self._write_timestamps(timestamps) @_read_lock def get_pending_imports(self): """Return a sequence of source package names that are pending import Source package names are returned in order of priority. Requests for reimports are treated as separate for requests for incremental imports, but if a package has both types of request marked as 'needed', only the reimport request will be returned. However, if a package has one type of request marked as 'running', this method will currently still return the other type of request as 'needed' if it exists. :rtype: sequence(ImportRequest) :returns: the package requests that are pending. """ # The LEFT OUTER JOIN serves to "disqualify" any request that has a # matching request (one with the same srcpkg) that is already running. # Requests on the left are attempted to be matched with a disqualifying # request on the right. If a request on the left has no matching # disqualifying request, then the right side will be NULL. So filtering # the result where the right side is NULL gives us the qualifying # requests only. # # We then GROUP BY to collapse multiple requests for the same package # into one, such as if there is both a reimport and an incremental # import pending. max(req_left.reimport) causes us to favour reimport # (reimport=1) over incremental (reimport=0). cursor = self.db.cursor() cursor.execute("""SELECT req_left.srcpkg, max(req_left.reimport) FROM request AS req_left LEFT OUTER JOIN request AS req_right ON req_left.srcpkg = req_right.srcpkg AND req_right.status == 'running' WHERE req_right.srcpkg IS NULL AND req_left.status = 'needed' GROUP BY req_left.srcpkg ORDER BY req_left.priority """) return [ ImportRequest(package=package, level=ImportRequest.Level(reimport)) for package, reimport in cursor ] @_write_lock def mark_running(self, requests): self._mark_running(requests) def _mark_running(self, requests): """Mark packages as being imported Change the status to 'running' for each of the listed import requests. This list must be the set or a subset of packages listed by a previous call to get_pending_imports(). :param sequence(ImportRequest) requests: requests to mark as running :returns: None """ self.db.executemany( """ UPDATE request SET status='running' WHERE srcpkg=? and reimport=?""", ((request.package, request.level.value) for request in requests), ) self._normalize(self.db) @_write_lock def unmark_running(self): """Mark packages marked as running as failed Change the status to 'error' for every package that is currently listed as running. This can be used, for example, on restart, where we know that there are no packages running so they must all have failed. """ self.db.execute( "UPDATE request SET status='error' WHERE status='running'" ) self._normalize(self.db) @_write_lock def mark_import_outcomes( self, reckoning_time, completed_requests, failed_requests, ): """Mark some package import requests as complete :param float reckoning_time: a time prior to the time the imports began. Only requests whose timestamps are before this time are considered to have been completed by the report made by this method call. :param sequence(ImportRequest) completed_requests: requests whose corresponding imports have completed successfully. :param sequence(ImportRequest) failed_pkgs: requests whose corresponding imports have failed. """ # A reimport also fulfils an incremental import request. However an # incremental import does not fulfil a reimport request. Since reimport # is an integer (either 0 or 1) in sqlite3, we can use the '<=' # operator to mark as done any request that is fulfilled but avoid # marking a reimport request as done if we only completed an # incremental import. self.db.executemany( """UPDATE request SET status='done' WHERE srcpkg=? AND reimport <= ? AND timestamp < ?""", [ (request.package, request.level.value, reckoning_time) for request in completed_requests ], ) # To mark failed requests as errored, we don't need do any "one type # fulfils the other type" thinking like above. We just match exactly # against the request and mark that one as errored. self.db.executemany( """UPDATE request SET status='error' WHERE srcpkg=? AND reimport = ? AND timestamp < ?""", [ (request.package, request.level.value, reckoning_time) for request in failed_requests ], ) self._normalize(self.db) @_write_lock def assign_ipc_worker(self, identity, request, reckoning_time=None): """Mark a request as in progress by an IPC worker This will mark the request as "running" in the request table and also add an entry to the ipc_worker table in a single transaction. :param str identity: the identity of the worker (an arbitrary string) :param ImportRequest request: the import request the worker will work on :param float reckoning_time: the latest point in time that is definitely before the worker begins work on this request. If None, then the current time is used. :returns: None """ if reckoning_time is None: reckoning_time = time.time() self.db.execute( '''INSERT INTO ipc_worker (identity, package, reimport, reckoning_time) VALUES (?, ?, ?, ?)''', (identity, request.package, request.level.value, reckoning_time), ) self._mark_running([request]) @_read_lock def lookup_ipc_worker(self, identity): """Find details of a request that is in progress by an IPC worker The request must exist in the ipc_worker table. :param str identity: the identity of the worker (an arbitrary string) :rtype: tuple(str, ImportRequest.Level, float) :returns: a tuple of the package name, the import request level, and the reckoning time, as previously stored in the ipc_worker table. """ cursor = self.db.cursor() cursor.execute(''' SELECT package, reimport, reckoning_time FROM ipc_worker WHERE identity=?''', (identity,), ) row = cursor.fetchone() assert row package, reimport, reckoning_time = row return package, ImportRequest.Level(reimport), reckoning_time @_write_lock def clear_ipc_worker(self, identity): """Clear our record of work being done by a particular worker Unmark a worker from working on a particular request. If the request is complete, it is assumed that the request itself has been marked as 'done' already. If we knew about the given worker and it is recorded as still working on a request, mark the worker as no longer working on that request and that request as having errored. This provides a "self-healing" property in case the worker and broker end up out-of-sync (eg. if one or the other is restarted). If the given worker is not recorded as currently doing any work, then this method does nothing and succeeds. :param str identity: the identity of the worker (an arbitrary string) :returns: None """ cursor = self.db.cursor() cursor.execute( 'SELECT package, reimport FROM ipc_worker WHERE identity=?', (identity,), ) row = cursor.fetchone() if row: package, reimport = row cursor.execute(""" UPDATE request SET status='error' WHERE srcpkg=? AND status='running' AND reimport=?""", (package, reimport), ) cursor.execute( 'DELETE FROM ipc_worker WHERE identity=?', (identity,), ) def read_package_list(path): """Read a list of package names from a text file The file must contain one package name per line. Whitespace is stripped. Blank lines are permitted. '#' is the comment character: lines that begin with '#' (ignoring whitespace) are ignored. Comments are not currently permitted on the same line as a package name. :param str path: the path to the text file :rtype: list(str) :returns: the list of packages in the given file """ try: fobj = open(path, 'r') except FileNotFoundError: return [] with fobj: return [ line.strip() for line in fobj if not line.startswith('#') ] def has_import_repository(lp, team, package): """Determine if an import repository exists for a given package :param launchpadlib.launchpad.Launchpad lp: Launchpad object :param str team: name of Launchpad team that contains imported repositories :param str package: source package name :rtype: bool :returns: True if the import repository exists, or False otherwise """ return bool( lp.git_repositories.getByPath( path='~%s/ubuntu/+source/%s' % (team, package), ), ) def request_new_imports( lp, num_days_ago, service_state, filter_func=None, request_time=None, ): """Import all new publications since a prior execution The purpose of this function is the side-effect of moving along service_state's last_seen_publication timestamps, sweeping any publications found along the way into import requests as required. :param launchpadlib.launchpad.Launchpad lp: the Launchpad connection to use to query Launchpad. :param int num_days_ago: how many days ago to start iterating Launchpad publications at if we've never looked at Launchpad publications before :param importer_service.State service_state: the persistent data on the status of the import :param filter_func: a function that when called with a sole parameter string providing a package name, returns a bool that specifies whether or not that package should be imported. :param float request_time: the reckoning time of the generated requests. If this evaluates to False, use the current time. Useful for testing. :returns: None """ timestamps = service_state.read_timestamps(num_days_ago) # filtered_pkgnames is the list of source package names across all # distributions we will want to process, based upon # scriptutils.should_import_srcpkg filtered_pkgnames = set() for dist_name in ['debian', 'ubuntu']: timestamp = timestamps[dist_name] # dist_newest_spphr is the most recent publication record with a # valid published date in the distribution dist_newest_spphr = None # dist_filtered_pkgnames is the set of source package names in # the dist_name distribution that qualify for import according to our # allowlists, denylists and phasing requirements dist_filtered_pkgnames = set() print( "Examining publications in %s since %s" % ( dist_name, datetime.datetime.fromtimestamp( timestamp.time, ).strftime("%Y-%m-%d %H:%M:%S"), ) ) # spph is the raw publication history for a distribution from # Launchpad, sorted by publication date in reverse chronological order dist = lp.distributions[dist_name] spph = dist.main_archive.getPublishedSources(order_by_date=True) if not spph: print("No publishing data found in %s" % dist_name) continue for spphr in spph: # this is the matching upload to the last publish seen if str(spphr) == timestamp.link: break # only check if we should stop iterating due to the # timestamps if there is a timestamp to compare to, which # means the source package is actually published. if spphr.date_published: if not dist_newest_spphr: dist_newest_spphr = spphr # stop iterating (backwards chronologically) when we see # a publish timestamp before the last run if spphr.date_published.timestamp() < timestamp.time: break package = spphr.source_package_name if filter_func and filter_func(package): dist_filtered_pkgnames.add(package) if not dist_filtered_pkgnames: print( "No new relevant publications found in %s relative to %s" % ( dist_name, datetime.datetime.fromtimestamp( timestamp.time ).strftime("%Y-%m-%d %H:%M:%S"), ) ) filtered_pkgnames = filtered_pkgnames | dist_filtered_pkgnames if dist_newest_spphr: # Update timestamp timestamps[dist_name] = timestamp = Timestamp( time=dist_newest_spphr.date_published.timestamp(), link=str(dist_newest_spphr), ) service_state.request_imports( pkgnames=filtered_pkgnames, timestamps=timestamps, request_time=request_time, ) print("Requested imports for:") pprint.pprint(filtered_pkgnames)