lib/galaxy/webapps/galaxy/services/history_contents.py
def contents_near(
        self, trans,
        history_id: EncodedDatabaseIdField,
        serialization_params: SerializationParams,
        filter_params: HistoryContentsFilterList,
        hid: int,
        limit: int,
        since: Optional[datetime.datetime] = None,
    ):
        """
        This endpoint provides random access to a large history without having
        to know exactly how many pages are in the final query. Pick a target HID
        and filters, and the endpoint will get LIMIT counts above and below that
        target regardless of how many gaps may exist in the HID due to
        filtering.

        It does 2 queries, one up and one down from the target hid with a
        result size of limit. Additional counts for total matches of both seeks
        provided in the http headers.

        I've also abandoned the wierd q/qv syntax.
        """
        history: History = self.history_manager.get_accessible(
            self.decode_id(history_id), trans.user, current_history=trans.history
        )

        # while polling, check to see if the history has changed
        # if it hasn't then we can short-circuit the poll request
        if since:
            # sqlalchemy DateTime columns are not timezone aware, but `since` may be a timezone aware
            # DateTime object if a timezone offset is provided (https://github.com/samuelcolvin/pydantic/blob/5ccbdcb5904f35834300b01432a665c75dc02296/pydantic/datetime_parse.py#L179).
            # If a timezone is provided (since.tzinfo is not None) we convert to UTC and remove tzinfo so that comparison with history.update_time is correct.
            since = since if since.tzinfo is None else since.astimezone(datetime.timezone.utc).replace(tzinfo=None)
            if history.update_time <= since:
                trans.response.status = 204
                return

        # SEEK UP, contents > hid
        up_params = filter_params + self._hid_greater_than(hid)
        up_order = 'hid-asc'
        contents_up, up_count = self._seek(history, up_params, up_order, limit, serialization_params)

        # SEEK DOWN, contents <= hid
        down_params = filter_params + self._hid_less_or_equal_than(hid)
        down_order = 'hid-dsc'
        contents_down, down_count = self._seek(history, down_params, down_order, limit, serialization_params)

        # min/max hid values
        min_hid, max_hid = self._get_filtered_extrema(history, filter_params)

        # results
        up = self._expand_contents(trans, contents_up, serialization_params)
        up.reverse()
        down = self._expand_contents(trans, contents_down, serialization_params)
        contents = up + down

        # Put stats in http headers
        trans.response.headers['matches_up'] = len(contents_up)
        trans.response.headers['matches_down'] = len(contents_down)
        trans.response.headers['total_matches_up'] = up_count
        trans.response.headers['total_matches_down'] = down_count
        trans.response.headers['max_hid'] = max_hid
        trans.response.headers['min_hid'] = min_hid
        trans.response.headers['history_size'] = str(history.disk_size)
        trans.response.headers['history_empty'] = json.dumps(history.empty)  # convert to proper bool

        return json.dumps(contents)

        This endpoint provides random access to a large history without having
        to know exactly how many pages are in the final query. Pick a target HID
        and filters, and the endpoint will get LIMIT counts above and below that
        target regardless of how many gaps may exist in the HID due to
        filtering.

        It does 2 queries, one up and one down from the target hid with a
        result size of limit. Additional counts for total matches of both seeks
        provided in the http headers.

        I've also abandoned the wierd q/qv syntax.
        
This endpoint provides random access to a large history without having
to know exactly how many pages are in the final query. Pick a target HID
and filters, and the endpoint will get LIMIT counts above and below that
target regardless of how many gaps may exist in the HID due to
filtering.

It does 2 queries, one up and one down from the target hid with a
result size of limit. Additional counts for total matches of both seeks
provided in the http headers.

I've also abandoned the wierd q/qv syntax.
This endpoint provides random access to a large history without having to know exactly how many pages are in the final query . Pick a target HID and filters and the endpoint will get LIMIT counts above and below that target regardless of how many gaps may exist in the HID due to filtering . It does 2 queries one up and one down from the target hid with a result size of limit . Additional counts for total matches of both seeks provided in the http headers . I ' ve also abandoned the wierd q / qv syntax .
def contents_near(
        self, trans,
        history_id: EncodedDatabaseIdField,
        serialization_params: SerializationParams,
        filter_params: HistoryContentsFilterList,
        hid: int,
        limit: int,
        since: Optional[datetime.datetime] = None,
    ):
        history: History = self.history_manager.get_accessible(
            self.decode_id(history_id), trans.user, current_history=trans.history
        )
        if since:
            since = since if since.tzinfo is None else since.astimezone(datetime.timezone.utc).replace(tzinfo=None)
            if history.update_time <= since:
                trans.response.status = 204
                return
        up_params = filter_params + self._hid_greater_than(hid)
        up_order = 'hid-asc'
        contents_up, up_count = self._seek(history, up_params, up_order, limit, serialization_params)
        down_params = filter_params + self._hid_less_or_equal_than(hid)
        down_order = 'hid-dsc'
        contents_down, down_count = self._seek(history, down_params, down_order, limit, serialization_params)
        min_hid, max_hid = self._get_filtered_extrema(history, filter_params)
        up = self._expand_contents(trans, contents_up, serialization_params)
        up.reverse()
        down = self._expand_contents(trans, contents_down, serialization_params)
        contents = up + down
        trans.response.headers['matches_up'] = len(contents_up)
        trans.response.headers['matches_down'] = len(contents_down)
        trans.response.headers['total_matches_up'] = up_count
        trans.response.headers['total_matches_down'] = down_count
        trans.response.headers['max_hid'] = max_hid
        trans.response.headers['min_hid'] = min_hid
        trans.response.headers['history_size'] = str(history.disk_size)
        trans.response.headers['history_empty'] = json.dumps(history.empty)  
        return json.dumps(contents)
def contents_near ( self , trans , history_id : EncodedDatabaseIdField , serialization_params : SerializationParams , filter_params : HistoryContentsFilterList , hid : int , limit : int , since : Optional [ datetime . datetime ] = None , ) : history : History = self . history_manager . get_accessible ( self . decode_id ( history_id ) , trans . user , current_history = trans . history ) if since : since = since if since . tzinfo is None else since . astimezone ( datetime . timezone . utc ) . replace ( tzinfo = None ) if history . update_time <= since : trans . response . status = 204 return up_params = filter_params + self . _hid_greater_than ( hid ) up_order = 'hid-asc' contents_up , up_count = self . _seek ( history , up_params , up_order , limit , serialization_params ) down_params = filter_params + self . _hid_less_or_equal_than ( hid ) down_order = 'hid-dsc' contents_down , down_count = self . _seek ( history , down_params , down_order , limit , serialization_params ) min_hid , max_hid = self . _get_filtered_extrema ( history , filter_params ) up = self . _expand_contents ( trans , contents_up , serialization_params ) up . reverse ( ) down = self . _expand_contents ( trans , contents_down , serialization_params ) contents = up + down trans . response . headers [ 'matches_up' ] = len ( contents_up ) trans . response . headers [ 'matches_down' ] = len ( contents_down ) trans . response . headers [ 'total_matches_up' ] = up_count trans . response . headers [ 'total_matches_down' ] = down_count trans . response . headers [ 'max_hid' ] = max_hid trans . response . headers [ 'min_hid' ] = min_hid trans . response . headers [ 'history_size' ] = str ( history . disk_size ) trans . response . headers [ 'history_empty' ] = json . dumps ( history . empty ) return json . dumps ( contents )
This endpoint provides random access to a large history without having
to know exactly how many pages are in the final query.
This endpoint provides random access to a large history without having to know exactly how many pages are in the final query .
"""
        This endpoint provides random access to a large history without having
        to know exactly how many pages are in the final query. Pick a target HID
        and filters, and the endpoint will get LIMIT counts above and below that
        target regardless of how many gaps may exist in the HID due to
        filtering.

        It does 2 queries, one up and one down from the target hid with a
        result size of limit. Additional counts for total matches of both seeks
        provided in the http headers.

        I've also abandoned the wierd q/qv syntax.
        """ # while polling, check to see if the history has changed # if it hasn't then we can short-circuit the poll request # sqlalchemy DateTime columns are not timezone aware, but `since` may be a timezone aware # DateTime object if a timezone offset is provided (https://github.com/samuelcolvin/pydantic/blob/5ccbdcb5904f35834300b01432a665c75dc02296/pydantic/datetime_parse.py#L179). # If a timezone is provided (since.tzinfo is not None) we convert to UTC and remove tzinfo so that comparison with history.update_time is correct. # SEEK UP, contents > hid # SEEK DOWN, contents <= hid # min/max hid values # results # Put stats in http headers # convert to proper bool
self
trans
history_id
EncodedDatabaseIdField
serialization_params
SerializationParams
filter_params
HistoryContentsFilterList
hid
int
limit
int
since
Optional[datetime.datetime]