temporarily disable streaming responses

Author: OriHoch

open_bus_stride_api/common/sql_route.py

@@ -31,24 +31,27 @@ def list_(sql, sql_params, default_limit, limit, offset, get_count, order_by, sk
                 sql_order_by = ' order by ' + ', '.join([f'{fieldname} {direction}' for direction, fieldname in order_by_args])
             if limit is not None:
                 sql_limit = f' limit {limit}'
+            assert sql_limit and 0 < limit <= 1000, "due to abuse, limit must be between 1 and 1000, contact us if you need more"
             if offset is not None:
                 sql_offset = f' offset {offset}'
             if sql_order_by or sql_limit or sql_offset:
                 sql = f'select * from ({sql}) a {sql_order_by}{sql_limit}{sql_offset}'
-            iterator = (o for o in session.execute(sql, sql_params, execution_options={'stream_results': True}))
-            first_items = list(itertools.islice(iterator, QUERY_PAGE_SIZE + 1))
-            if len(first_items) <= QUERY_PAGE_SIZE:
-                common.debug_print(f'got {len(first_items)} items - returning without streaming')
-                data = [common.post_process_response_obj(obj, None) for obj in first_items]
-                session.close()
-                return data
-            else:
-                raise Exception("streaming responses disabled due to abuse, if you need this feature please contact us")
-                common.debug_print(f'got {len(first_items)} items - returning using streaming')
-                return fastapi.responses.StreamingResponse(
-                    common.streaming_response_iterator(session, first_items, iterator, None),
-                    media_type="application/json"
-                )
+            data = [common.post_process_response_obj(obj, None) for obj in session.execute(sql, sql_params)]
+            session.close()
+            return data
+            # iterator = (o for o in session.execute(sql, sql_params, execution_options={'stream_results': True}))
+            # first_items = list(itertools.islice(iterator, QUERY_PAGE_SIZE + 1))
+            # if len(first_items) <= QUERY_PAGE_SIZE:
+            #     common.debug_print(f'got {len(first_items)} items - returning without streaming')
+            #     data = [common.post_process_response_obj(obj, None) for obj in first_items]
+            #     session.close()
+            #     return data
+            # else:
+            #     common.debug_print(f'got {len(first_items)} items - returning using streaming')
+            #     return fastapi.responses.StreamingResponse(
+            #         common.streaming_response_iterator(session, first_items, iterator, None),
+            #         media_type="application/json"
+            #     )
     except:
         session.close()
         raise

open_bus_stride_api/routers/common.py

@@ -81,23 +81,25 @@ def get_list(*args, convert_to_dict=None, **kwargs):
             return fastapi.Response(content=str(q_count), media_type="application/json")
         else:
             debug_print(f'Getting results for query: {q}')
-            if not hasattr(q, '__q_limit') or not q.__q_limit or q.__q_limit > QUERY_PAGE_SIZE:
-                debug_print(f'adding yield_per({QUERY_PAGE_SIZE}) to query')
-                q = q.yield_per(QUERY_PAGE_SIZE)
-            q_iterator = (obj for obj in q)
-            first_items = list(itertools.islice(q_iterator, QUERY_PAGE_SIZE + 1))
-            if len(first_items) <= QUERY_PAGE_SIZE:
-                debug_print(f'got {len(first_items)} items - returning without streaming')
-                data = [post_process_response_obj(obj, convert_to_dict) for obj in first_items]
-                session.close()
-                return data
-            else:
-                raise Exception("streaming responses disabled due to abuse, if you need this feature please contact us")
-                debug_print(f'got {len(first_items)} items - returning using streaming')
-                return fastapi.responses.StreamingResponse(
-                    streaming_response_iterator(session, first_items, q_iterator, convert_to_dict),
-                    media_type="application/json"
-                )
+            data = [post_process_response_obj(obj, convert_to_dict) for obj in q]
+            session.close()
+            return data
+            # if not hasattr(q, '__q_limit') or not q.__q_limit or q.__q_limit > QUERY_PAGE_SIZE:
+            #     debug_print(f'adding yield_per({QUERY_PAGE_SIZE}) to query')
+            #     q = q.yield_per(QUERY_PAGE_SIZE)
+            # q_iterator = (obj for obj in q)
+            # first_items = list(itertools.islice(q_iterator, QUERY_PAGE_SIZE + 1))
+            # if len(first_items) <= QUERY_PAGE_SIZE:
+            #     debug_print(f'got {len(first_items)} items - returning without streaming')
+            #     data = [post_process_response_obj(obj, convert_to_dict) for obj in first_items]
+            #     session.close()
+            #     return data
+            # else:
+            #     debug_print(f'got {len(first_items)} items - returning using streaming')
+            #     return fastapi.responses.StreamingResponse(
+            #         streaming_response_iterator(session, first_items, q_iterator, convert_to_dict),
+            #         media_type="application/json"
+            #     )
     except:
         session.close()
         raise
@@ -166,6 +168,7 @@ def get_list_query(session, db_model, limit, offset, filters=None, default_limit
         limit = default_limit
     if limit:
         limit = int(limit)
+    assert 0 < limit <= 1000, "due to abuse, maximum limit per request is 1000 items, contact us if you need more"
     if filters is None:
         filters = []
     if get_base_session_query_callback is None: