From e491dbd3aeccc9d88994832554c73263a738d2d5 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Javier=20R=C3=ADos?= <javier.rios-urbano@epitech.eu>
Date: Fri, 20 Sep 2024 00:57:59 +0200
Subject: [PATCH] add: `dagster-k8s/config` tag to eas asset (#2180)

---
 warehouse/oso_dagster/assets/eas_optimism.py | 31 ++++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/warehouse/oso_dagster/assets/eas_optimism.py b/warehouse/oso_dagster/assets/eas_optimism.py
index d6542274..576d624f 100644
--- a/warehouse/oso_dagster/assets/eas_optimism.py
+++ b/warehouse/oso_dagster/assets/eas_optimism.py
@@ -35,6 +35,30 @@ class Attestation(BaseModel):
 # A sensible limit for the number of nodes to fetch per page
 EAS_OPTIMISM_STEP_NODES_PER_PAGE = 10_000
 
+# Kubernetes configuration for the asset materialization
+K8S_CONFIG = {
+    "merge_behavior": "SHALLOW",
+    "container_config": {
+        "resources": {
+            "requests": {"cpu": "2000m", "memory": "3584Mi"},
+            "limits": {"cpu": "2000m", "memory": "3584Mi"},
+        },
+    },
+    "pod_spec_config": {
+        "node_selector": {
+            "pool_type": "spot",
+        },
+        "tolerations": [
+            {
+                "key": "pool_type",
+                "operator": "Equal",
+                "value": "spot",
+                "effect": "NoSchedule",
+            }
+        ],
+    },
+}
+
 
 def get_optimism_eas_data(
     context: AssetExecutionContext, client: Client, date_from: float, date_to: float
@@ -158,9 +182,12 @@ def get_optimism_eas(context: AssetExecutionContext, client: Client):
 @dlt_factory(
     key_prefix="ethereum_attestation_service_optimism",
     partitions_def=WeeklyPartitionsDefinition(
-        start_date=EAS_OPTIMISM_FIRST_ATTESTATION.isoformat().split("T")[0],
-        end_date=(datetime.now()).isoformat().split("T")[0],
+        start_date=EAS_OPTIMISM_FIRST_ATTESTATION.isoformat().split("T", maxsplit=1)[0],
+        end_date=(datetime.now()).isoformat().split("T", maxsplit=1)[0],
     ),
+    op_tags={
+        "dagster-k8s/config": K8S_CONFIG,
+    },
 )
 def attestations(context: AssetExecutionContext):
     """