debugging another try

Author: ero67

pipeline-project/k8s/base/taxi-data-processing-job.yaml

@@ -10,53 +10,14 @@ spec:
       - name: taxi-flow
         image: gcr.io/teak-gamma-442315-f8/taxi-flow:latest
         imagePullPolicy: Always
-        command: ["python"]
-        args: [deploy.py]
-        resources:
-          requests:
-            memory: "1Gi"
-            cpu: "200m"
-          limits:
-            memory: "2Gi"
-            cpu: "400m"
+        command: ["python", "deploy.py"]
         env:
         - name: PREFECT_API_URL
           value: "http://prefect-server:4200/api"
         - name: PREFECT_LOGGING_LEVEL
           value: "DEBUG"
         - name: PYTHONUNBUFFERED
           value: "1"
-        - name: DB_HOST
-          value: "cloudsql-proxy"
-        - name: DB_PORT
-          value: "5432"
-        - name: CLOUDSQL_CONNECTION_NAME
-          valueFrom:
-            configMapKeyRef:
-              name: pipeline-config
-              key: CLOUDSQL_CONNECTION_NAME
-        - name: DB_USER
-          valueFrom:
-            secretKeyRef:
-              name: cloudsql-credentials
-              key: username
-        - name: DB_PASSWORD
-          valueFrom:
-            secretKeyRef:
-              name: cloudsql-credentials
-              key: password
-        - name: DB_NAME
-          valueFrom:
-            secretKeyRef:
-              name: cloudsql-credentials
-              key: database
-        - name: GCS_BUCKET
-          valueFrom:
-            configMapKeyRef:
-              name: pipeline-config
-              key: GCS_BUCKET
-        - name: GOOGLE_APPLICATION_CREDENTIALS
-          value: /var/secrets/google/key.json
         volumeMounts:
         - name: gcp-key
           mountPath: "/var/secrets/google"
@@ -70,4 +31,4 @@ spec:
       - name: temp-storage
         emptyDir: {}
       restartPolicy: Never
-  backoffLimit: 0
+  backoffLimit: 2

pipeline-project/src/processing/flows/deploy.py

@@ -1,36 +1,15 @@
-# import os
-# from prefect import flow
-# from prefect_kubernetes.jobs import KubernetesJob
-# from taxi_data_flow import NY_Taxi_Data_Flow
-
-# # Set environment variable for deployment creation
-# os.environ["PREFECT_API_URL"] = "http://prefect-server:4200/api"
-
-# # Create deployment
-# deployment = NY_Taxi_Data_Flow.to_deployment(
-#     name="taxi-data-flow",
-#     work_pool_name="k8s-pool",
-#     work_queue_name="default"
-# )
-
-# if __name__ == "__main__":
-#     print("Registering flow deployment with Prefect server...")
-#     deployment.apply()  # Just register and exit
-#     print("Flow deployment registered successfully!")
-import os
 from prefect import serve
-from prefect_kubernetes.jobs import KubernetesJob
+from prefect.deployments import Deployment
 from taxi_data_flow import NY_Taxi_Data_Flow
 
-# Set environment variable for deployment creation
-os.environ["PREFECT_API_URL"] = "http://prefect-server:4200/api"
-
-# Create deployment with job configuration
-deployment = NY_Taxi_Data_Flow.to_deployment(
+# Create deployment
+deployment = Deployment.build_from_flow(
+    flow=NY_Taxi_Data_Flow,
     name="taxi-data-flow",
     work_pool_name="k8s-pool",
     work_queue_name="default"
 )
 
 if __name__ == "__main__":
-    serve(deployment)
+    deployment.apply()
+    print("Deployment registered successfully!")