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

Export command and function edits

Browse Source
This commit is contained in:
Gui Heise 2022-02-14 12:37:52 -05:00
parent 95444eae24
commit c6f7fd509e
3 changed files with 35 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
Tiltfile
View File

@ -95,25 +95,25 @@ local_resource(
) )
# if using local S3 exports # if using local S3 exports
k8s_yaml(secret_from_dict("mev-inspect-export", inputs = { #k8s_yaml(secret_from_dict("mev-inspect-export", inputs = {
"export-bucket-name" : "local-export", # "export-bucket-name" : "local-export",
"export-bucket-region": "us-east-1", # "export-bucket-region": "us-east-1",
"export-aws-access-key-id": "foobar", # "export-aws-access-key-id": "foobar",
"export-aws-secret-access-key": "foobar", # "export-aws-secret-access-key": "foobar",
})) #}))
helm_remote( #helm_remote(
"localstack", # "localstack",
repo_name="localstack-charts", # repo_name="localstack-charts",
repo_url="https://localstack.github.io/helm-charts", # repo_url="https://localstack.github.io/helm-charts",
) #)
local_resource( l#ocal_resource(
'localstack-port-forward', # 'localstack-port-forward',
serve_cmd='kubectl port-forward --namespace default svc/localstack 4566:4566', # serve_cmd='kubectl port-forward --namespace default svc/localstack 4566:4566',
resource_deps=["localstack"] # resource_deps=["localstack"]
) #)
k8s_yaml(configmap_from_dict("mev-inspect-export", inputs = { #k8s_yaml(configmap_from_dict("mev-inspect-export", inputs = {
"services": "s3", # "services": "s3",
})) #}))

11
cli.py
View File

@ -12,7 +12,7 @@ from mev_inspect.db import get_inspect_session, get_trace_session
from mev_inspect.inspector import MEVInspector from mev_inspect.inspector import MEVInspector
from mev_inspect.prices import fetch_prices, fetch_prices_range from mev_inspect.prices import fetch_prices, fetch_prices_range
from mev_inspect.queue.broker import connect_broker from mev_inspect.queue.broker import connect_broker
from mev_inspect.queue.tasks import inspect_many_blocks_task from mev_inspect.queue.tasks import export_block_task, inspect_many_blocks_task
RPC_URL_ENV = "RPC_URL" RPC_URL_ENV = "RPC_URL"
@ -132,6 +132,15 @@ def fetch_all_prices():
write_prices(inspect_db_session, prices) write_prices(inspect_db_session, prices)
@cli.command()
@click.argument("block_number", type=int)
def s3_export(block_number: int):
broker = connect_broker()
export_actor = dramatiq.actor(export_block_task, broker=broker)
logger.info(f"Sending block {block_number} for export")
export_actor.send(block_number)
@cli.command() @cli.command()
@click.argument("after", type=click.DateTime(formats=["%Y-%m-%d", "%m-%d-%Y"])) @click.argument("after", type=click.DateTime(formats=["%Y-%m-%d", "%m-%d-%Y"]))
@click.argument("before", type=click.DateTime(formats=["%Y-%m-%d", "%m-%d-%Y"])) @click.argument("before", type=click.DateTime(formats=["%Y-%m-%d", "%m-%d-%Y"]))

8
listener.py
View File

@ -40,6 +40,9 @@ async def run():
inspect_db_session = get_inspect_session() inspect_db_session = get_inspect_session()
trace_db_session = get_trace_session() trace_db_session = get_trace_session()
broker = connect_broker()
export_actor = dramatiq.actor(export_block_task, broker=broker)
inspector = MEVInspector(rpc) inspector = MEVInspector(rpc)
base_provider = get_base_provider(rpc) base_provider = get_base_provider(rpc)
@ -50,6 +53,7 @@ async def run():
trace_db_session, trace_db_session,
base_provider, base_provider,
healthcheck_url, healthcheck_url,
export_actor,
) )
logger.info("Stopping...") logger.info("Stopping...")
@ -61,7 +65,9 @@ async def inspect_next_block(
trace_db_session, trace_db_session,
base_provider, base_provider,
healthcheck_url, healthcheck_url,
export_actor,
): ):
latest_block_number = await get_latest_block_number(base_provider) latest_block_number = await get_latest_block_number(base_provider)
last_written_block = find_latest_block_update(inspect_db_session) last_written_block = find_latest_block_update(inspect_db_session)
@ -85,8 +91,6 @@ async def inspect_next_block(
update_latest_block(inspect_db_session, block_number) update_latest_block(inspect_db_session, block_number)
broker = connect_broker()
export_actor = dramatiq.actor(export_block_task, broker=broker)
logger.info(f"Sending block {block_number} for export") logger.info(f"Sending block {block_number} for export")
export_actor.send(block_number) export_actor.send(block_number)