Extractors: Update ingestion

.gitignore

@@ -173,6 +173,7 @@ deployments
 logs
 
 # local data
+clickhouse_data
 postgres-data
 parquet-data
 data

docker-compose.yml

@@ -9,7 +9,7 @@ services:
         soft: 262144
         hard: 262144
     volumes:
-      - clickhouse_data:/var/lib/clickhouse
+      - ./clickhouse_data:/var/lib/clickhouse
       - ./parquet-data:/var/lib/clickhouse/user_files/parquet-data
     ports:
       - 8123:8123
@@ -85,6 +85,8 @@ services:
     image: ghcr.io/synthetixio/data/extractors:${VERSION}
     build:
       context: ./extractors
+    networks:
+      - data
     env_file:
       - .env
     volumes:

extractors/pyproject.toml

@@ -5,6 +5,7 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
+    "clickhouse-connect>=0.8.9",
     "cryo==0.3.0",
     "duckdb>=1.1.3",
     "maturin>=1.7.7",

extractors/src/clean.py

@@ -5,6 +5,7 @@
 from eth_utils import decode_hex
 import polars as pl
 import duckdb
+from .insert import insert_data
 
 
 def fix_labels(labels):
@@ -58,7 +59,7 @@ def decode_output(contract, function_name, call):
 
 
 def camel_to_snake(name):
-    name = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name).lower()
+    name = re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()
     return name
 
 
@@ -75,7 +76,12 @@ def get_labels(contract, function_name):
     return input_names, output_names
 
 
-def clean_data(chain_name, protocol, contract, function_name, write=True):
+def clean_data(
+    chain_name, protocol, contract, contract_name, function_name, write=True
+):
+    function_name_snake = camel_to_snake(function_name)
+    contract_name_snake = camel_to_snake(contract_name)
+    directory_name = f"{contract_name_snake}_function_{function_name_snake}"
     input_labels, output_labels = get_labels(contract, function_name)
 
     # fix labels
@@ -86,7 +92,7 @@ def clean_data(chain_name, protocol, contract, function_name, write=True):
     df = duckdb.sql(
         f"""
         SELECT DISTINCT *
-        FROM '../parquet-data/extractors/raw/{chain_name}/{protocol}/{function_name}/*.parquet'
+        FROM '../parquet-data/extractors/raw/{chain_name}/{protocol}/{directory_name}/*.parquet'
         WHERE
             call_data IS NOT NULL
             AND output_data IS NOT NULL
@@ -127,7 +133,7 @@ def clean_data(chain_name, protocol, contract, function_name, write=True):
 
     # write the data
     if write:
-        file_path = f"../parquet-data/extractors/clean/{chain_name}/{protocol}/{function_name}.parquet"
+        file_path = f"../parquet-data/extractors/clean/{chain_name}/{protocol}/{directory_name}/{directory_name}.parquet"
 
         ensure_directory_exists(file_path)
         # write the data
@@ -137,6 +143,9 @@ def clean_data(chain_name, protocol, contract, function_name, write=True):
         """
         )
 
+        # insert to clickhouse
+        insert_data(chain_name, protocol, contract_name_snake, function_name_snake)
+
     return df
 
 

extractors/src/extract.py

@@ -1,7 +1,7 @@
 import cryo
 from synthetix import Synthetix
 from .constants import CHAIN_CONFIGS
-from .clean import clean_data, clean_blocks
+from .clean import clean_data, clean_blocks, camel_to_snake
 
 
 def get_synthetix(chain_config):
@@ -38,7 +38,9 @@ def extract_data(
     # get synthetix
     chain_config = CHAIN_CONFIGS[network_id]
     snx = get_synthetix(chain_config)
-    output_dir = f"/parquet-data/extractors/raw/{chain_config['name']}/{protocol}/{function_name}"
+    function_name_snake = camel_to_snake(function_name)
+    contract_name_snake = camel_to_snake(contract_name)
+    output_dir = f"/parquet-data/extractors/raw/{chain_config['name']}/{protocol}/{contract_name_snake}_function_{function_name_snake}"
 
     # encode the call data
     contract = snx.contracts[package_name][contract_name]["contract"]
@@ -61,7 +63,9 @@ def extract_data(
     )
 
     if clean:
-        df_clean = clean_data(chain_config["name"], protocol, contract, function_name)
+        df_clean = clean_data(
+            chain_config["name"], protocol, contract, contract_name, function_name
+        )
         return df_clean
 
 

extractors/src/insert.py

@@ -0,0 +1,25 @@
+import re
+import clickhouse_connect
+from clickhouse_connect.driver.client import Client
+
+CLICKHOUSE_INTERNAL_PATH = (
+    "/var/lib/clickhouse/user_files/parquet-data/extractors/clean"
+)
+RAW_DATA_PATH = "/parquet-data/extractors/raw"
+CLEAN_DATA_PATH = "/parquet-data/extractors/clean"
+
+
+def convert_case(name):
+    snake_case = re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()
+    return snake_case
+
+
+def insert_data(network: str, protocol: str, contract_name: str, function_name: str):
+    client: Client = clickhouse_connect.get_client(
+        host="clickhouse", port=8123, user="default"
+    )
+    table_name = f"raw_{network}.{protocol}_{contract_name}_function_{function_name}"
+    file_path = f"{CLICKHOUSE_INTERNAL_PATH}/{network}/{protocol}/{contract_name}_function_{function_name}/*.parquet"
+    query = f"insert into {table_name} select * from file('{file_path}', 'Parquet')"
+    client.command(query)
+    client.close()