def METHOD_NAME(
    self, from_createdon: datetime, stop_time: datetime
) -> Iterable[Tuple[MetadataChangeProposalWrapper, datetime]]:
    with self.engine.connect() as conn:
        ts = from_createdon
        offset = 0
        while ts.timestamp() <= stop_time.timestamp():
            logger.debug(f"Polling database aspects from {ts}")
            rows = conn.execute(
                self.query,
                since_createdon=ts.strftime(DATETIME_FORMAT),
                limit=self.config.database_query_batch_size,
                offset=offset,
            )
            if not rows.rowcount:
                return
            for i, row in enumerate(rows):
                # TODO: Replace with namedtuple usage once we drop sqlalchemy 1.3
                if hasattr(row, "_asdict"):
                    row_dict = row._asdict()
                else:
                    row_dict = dict(row)
                mcp = self._parse_row(row_dict)
                if mcp:
                    yield mcp, row_dict["createdon"]
            if ts == row_dict["createdon"]:
                offset += i
            else:
                ts = row_dict["createdon"]
                offset = 0