Split prefect deployments into dev and prod

.github/workflows/prefect-dev.yaml

@@ -0,0 +1,27 @@
+name: prefect-dev
+on:
+  push:
+    branches: [ dev ]
+jobs:
+  dev-deployment:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.12"]
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Deploy prefect deployment
+        env:
+          PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements/prefect.txt
+          pip install -r requirements/prod.txt
+          prefect config set PREFECT_API_URL=$PREFECT_API_URL
+          python -m src.deploy_prefect.dev_deployment

.github/workflows/prefect.yaml → .github/workflows/prefect-prod.yaml

@@ -1,9 +1,9 @@
-name: prefect
+name: prefect-prod
 on:
   push:
     branches: [ main ]
 jobs:
-  deployment:
+  prod-deployment:
     runs-on: ubuntu-latest
     strategy:
       matrix:
@@ -24,4 +24,4 @@ jobs:
           pip install -r requirements/prefect.txt
           pip install -r requirements/prod.txt
           prefect config set PREFECT_API_URL=$PREFECT_API_URL
-          python -m src.deploy_prefect.deployment
+          python -m src.deploy_prefect.prod_deployment

.github/workflows/pull-request.yaml

@@ -49,3 +49,27 @@ jobs:
         env:
           PROD_DB_URL: ${{ secrets.PROD_DB_URL }}
           BARN_DB_URL: ${{ secrets.BARN_DB_URL }}
+
+  prefect-pr-test-run:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.12"]
+
+    env:
+      PREFECT_API_URL: ${{ secrets.PREFECT_API_URL }}
+      BRANCH_NAME: ${{ github.head_ref }}
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements/prefect.txt
+          pip install -r requirements/prod.txt
+          prefect config set PREFECT_API_URL=$PREFECT_API_URL
+          python -m src.deploy_prefect.pr_deployment
+          prefect deployment run 'dev-order-rewards/dune-sync-pr-order-rewards'

src/deploy_prefect/dev_deployment.py

@@ -0,0 +1,23 @@
+"""Prefect Deployment for Order Rewards Data"""
+# pylint: disable=import-error
+from prefect import flow  # type: ignore
+from prefect.runner.storage import GitRepository  # type: ignore
+
+
+# pylint: disable=duplicate-code
+if __name__ == "__main__":
+    git_source = GitRepository(
+        url="https://github.com/cowprotocol/dune-sync.git",
+        branch="dev",
+    )
+    flow.from_source(
+        source=git_source,
+        entrypoint="src/deploy_prefect/flows.py:dev_order_rewards",
+    ).deploy(
+        name="dune-sync-dev-order-rewards",
+        work_pool_name="cowbarn",
+        cron="*/30 * * * *",  # Every 30 minutes
+        tags=["dev", "solver", "dune-sync"],
+        description="Run the dune sync order_rewards query",
+        version="0.0.2",
+    )

src/deploy_prefect/flows.py

@@ -0,0 +1,37 @@
+"""Prefect Deployment for Order Rewards Data"""
+# pylint: disable=import-error
+from prefect import flow  # type: ignore
+
+from src.deploy_prefect.tasks import (
+    get_block_range,
+    fetch_orderbook,
+    cast_orderbook_to_dune_string,
+    upload_data_to_dune,
+    update_aggregate_query,
+)
+from src.deploy_prefect.models import ENV, CHAIN, Config
+
+
+@flow(retries=3, retry_delay_seconds=60, log_prints=True)  # type: ignore[misc]
+def dev_order_rewards() -> None:
+    """Defines a flow for updating the order_rewards table"""
+    config = Config(CHAIN.MAINNET, ENV.DEV)
+
+    blockrange = get_block_range()
+    orderbook = fetch_orderbook(blockrange)
+    data = cast_orderbook_to_dune_string(orderbook)
+    table_name = upload_data_to_dune(
+        data, blockrange.block_from, blockrange.block_to, config
+    )
+    update_aggregate_query(table_name, config)
+
+
+@flow(retries=3, retry_delay_seconds=60, log_prints=True)  # type: ignore[misc]
+def prod_order_rewards() -> None:
+    """Defines a flow for updating the order_rewards table"""
+    config = Config(CHAIN.MAINNET, ENV.PROD)
+    blockrange = get_block_range()
+    orderbook = fetch_orderbook(blockrange)
+    data = cast_orderbook_to_dune_string(orderbook)
+    table_name = upload_data_to_dune(data, blockrange.block_from, blockrange.block_to)
+    update_aggregate_query(table_name, config)

src/deploy_prefect/local_deploy.py

@@ -5,7 +5,7 @@
 # pylint: disable=import-error
 from prefect import flow  # type: ignore
 from dotenv import load_dotenv
-from src.deploy_prefect.deployment import (
+from src.deploy_prefect.tasks import (
     get_block_range,
     fetch_orderbook,
     cast_orderbook_to_dune_string,

src/deploy_prefect/models.py

@@ -0,0 +1,100 @@
+"""Dataclasses for the prefect deployments"""
+import os
+from enum import Enum
+from dataclasses import dataclass, field
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class ENV(Enum):
+    """
+    Enum ENV class to change environment variables for DEV And PROD
+    """
+
+    DEV = "DEV"
+    PR = "PR"
+    PROD = "PROD"
+
+    def is_dev(self) -> bool:
+        """Check if the environment is DEV."""
+        return self == ENV.DEV
+
+    def is_pr(self) -> bool:
+        """Check if the environment is PR."""
+        return self == ENV.PR
+
+    def is_prod(self) -> bool:
+        """Check if the environment is PROD."""
+        return self == ENV.PROD
+
+
+class CHAIN(Enum):
+    """
+    Enum CHAIN class to change environment variables different chains
+    """
+
+    ARBITRUM = "ARBITRUM"
+    GNOSIS = "GNOSIS"
+    MAINNET = "MAINNET"
+
+    def is_arbitrum(self) -> bool:
+        """Check if the chain is Arbitrum"""
+        return self == CHAIN.ARBITRUM
+
+    def is_gnosis(self) -> bool:
+        """Check if the chain is Gnosis"""
+        return self == CHAIN.GNOSIS
+
+    def is_mainnet(self) -> bool:
+        """Check if the chain is mainnet"""
+        return self == CHAIN.MAINNET
+
+
+@dataclass(frozen=True)
+class Config:
+    """
+    Config dataclass to setup config based on chain&env combination.
+    """
+
+    _chain: CHAIN
+    _env: ENV
+
+    _dune_query_id: str = field(init=False)
+    _etherscan_api_key: str = field(init=False)
+
+    def __post_init__(self) -> None:
+        etherscan_api_value = os.environ["ETHERSCAN_API_KEY"]
+
+        if self._env.is_dev():
+            dune_query_id_value = os.environ["AGGREGATE_QUERY_DEV_ID"]
+        elif self._env.is_pr():
+            dune_query_id_value = os.environ["AGGREGATE_QUERY_PR_ID"]
+        elif self._env.is_prod():
+            dune_query_id_value = os.environ["AGGREGATE_QUERY_ID"]
+        else:
+            raise ValueError("ENV is neither DEV, PR, nor PROD")
+
+        object.__setattr__(self, "_etherscan_api_key", etherscan_api_value)
+        object.__setattr__(self, "_dune_query_id", dune_query_id_value)
+
+    @property
+    def etherscan_api_key(self) -> str:
+        """Etherscan API Key Getter"""
+        return self._etherscan_api_key
+
+    @property
+    def dune_query_id(self) -> str:
+        """Dune Aggregate Query Getter"""
+        return self._dune_query_id
+
+    @property
+    def env(self) -> str:
+        """ENV Getter"""
+        return self._env
+
+    @property
+    def chain(self) -> str:
+        """Chain Getter"""
+        return self._chain

src/deploy_prefect/pr_deployment.py

@@ -0,0 +1,25 @@
+"""Prefect Deployment for Order Rewards Data"""
+import os
+
+# pylint: disable=import-error
+from prefect import flow  # type: ignore
+from prefect.runner.storage import GitRepository  # type: ignore
+
+
+# pylint: disable=duplicate-code
+if __name__ == "__main__":
+    branch_name = os.getenv("BRANCH_NAME")
+    git_source = GitRepository(
+        url="https://github.com/cowprotocol/dune-sync.git",
+        branch=branch_name,
+    )
+    flow.from_source(
+        source=git_source,
+        entrypoint="src/deploy_prefect/flows.py:dev_order_rewards",
+    ).deploy(
+        name="dune-sync-pr-order-rewards",
+        work_pool_name="cowbarn",
+        tags=["dev", "solver", "dune-sync"],
+        description="Run the dune sync order_rewards query",
+        version="0.0.2",
+    )

src/deploy_prefect/prod_deployment.py

@@ -0,0 +1,22 @@
+"""Prefect Deployment for Order Rewards Data"""
+# pylint: disable=import-error
+from prefect import flow  # type: ignore
+from prefect.runner.storage import GitRepository  # type: ignore
+
+
+# pylint: disable=duplicate-code
+if __name__ == "__main__":
+    git_source = GitRepository(
+        url="https://github.com/cowprotocol/dune-sync.git",
+    )
+    flow.from_source(
+        source=git_source,
+        entrypoint="src/deploy_prefect/flows.py:prod_order_rewards",
+    ).deploy(
+        name="dune-sync-prod-order-rewards",
+        work_pool_name="cowbarn",
+        cron="*/30 * * * *",  # Every 30 minutes
+        tags=["prod", "solver", "dune-sync"],
+        description="Run the dune sync order_rewards query",
+        version="0.0.2",
+    )