def get_snapshot_closest_to_state_change(
            self,
            state_change_identifier: int,
    ) -> Tuple[int, Any]:
        """ Get snapshots earlier than state_change with provided ID. """

        if not (state_change_identifier == 'latest' or isinstance(state_change_identifier, int)):
            raise ValueError("from_identifier must be an integer or 'latest'")

        cursor = self.conn.cursor()
        if state_change_identifier == 'latest':
            cursor.execute(
                'SELECT identifier FROM state_changes ORDER BY identifier DESC LIMIT 1',
            )
            result = cursor.fetchone()

            if result:
                state_change_identifier = result[0]
            else:
                state_change_identifier = 0

        cursor = self.conn.execute(
            'SELECT statechange_id, data FROM state_snapshot '
            'WHERE statechange_id <= ? '
            'ORDER BY identifier DESC LIMIT 1',
            (state_change_identifier, ),
        )
        rows = cursor.fetchall()

        if rows:
            assert len(rows) == 1, 'LIMIT 1 must return one element'
            last_applied_state_change_id = rows[0][0]
            snapshot_state = rows[0][1]
            result = (last_applied_state_change_id, snapshot_state)
        else:
            result = (0, None)

        return result
def get_snapshot_closest_to_state_change ( self , state_change_identifier : int , ) -> Tuple [ int , Any ] : if not ( state_change_identifier == 'latest' or isinstance ( state_change_identifier , int ) ) : raise ValueError ( "from_identifier must be an integer or 'latest'" ) cursor = self . conn . cursor ( ) if state_change_identifier == 'latest' : cursor . execute ( 'SELECT identifier FROM state_changes ORDER BY identifier DESC LIMIT 1' , ) result = cursor . fetchone ( ) if result : state_change_identifier = result [ 0 ] else : state_change_identifier = 0 cursor = self . conn . execute ( 'SELECT statechange_id, data FROM state_snapshot ' 'WHERE statechange_id <= ? ' 'ORDER BY identifier DESC LIMIT 1' , ( state_change_identifier , ) , ) rows = cursor . fetchall ( ) if rows : assert len ( rows ) == 1 , 'LIMIT 1 must return one element' last_applied_state_change_id = rows [ 0 ] [ 0 ] snapshot_state = rows [ 0 ] [ 1 ] result = ( last_applied_state_change_id , snapshot_state ) else : result = ( 0 , None ) return result
Get snapshots earlier than state_change with provided ID.