fixed polar data type issue, where it would not save the data properly because a datetime variable would be str instead of datetime

Author: Luis Gonzalez

airflow/dags/my_dag.py

@@ -0,0 +1,21 @@
+from airflow.sdk import dag, task
+from pendulum import datetime
+
+@dag(
+    schedule="@hourly",
+    start_date=datetime(2026, 2, 2),
+    description="Run Celery queue with RabbitMQ as the broker \
+                 in order to get GitHub data from the GitHub API",
+    tags=["celery_queue"],
+    max_consecutive_failed_dag_runs=3,
+)
+def run_queue():
+    
+    @task
+    def run_the_queue():
+        print("hello")
+
+    run_the_queue()
+
+
+run_queue()

airflow/dags/run_queue.py

@@ -7,7 +7,7 @@
 import polars as pl
 
 
-today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+today = datetime.now().strftime("%Y-%m-%d")
 
 print("Waiting for Celery task to complete")
 

client.py

@@ -1,10 +1,14 @@
 import duckdb 
 
-df = duckdb.read_parquet("../2026-01-14/hey.parquet")
+duckdb.sql("SET display_max_rows=10000")
 
-duckdb.sql("DESCRIBE SELECT * FROM df").show()
+df = duckdb.read_parquet("../2026-02-02/github_data.parquet")
 
-duckdb.sql("SELECT language, COUNT(language) AS c_p \
-            FROM df \
-            GROUP BY language \
-            ORDER BY c_p DESC").show()
+# duckdb.sql("DESCRIBE SELECT * FROM df").show()
+
+# duckdb.sql("SELECT language, COUNT(language) AS c_p \
+#             FROM df \
+#             GROUP BY language \
+#             ORDER BY c_p DESC").show()
+
+duckdb.sql("DESCRIBE SELECT * FROM df").show()

data/analytics/ducky.py

@@ -84,6 +84,50 @@ services:
       celery_worker:
         condition: service_healthy
 
+
+  # airflow-init:
+  #   image: apache/airflow:2.9.2
+  #   environment:
+  #     AIRFLOW__CORE__EXECUTOR: LocalExecutor
+  #     AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: sqlite:////opt/airflow/airflow.db
+  #     AIRFLOW__CORE__LOAD_EXAMPLES: "false"
+  #   volumes:
+  #     - ./airflow/dags:/opt/airflow/dags
+  #     - ./:/opt/airflow/project
+  #     - airflow_data:/opt/airflow
+  #   command: ["airflow", "db", "init"]
+
+  # airflow-scheduler:
+  #   image: apache/airflow:2.9.2
+  #   depends_on:
+  #     - airflow-init
+  #   environment:
+  #     AIRFLOW__CORE__EXECUTOR: LocalExecutor
+  #     AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: sqlite:////opt/airflow/airflow.db
+  #     AIRFLOW__CORE__LOAD_EXAMPLES: "false"
+  #   volumes:
+  #     - ./airflow/dags:/opt/airflow/dags
+  #     - ./:/opt/airflow/project
+  #     - airflow_data:/opt/airflow
+  #   command: ["airflow", "scheduler"]
+
+  # airflow-webserver:
+  #   image: apache/airflow:2.9.2
+  #   depends_on:
+  #     - airflow-init
+  #   environment:
+  #     AIRFLOW__CORE__EXECUTOR: LocalExecutor
+  #     AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: sqlite:////opt/airflow/airflow.db
+  #     AIRFLOW__CORE__LOAD_EXAMPLES: "false"
+  #   volumes:
+  #     - ./airflow/dags:/opt/airflow/dags
+  #     - ./:/opt/airflow/project
+  #     - airflow_data:/opt/airflow
+  #   ports:
+  #     - "8080:8080"
+  #   command: ["airflow", "webserver"]
+  
+
 volumes:
   rabbitmq_data:
   redis_data:

docker-compose.yml

@@ -6,7 +6,7 @@
 class RabbitMQ_Data_Validation(BaseModel):
     # ===== MESSAGE METADATA =====
     message_id: str
-    got_data_in: str = Field(default_factory=lambda: datetime.now().isoformat())
+    got_data_in: datetime = Field(default_factory=lambda: datetime.now())
 
     # ===== BASIC INFO =====
     repo_id: int

pydantic_models/github.py

@@ -15,7 +15,7 @@
 
 
 logger = get_task_logger(__name__)
-todays_date = datetime.now().isoformat()
+todays_date = datetime.now()
 S3_BUCKET_NAME = "github-etl-data-bucket"
 
 
@@ -70,7 +70,7 @@ def get_github_data(self, start_in_repo_num: int = 0, batch_size: int = 500, git
             try:
                 github_data_points = {
                     "message_id": self.request.id,
-                    "got_data_in": todays_date if todays_date else datetime.now().isoformat(),
+                    "got_data_in": todays_date if todays_date else datetime.now(),
                     "repo_id": repo.id if repo.id is not None else 0,
                     "name": repo.name if repo.name else "unknown",
                     "full_name": repo.full_name if repo.full_name else "unknown/unknown",