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

Add actor

Browse Source
This commit is contained in:
Gui Heise 2022-02-11 18:43:39 -05:00
parent bb06c8a958
commit 95444eae24
5 changed files with 33 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
Tiltfile
View File

@ -95,21 +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( local_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 = {
"services": "s3",
}))

9
cli.py
View File

@ -13,7 +13,6 @@ 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 inspect_many_blocks_task
from mev_inspect.s3_export import export_block
RPC_URL_ENV = "RPC_URL" RPC_URL_ENV = "RPC_URL"
@ -133,14 +132,6 @@ def fetch_all_prices():
write_prices(inspect_db_session, prices) write_prices(inspect_db_session, prices)
@cli.command()
@click.argument("after_block_number", type=int)
@click.argument("before_block_number", type=int)
def s3_export(block_number: int):
inspect_db_session = get_inspect_session()
export_block(inspect_db_session, 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

@ -3,6 +3,7 @@ import logging
import os import os
import aiohttp import aiohttp
import dramatiq
from mev_inspect.block import get_latest_block_number from mev_inspect.block import get_latest_block_number
from mev_inspect.concurrency import coro from mev_inspect.concurrency import coro
@ -13,6 +14,8 @@ from mev_inspect.crud.latest_block_update import (
from mev_inspect.db import get_inspect_session, get_trace_session 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.provider import get_base_provider from mev_inspect.provider import get_base_provider
from mev_inspect.queue.broker import connect_broker
from mev_inspect.queue.tasks import export_block_task
from mev_inspect.signal_handler import GracefulKiller from mev_inspect.signal_handler import GracefulKiller
logging.basicConfig(filename="listener.log", filemode="a", level=logging.INFO) logging.basicConfig(filename="listener.log", filemode="a", level=logging.INFO)
@ -82,6 +85,11 @@ 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")
export_actor.send(block_number)
if healthcheck_url: if healthcheck_url:
await ping_healthcheck_url(healthcheck_url) await ping_healthcheck_url(healthcheck_url)
else: else:

3
worker.py
View File

@ -10,7 +10,7 @@ from mev_inspect.queue.middleware import (
DbMiddleware, DbMiddleware,
InspectorMiddleware, InspectorMiddleware,
) )
from mev_inspect.queue.tasks import inspect_many_blocks_task from mev_inspect.queue.tasks import export_block_task, inspect_many_blocks_task
logging.basicConfig(stream=sys.stdout, level=logging.INFO) logging.basicConfig(stream=sys.stdout, level=logging.INFO)
@ -21,3 +21,4 @@ broker.add_middleware(InspectorMiddleware(os.environ["RPC_URL"]))
dramatiq.set_broker(broker) dramatiq.set_broker(broker)
dramatiq.actor(inspect_many_blocks_task) dramatiq.actor(inspect_many_blocks_task)
dramatiq.actor(export_block_task)