Add tables
This commit is contained in:
parent
818a0c7ff4
commit
ac376311f4
@ -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]:
|
||||
|
Loading…
x
Reference in New Issue
Block a user