crowetic/mev-inspect-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #290 from flashbots/export-deletions

Browse Source 
Support export deletions
This commit is contained in:
Gui Heise 2022-03-22 17:05:32 -04:00 committed by GitHub
commit 22135c42dc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 16 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
mev_inspect/s3_export.py
View File

@ -2,7 +2,6 @@ import itertools
import json import json
import logging import logging
import os import os
from datetime import datetime
from typing import Iterator, Optional, Tuple, TypeVar from typing import Iterator, Optional, Tuple, TypeVar
import boto3 import boto3
@ -36,9 +35,8 @@ def _export_block_by_table(inspect_db_session, block_number: int, table: str) ->
client = get_s3_client() client = get_s3_client()
export_bucket_name = get_export_bucket_name() export_bucket_name = get_export_bucket_name()
export_statement = _get_export_statement(table) export_statement = _get_export_statement(table)
date = round(datetime.utcnow().timestamp())
object_key = f"{table}/flashbots_{block_number}_{date}.json" object_key = f"{table}/flashbots_{block_number}.json"
mev_summary_json_results = inspect_db_session.execute( mev_summary_json_results = inspect_db_session.execute(
statement=export_statement, statement=export_statement,
@ -52,12 +50,21 @@ def _export_block_by_table(inspect_db_session, block_number: int, table: str) ->
existing_object_size = _get_object_size(client, export_bucket_name, object_key) existing_object_size = _get_object_size(client, export_bucket_name, object_key)
if existing_object_size is None or existing_object_size == 0: if existing_object_size is None or existing_object_size == 0:
logger.info(f"Skipping {table} for block {block_number} - no data") logger.info(f"Skipping {table} for block {block_number} - no data")
client.delete_object(
Bucket=export_bucket_name,
Key=object_key,
)
return return
mev_summary_json_fileobj = BytesIteratorIO( mev_summary_json_fileobj = BytesIteratorIO(
(f"{json.dumps(row)}\n".encode("utf-8") for (row,) in mev_summary_json_results) (f"{json.dumps(row)}\n".encode("utf-8") for (row,) in mev_summary_json_results)
) )
client.delete_object(
Bucket=export_bucket_name,
Key=object_key,
)
client.upload_fileobj( client.upload_fileobj(
mev_summary_json_fileobj, mev_summary_json_fileobj,
Bucket=export_bucket_name, Bucket=export_bucket_name,
@ -69,8 +76,12 @@ def _export_block_by_table(inspect_db_session, block_number: int, table: str) ->
def _get_export_statement(table: str) -> str: def _get_export_statement(table: str) -> str:
return f""" return f"""
SELECT to_json({table}) SELECT to_json(json)
FROM (
SELECT *, CURRENT_TIMESTAMP(0) as timestamp
FROM {table} FROM {table}
) json
WHERE WHERE
block_number = :block_number block_number = :block_number
""" """