push dune sync project

.env.sample

@@ -0,0 +1,4 @@
+DUNE_API_KEY=
+AWS_ROLE=
+APP_DATA_BUCKET=
+VOLUME_PATH=

.pylintrc

@@ -0,0 +1,2 @@
+[MASTER]
+disable=fixme,logging-fstring-interpolation

Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.10
+
+WORKDIR /app
+
+COPY requirements/* requirements/
+RUN pip install -r requirements/prod.txt
+COPY ./src ./src
+
+ENTRYPOINT [ "python3", "-m" , "src.main"]

Makefile

@@ -0,0 +1,34 @@
+VENV = venv
+PYTHON = $(VENV)/bin/python3
+PIP = $(VENV)/bin/pip
+PROJECT_ROOT = src
+
+
+$(VENV)/bin/activate: requirements/dev.txt
+	python3 -m venv $(VENV)
+	$(PIP) install --upgrade pip
+	$(PIP) install -r requirements/dev.txt
+
+
+install:
+	make $(VENV)/bin/activate
+
+clean:
+	rm -rf __pycache__
+
+fmt:
+	black ./
+
+lint:
+	pylint ${PROJECT_ROOT}/
+
+types:
+	mypy ${PROJECT_ROOT}/ --strict
+
+check:
+	make fmt
+	make lint
+	make types
+
+test:
+	python -m pytest tests

README.md

@@ -1,2 +1,30 @@
 # dune-sync
 Components for syncing off-chain data with Dune Community Sources
+
+
+# Local Development
+
+
+1. clone repo
+2. Several Makefile commands:
+```shell
+make install
+```
+```shell
+make check # (runs black, pylint and mypy --strict)
+```
+```shell
+make test # Runs all tests
+```
+
+## Docker
+### Build
+```shell
+docker build -t local_dune_sync .
+```
+
+You must provide valid environment variables as specified in [.env.sample](.env.sample)
+### Run
+```shell
+docker run -v ${PWD}data:/app/data --env-file .env local_dune_sync
+```

mypy.ini

@@ -0,0 +1,2 @@
+[mypy-src.sync]
+implicit_reexport = True

requirements.txt

@@ -0,0 +1,18 @@
+# Prod
+dune-client==0.2.2
+psycopg2-binary==2.9.3
+python-dotenv>=0.20.0
+requests>=2.28.1
+pandas==1.5.0
+ndjson>=0.3.1
+py-multiformats-cid>=0.4.3
+boto3==1.26.12
+
+
+# Dev
+pandas-stubs==1.5.1.221024
+boto3-stubs==1.26.12
+black==22.6.0
+mypy==0.982
+pylint==2.14.4
+pytest==7.1.2

requirements/dev.txt

@@ -0,0 +1,7 @@
+-r prod.txt
+pandas-stubs==1.5.1.221024
+boto3-stubs==1.26.12
+black==22.6.0
+mypy==0.982
+pylint==2.14.4
+pytest==7.1.2

requirements/prod.txt

@@ -0,0 +1,8 @@
+dune-client==0.2.2
+psycopg2-binary==2.9.3
+python-dotenv>=0.20.0
+requests>=2.28.1
+pandas==1.5.0
+ndjson>=0.3.1
+py-multiformats-cid>=0.4.3
+boto3==1.26.12

src/dune_queries.py

@@ -0,0 +1,44 @@
+"""
+Localized account of all Queries related to this project's main functionality
+"""
+from __future__ import annotations
+
+from copy import copy
+from dataclasses import dataclass
+
+from dune_client.query import Query
+from dune_client.types import QueryParameter
+
+
+@dataclass
+class QueryData:
+    """Stores name and a version of the query for each query."""
+
+    name: str
+    query: Query
+
+    def __init__(self, name: str, query_id: int, filename: str) -> None:
+        self.name = name
+        self.filepath = filename
+        self.query = Query(query_id, name)
+
+    def with_params(self, params: list[QueryParameter]) -> Query:
+        """
+        Copies the query and adds parameters to it, returning the copy.
+        """
+        # We currently default to the V1 Queries, soon to switch them out.
+        query_copy = copy(self.query)
+        query_copy.params = params
+        return query_copy
+
+
+QUERIES = {
+    "APP_HASHES": QueryData(
+        query_id=1610025, name="Unique App Hashes", filename="app_hashes.sql"
+    ),
+    "LATEST_APP_HASH_BLOCK": QueryData(
+        query_id=1615490,
+        name="Latest Possible App Hash Block",
+        filename="app_hash_latest_block.sql",
+    ),
+}

src/environment.py

@@ -0,0 +1,7 @@
+# """
+# Collection of runtime constants
+# """
+# from pathlib import Path
+#
+# PROJECT_ROOT = Path(__file__).parent.parent
+# OUT_DIR = PROJECT_ROOT / Path("data")

src/fetch/dune.py

@@ -0,0 +1,90 @@
+"""
+All Dune Query executions should be routed through this file.
+TODO - Move reusable components into dune-client:
+    https://github.com/cowprotocol/dune-bridge/issues/40
+"""
+import asyncio
+import logging
+import sys
+
+from requests import HTTPError
+
+from dune_client.client import DuneClient
+from dune_client.query import Query
+from dune_client.types import DuneRecord
+
+from src.dune_queries import QUERIES
+from src.models.block_range import BlockRange
+
+
+class DuneFetcher:
+    """
+    Class containing, DuneClient, FileIO and a logger for convenient Dune Fetching.
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+    ) -> None:
+        """
+        Class constructor.
+        Builds DuneClient from `api_key` along with a logger and FileIO object.
+        """
+        # It's a bit weird that the DuneClient also declares a log like this,
+        # but it also doesn't make sense to inherit that log. Not sure what's best practise here.
+        self.log = logging.getLogger(__name__)
+        logging.basicConfig(format="%(asctime)s %(levelname)s %(name)s %(message)s")
+        # TODO - use runtime parameter. https://github.com/cowprotocol/dune-bridge/issues/41
+        self.log.setLevel(logging.DEBUG)
+        self.dune = DuneClient(api_key)
+
+    async def fetch(self, query: Query) -> list[DuneRecord]:
+        """Async Dune Fetcher with some exception handling."""
+        self.log.debug(f"Executing {query}")
+
+        try:
+            # Tried to use the AsyncDuneClient, without success:
+            # https://github.com/cowprotocol/dune-client/pull/31#issuecomment-1316045313
+            response = await asyncio.to_thread(
+                self.dune.refresh, query, ping_frequency=10
+            )
+            if response.state.is_complete():
+                response_rows = response.get_rows()
+                self.log.debug(
+                    f"Got {len(response_rows)} results for execution {response.execution_id}"
+                )
+                return response_rows
+
+            message = (
+                f"query execution {response.execution_id} incomplete {response.state}"
+            )
+            self.log.error(message)
+            raise RuntimeError(f"no results for {message}")
+        except HTTPError as err:
+            self.log.error(f"Got {err} - Exiting")
+            sys.exit()
+
+    async def latest_app_hash_block(self) -> int:
+        """
+        Block Range is used to app hash fetcher where to find the new records.
+        block_from: read from file `fname` as a loaded singleton.
+            - uses genesis block is no file exists (should only ever happen once)
+            - raises RuntimeError if column specified does not exist.
+        block_to: fetched from Dune as the last indexed block for "GPv2Settlement_call_settle"
+        """
+        return int(
+            # KeyError here means the query has been modified and column no longer exists
+            # IndexError means no results were returned from query (which is unlikely).
+            (await self.fetch(QUERIES["LATEST_APP_HASH_BLOCK"].query))[0][
+                "latest_block"
+            ]
+        )
+
+    async def get_app_hashes(self, block_range: BlockRange) -> list[DuneRecord]:
+        """
+        Executes APP_HASHES query for the given `block_range` and returns the results
+        """
+        app_hash_query = QUERIES["APP_HASHES"].with_params(
+            block_range.as_query_params()
+        )
+        return await self.fetch(app_hash_query)

src/fetch/ipfs.py

@@ -0,0 +1,51 @@
+"""IPFS CID (de)serialization"""
+from __future__ import annotations
+
+from typing import Any, Optional
+
+import requests
+from multiformats_cid.cid import from_bytes  # type: ignore
+
+
+class Cid:
+    """Holds logic for constructing and converting various representations of a Delegation ID"""
+
+    def __init__(self, hex_str: str) -> None:
+        """Builds Object (bytes as base representation) from hex string."""
+        stripped_hex = hex_str.replace("0x", "")
+        # Anatomy of a CID: https://proto.school/anatomy-of-a-cid/04
+        prefix = bytearray([1, 112, 18, 32])
+        self.bytes = bytes(prefix + bytes.fromhex(stripped_hex))
+
+    @property
+    def hex(self) -> str:
+        """Returns hex representation"""
+        without_prefix = self.bytes[4:]
+        return "0x" + without_prefix.hex()
+
+    def __str__(self) -> str:
+        """Returns string representation"""
+        return str(from_bytes(self.bytes))
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, Cid):
+            return False
+        return self.bytes == other.bytes
+
+    def url(self) -> str:
+        """IPFS URL where content can be recovered"""
+        return f"https://gnosis.mypinata.cloud/ipfs/{self}"
+
+    def get_content(self, max_retries: int = 3) -> Optional[Any]:
+        """
+        Attempts to fetch content at cid with a timeout of 1 second.
+        Trys `max_retries` times and otherwise returns None`
+        """
+        attempts = 0
+        while attempts < max_retries:
+            try:
+                response = requests.get(self.url(), timeout=1)
+                return response.json()
+            except requests.exceptions.ReadTimeout:
+                attempts += 1
+        return None

src/main.py

@@ -0,0 +1,32 @@
+"""Main Entry point for app_hash sync"""
+import asyncio
+import logging.config
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+from src.sync import sync_app_data
+from src.fetch.dune import DuneFetcher
+from src.sync.config import AppDataSyncConfig
+
+log = logging.getLogger(__name__)
+logging.basicConfig(format="%(asctime)s %(levelname)s %(name)s %(message)s")
+log.setLevel(logging.DEBUG)
+
+
+GIVE_UP_THRESHOLD = 10
+
+
+if __name__ == "__main__":
+    load_dotenv()
+    asyncio.run(
+        sync_app_data(
+            dune=DuneFetcher(os.environ["DUNE_API_KEY"]),
+            config=AppDataSyncConfig(
+                aws_role=os.environ["AWS_ROLE"],
+                aws_bucket=os.environ["AWS_BUCKET"],
+                volume_path=Path(os.environ["VOLUME_PATH"]).absolute(),
+            ),
+        )
+    )

src/models/block_range.py

@@ -0,0 +1,28 @@
+"""
+BlockRange Model is just a data class for left and right bounds
+"""
+from dataclasses import dataclass
+
+from dune_client.types import QueryParameter
+
+
+@dataclass
+class BlockRange:
+    """
+    Basic dataclass for an Ethereum block range with some Dune compatibility methods.
+    TODO (easy) - this data class could probably live in dune-client.
+    https://github.com/cowprotocol/dune-bridge/issues/40
+    """
+
+    block_from: int
+    block_to: int
+
+    def __str__(self) -> str:
+        return f"({self.block_from}, {self.block_to})"
+
+    def as_query_params(self) -> list[QueryParameter]:
+        """Returns self as Dune QueryParameters"""
+        return [
+            QueryParameter.number_type("BlockFrom", self.block_from),
+            QueryParameter.number_type("BlockTo", self.block_to),
+        ]