diff --git a/mev_inspect/s3_export.py b/mev_inspect/s3_export.py index 39da5f9..bdceccf 100644 --- a/mev_inspect/s3_export.py +++ b/mev_inspect/s3_export.py @@ -21,39 +21,71 @@ WHERE block_number = :block_number """ +ARBITRAGES_EXPORT_QUERY = """ + SELECT to_json(arbitrages) + FROM arbitrages +WHERE + block_number = :block_number + """ +LIQUIDATIONS_EXPORT_QUERY = """ + SELECT to_json(liquidations) + FROM liquidations +WHERE + block_number = :block_number + """ +SANDWICHES_EXPORT_QUERY = """ + SELECT to_json(sandwiches) + FROM sandwiches +WHERE + block_number = :block_number + """ + +query_by_table = { + "mev_summary": MEV_SUMMARY_EXPORT_QUERY, + "arbitrages": ARBITRAGES_EXPORT_QUERY, + "liquidations": LIQUIDATIONS_EXPORT_QUERY, + "sandwiches": SANDWICHES_EXPORT_QUERY, +} + logger = logging.getLogger(__name__) def export_block(inspect_db_session, block_number: int) -> None: export_bucket_name = get_export_bucket_name() client = get_s3_client() - object_key = f"mev_summary/flashbots_{block_number}.json" - mev_summary_json_results = inspect_db_session.execute( - statement=MEV_SUMMARY_EXPORT_QUERY, - params={ - "block_number": block_number, - }, - ) + for table in query_by_table.keys(): + object_key = f"{table}/flashbots_{block_number}.json" + mev_summary_json_results = inspect_db_session.execute( + statement=query_by_table[table], + params={ + "block_number": block_number, + }, + ) - first_value, mev_summary_json_results = _peek(mev_summary_json_results) - if first_value is None: - existing_object_size = _get_object_size(client, export_bucket_name, object_key) - if existing_object_size is None or existing_object_size == 0: - logger.info(f"Skipping block {block_number} - no data") - return + first_value, mev_summary_json_results = _peek(mev_summary_json_results) + if first_value is None: + existing_object_size = _get_object_size( + client, export_bucket_name, object_key + ) + if existing_object_size is None or existing_object_size == 0: + logger.info(f"Skipping {table} for block {block_number} - no data") + continue - mev_summary_json_fileobj = BytesIteratorIO( - (f"{json.dumps(row)}\n".encode("utf-8") for (row,) in mev_summary_json_results) - ) + mev_summary_json_fileobj = BytesIteratorIO( + ( + f"{json.dumps(row)}\n".encode("utf-8") + for (row,) in mev_summary_json_results + ) + ) - client.upload_fileobj( - mev_summary_json_fileobj, - Bucket=export_bucket_name, - Key=object_key, - ) + client.upload_fileobj( + mev_summary_json_fileobj, + Bucket=export_bucket_name, + Key=object_key, + ) - logger.info(f"Exported to {object_key}") + logger.info(f"Exported to {object_key}") def _get_object_size(client, bucket: str, key: str) -> Optional[int]: