Save progress on contract labels ingestion

src/op_analytics/datasources/contractlabels/dataaccess.py

@@ -0,0 +1,13 @@
+from op_analytics.coreutils.logger import structlog
+from op_analytics.coreutils.partitioned.dailydata import DailyDataset
+
+log = structlog.get_logger()
+
+
+class ContractLabels(DailyDataset):
+    """Various sources of truth for contract labels."""
+
+    MANUAL = "manual_v1"
+    OLI = "oli_v1"
+    OPATLAS = "opatlas_v1"
+    GROWTHEPIE = "growthepie_v1"

src/op_analytics/datasources/contractlabels/execute.py

@@ -0,0 +1,66 @@
+from op_analytics.coreutils.partitioned.dailydata import DEFAULT_DT
+from op_analytics.coreutils.misc import raise_for_schema_mismatch
+
+import polars as pl
+from .dataaccess import ContractLabels
+from .growthepie import GrowThePieLabels
+from .oli import OLILabels
+from .manual.fetch import ManualLabels
+from .opatlas import OPAtlasLabels
+
+
+def execute_pull():
+    """Pull the latest contract labels from the various source."""
+
+    df = ManualLabels.fetch()
+    ContractLabels.MANUAL.write(validate(df), sort_by=["address", "chain_id"])
+
+    df = GrowThePieLabels.fetch()
+    ContractLabels.GROWTHEPIE.write(validate(df), sort_by=["address", "chain_id"])
+
+    df = OLILabels.fetch()
+    ContractLabels.OLI.write(validate(df), sort_by=["address", "chain_id"])
+
+    df = OPAtlasLabels.fetch()
+    ContractLabels.OPATLAS.write(validate(df), sort_by=["address", "chain_id"])
+
+
+CONTRACT_LABELS_DF_SCHEMA = {
+    "address": pl.String,
+    "chain_id": pl.Int32,
+    "label_source": pl.String,
+    "project_name": pl.String,
+    "contract_name": pl.String,
+    "metadata": pl.String,
+}
+
+
+def validate(df: pl.DataFrame) -> pl.DataFrame:
+    """Validate the contract labels dataframe."""
+    # Make sure the schem is as expected.
+    raise_for_schema_mismatch(
+        actual_schema=df.schema,
+        expected_schema=pl.Schema(CONTRACT_LABELS_DF_SCHEMA),
+    )
+
+    # Ensure that addresses are lower case and have the proper length.
+    # TODO: Add more checks.
+    errors_df = pl.concat(
+        [
+            df.select(
+                pl.lit("wrong_address").alias("error_msg"),
+                (pl.col("address").str.len() != 42).alias("is_error"),
+            ),
+            df.select(
+                pl.lit("not_lowercase").alias("error_msg"),
+                (pl.col("address").str.to_lowercase() != pl.col("address")).alias("is_error"),
+            ),
+            # TODO: Check that there are no duplicate contract address.
+        ]
+    )
+
+    errors = errors_df.filter(pl.col("is_error")).to_dicts()
+    if errors:
+        raise ValueError(f"Invalid contract labels: {errors}")
+
+    return df.with_columns(dt=pl.lit(DEFAULT_DT))

src/op_analytics/datasources/contractlabels/growthepie/fetch.py

@@ -0,0 +1,19 @@
+from dataclasses import dataclass
+
+import io
+import polars as pl
+
+from op_analytics.coreutils.request import new_session
+
+
+@dataclass
+class GrowThePieLabels:
+    @classmethod
+    def fetch(cls) -> pl.DataFrame:
+        session = new_session()
+
+        response = session.get("https://api.growthepie.xyz/v1/oli/labels_decoded.parquet")
+        parquet_data = io.BytesIO(response.content)
+        df = pl.read_parquet(parquet_data)
+
+        return df

src/op_analytics/datasources/contractlabels/manual/fetch.py

@@ -0,0 +1,20 @@
+import json
+import os
+from dataclasses import dataclass
+
+import polars as pl
+
+
+DIRNAME = os.path.dirname(__file__)
+
+
+@dataclass
+class ManualLabels:
+    @classmethod
+    def fetch(cls) -> pl.DataFrame:
+        with open(os.path.join(DIRNAME, "labels.json"), "r") as fobj:
+            data = json.load(fobj)
+            df = pl.DataFrame(data)
+
+        # TODO: Fixes, and adapt the schema to what it should be for the common table.
+        return df