<s> import base64 <EOL> import flask <EOL> import json <EOL> from marshmallow import ValidationError <EOL> import sqlalchemy as sa <EOL> from sqlalchemy import Column , sql <EOL> from . exceptions import ApiError <EOL> from . import meta <EOL> from . import utils <EOL> class LimitPagination ( object ) : <EOL> limit_arg = '<STR_LIT>' <EOL> def __init__ ( self , default_limit = None , max_limit = None ) : <EOL> self . _default_limit = utils . if_none ( default_limit , max_limit ) <EOL> self . _max_limit = max_limit <EOL> if self . _max_limit is not None : <EOL> assert self . _default_limit <= self . _max_limit , "<STR_LIT>" <EOL> def __call__ ( self , query , view ) : <EOL> limit = self . get_limit ( ) <EOL> if limit is not None : <EOL> query = query . limit ( limit + <NUM_LIT:1> ) <EOL> items = query . all ( ) <EOL> if limit is not None and len ( items ) > limit : <EOL> has_next_page = True <EOL> items = items [ : limit ] <EOL> else : <EOL> has_next_page = False <EOL> meta . set_response_meta ( has_next_page = has_next_page ) <EOL> return items <EOL> def get_limit ( self ) : <EOL> limit = flask . request . args . get ( self . limit_arg ) <EOL> try : <EOL> return self . parse_limit ( limit ) <EOL> except ApiError as e : <EOL> raise e . update ( { '<STR_LIT:source>' : { '<STR_LIT>' : self . limit_arg } } ) <EOL> def parse_limit ( self , limit ) : <EOL> if limit is None : <EOL> return self . _default_limit <EOL> try : <EOL> limit = int ( limit ) <EOL> except ValueError : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> if limit < <NUM_LIT:0> : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> if self . _max_limit is not None : <EOL> limit = min ( limit , self . _max_limit ) <EOL> return limit <EOL> def spec_declaration ( self , path , spec , ** kwargs ) : <EOL> path [ '<STR_LIT>' ] . add_parameter ( name = '<STR_LIT>' , <EOL> type = '<STR_LIT:int>' , <EOL> description = '<STR_LIT>' ) <EOL> path [ '<STR_LIT>' ] . add_property_to_response ( <EOL> prop_name = '<STR_LIT>' , <EOL> type = '<STR_LIT:object>' , <EOL> properties = { <EOL> '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' } <EOL> } <EOL> ) <EOL> class LimitOffsetPagination ( LimitPagination ) : <EOL> offset_arg = '<STR_LIT>' <EOL> def __call__ ( self , query , view ) : <EOL> offset = self . get_offset ( ) <EOL> query = query . offset ( offset ) <EOL> return super ( LimitOffsetPagination , self ) . __call__ ( query , view ) <EOL> def get_offset ( self ) : <EOL> offset = flask . request . args . get ( self . offset_arg ) <EOL> try : <EOL> return self . parse_offset ( offset ) <EOL> except ApiError as e : <EOL> raise e . update ( { '<STR_LIT:source>' : { '<STR_LIT>' : self . offset_arg } } ) <EOL> def parse_offset ( self , offset ) : <EOL> if offset is None : <EOL> return <NUM_LIT:0> <EOL> try : <EOL> offset = int ( offset ) <EOL> except ValueError : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> if offset < <NUM_LIT:0> : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> return offset <EOL> def spec_declaration ( self , path , spec , ** kwargs ) : <EOL> super ( LimitOffsetPagination , self ) . spec_declaration ( path , spec ) <EOL> path [ '<STR_LIT>' ] . add_parameter ( <EOL> name = '<STR_LIT>' , <EOL> type = '<STR_LIT:int>' , <EOL> description = '<STR_LIT>' ) <EOL> class PagePagination ( LimitOffsetPagination ) : <EOL> page_arg = '<STR_LIT>' <EOL> def __init__ ( self , page_size ) : <EOL> super ( PagePagination , self ) . __init__ ( ) <EOL> self . _page_size = page_size <EOL> def get_offset ( self ) : <EOL> return self . get_page ( ) * self . _page_size <EOL> def get_page ( self ) : <EOL> page = flask . request . args . get ( self . page_arg ) <EOL> try : <EOL> return self . parse_page ( page ) <EOL> except ApiError as e : <EOL> raise e . update ( { '<STR_LIT:source>' : { '<STR_LIT>' : self . page_arg } } ) <EOL> def parse_page ( self , page ) : <EOL> if page is None : <EOL> return <NUM_LIT:0> <EOL> try : <EOL> page = int ( page ) <EOL> except ValueError : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> if page < <NUM_LIT:0> : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> return page <EOL> def get_limit ( self ) : <EOL> return self . _page_size <EOL> def spec_declaration ( self , path , spec , ** kwargs ) : <EOL> super ( PagePagination , self ) . spec_declaration ( path , spec ) <EOL> path [ '<STR_LIT>' ] . add_parameter ( <EOL> name = '<STR_LIT>' , <EOL> type = '<STR_LIT:int>' , <EOL> description = '<STR_LIT>' ) <EOL> class IdCursorPagination ( LimitPagination ) : <EOL> cursor_arg = '<STR_LIT>' <EOL> def __call__ ( self , query , view ) : <EOL> column_specs = self . get_column_specs ( query , view ) <EOL> cursor_in = self . get_cursor ( view , column_specs ) <EOL> if cursor_in is not None : <EOL> query = query . filter ( self . get_filter ( column_specs , cursor_in ) ) <EOL> items = super ( IdCursorPagination , self ) . __call__ ( query , view ) <EOL> cursors_out = self . render_cursors ( view , column_specs , items ) <EOL> meta . set_response_meta ( cursors = cursors_out ) <EOL> return items <EOL> def get_column_specs ( self , query , view ) : <EOL> column_specs = tuple ( <EOL> self . get_column_spec ( expression ) for expression in query . _order_by <EOL> ) <EOL> for id_field in view . id_fields : <EOL> id_column = view . model . __table__ . c [ id_field ] <EOL> assert id_column in ( column for column , _ in column_specs ) , "<STR_LIT>" . format ( id_field ) <EOL> return column_specs <EOL> def get_column_spec ( self , expression ) : <EOL> if isinstance ( expression , Column ) : <EOL> return expression , True <EOL> column = expression . element <EOL> assert isinstance ( column , Column ) , "<STR_LIT>" <EOL> modifier = expression . modifier <EOL> assert modifier in ( sql . operators . asc_op , sql . operators . desc_op ) <EOL> asc = modifier == sql . operators . asc_op <EOL> return column , asc <EOL> def get_cursor ( self , view , column_specs ) : <EOL> cursor = flask . request . args . get ( self . cursor_arg ) <EOL> try : <EOL> return self . parse_cursor ( view , column_specs , cursor ) <EOL> except ApiError as e : <EOL> raise e . update ( { '<STR_LIT:source>' : { '<STR_LIT>' : self . cursor_arg } } ) <EOL> def parse_cursor ( self , view , column_specs , cursor ) : <EOL> if not cursor : <EOL> return None <EOL> cursor = self . decode_cursor ( cursor ) <EOL> if len ( cursor ) != len ( column_specs ) : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> deserializer = view . deserializer <EOL> column_fields = ( <EOL> deserializer . fields [ column . name ] for column , _ in column_specs <EOL> ) <EOL> try : <EOL> cursor = tuple ( <EOL> field . deserialize ( value ) <EOL> for field , value in zip ( column_fields , cursor ) <EOL> ) <EOL> except ValidationError as e : <EOL> errors = ( <EOL> self . format_validation_error ( message ) <EOL> for message , path in utils . iter_validation_errors ( e . messages ) <EOL> ) <EOL> raise ApiError ( <NUM_LIT> , * errors ) <EOL> return cursor <EOL> def format_validation_error ( self , message ) : <EOL> return { <EOL> '<STR_LIT:code>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : message , <EOL> } <EOL> def decode_cursor ( self , cursor ) : <EOL> try : <EOL> cursor = cursor . encode ( '<STR_LIT:ascii>' ) <EOL> cursor = base64 . urlsafe_b64decode ( cursor ) <EOL> cursor = cursor . decode ( '<STR_LIT:utf-8>' ) <EOL> cursor = json . loads ( cursor ) <EOL> except ( TypeError , ValueError ) : <EOL> raise ApiError ( <NUM_LIT> , { '<STR_LIT:code>' : '<STR_LIT>' } ) <EOL> return cursor <EOL> def get_filter ( self , column_specs , cursor ) : <EOL> filter_specs = tuple ( zip ( column_specs , cursor ) ) <EOL> return sa . or_ ( <EOL> self . get_filter_clause ( filter_specs [ : i + <NUM_LIT:1> ] ) <EOL> for i in range ( len ( filter_specs ) ) <EOL> ) <EOL> def get_filter_clause ( self , filter_specs ) : <EOL> previous_clauses = sa . and_ ( <EOL> column == value for ( column , _ ) , value in filter_specs [ : - <NUM_LIT:1> ] <EOL> ) <EOL> ( column , asc ) , value = filter_specs [ - <NUM_LIT:1> ] <EOL> if asc : <EOL> current_clause = column > value <EOL> else : <EOL> current_clause = column < value <EOL> return sa . and_ ( previous_clauses , current_clause ) <EOL> def render_cursors ( self , view , column_specs , items ) : <EOL> serializer = view . serializer <EOL> column_fields = tuple ( <EOL> serializer . fields [ column . name ] for column , _ in column_specs <EOL> ) <EOL> return tuple ( <EOL> self . render_cursor ( item , column_fields ) for item in items <EOL> ) <EOL> def render_cursor ( self , item , column_fields ) : <EOL> cursor = tuple ( <EOL> field . _serialize ( getattr ( item , field . name ) , field . name , item ) <EOL> for field in column_fields <EOL> ) <EOL> return self . encode_cursor ( cursor ) <EOL> def encode_cursor ( self , cursor ) : <EOL> cursor = json . dumps ( cursor ) <EOL> cursor = cursor . encode ( '<STR_LIT:utf-8>' ) <EOL> cursor = base64 . urlsafe_b64encode ( cursor ) <EOL> return cursor . decode ( '<STR_LIT:ascii>' ) <EOL> def spec_declaration ( self , path , spec , ** kwargs ) : <EOL> super ( IdCursorPagination , self ) . spec_declaration ( path , spec ) <EOL> path [ '<STR_LIT>' ] . add_parameter ( <EOL> name = '<STR_LIT>' , <EOL> type = '<STR_LIT:string>' , <EOL> description = '<STR_LIT>' ) </s>
