feat: Add datetime.datetime support in pydict_to_metadata() for metadata forwarding

apis/c++/node/Cargo.toml

@@ -35,6 +35,7 @@ serde = { version = "1.0.164", features = ["derive"] }
 serde_json = { version = "1.0" }
 serde-big-array = { version = "0.5.1", optional = true }
 arrow = { workspace = true, features = ["ffi"] }
+chrono = { version = "0.4", features = ["serde"] }
 
 [build-dependencies]
 cxx-build = "1.0.73"

apis/c++/node/src/lib.rs

@@ -2,6 +2,7 @@ use std::{any::Any, collections::BTreeMap, vec};
 
 use crate::ffi::MetadataValueType;
 
+use chrono::{DateTime, Utc};
 use dora_node_api::{
     self, Event, EventStream, Metadata as DoraMetadata,
     MetadataParameters as DoraMetadataParameters, Parameter as DoraParameter,
@@ -60,6 +61,7 @@ mod ffi {
         ListInt,
         ListFloat,
         ListString,
+        Timestamp,
     }
 
     pub struct CombinedEvents {
@@ -140,6 +142,7 @@ mod ffi {
         fn get_list_int(self: &Metadata, key: &str) -> Result<Vec<i64>>;
         fn get_list_float(self: &Metadata, key: &str) -> Result<Vec<f64>>;
         fn get_list_string(self: &Metadata, key: &str) -> Result<Vec<String>>;
+        fn get_timestamp(self: &Metadata, key: &str) -> Result<String>;
         fn get_json(self: &Metadata, key: &str) -> Result<String>;
         fn to_json(self: &Metadata) -> String;
         fn list_keys(self: &Metadata) -> Vec<String>;
@@ -150,6 +153,7 @@ mod ffi {
         fn set_list_int(self: &mut Metadata, key: &str, value: Vec<i64>) -> Result<()>;
         fn set_list_float(self: &mut Metadata, key: &str, value: Vec<f64>) -> Result<()>;
         fn set_list_string(self: &mut Metadata, key: &str, value: Vec<String>) -> Result<()>;
+        fn set_timestamp(self: &mut Metadata, key: &str, value: String) -> Result<()>;
         #[cxx_name = "type"]
         fn value_type(self: &Metadata, key: &str) -> Result<MetadataValueType>;
     }
@@ -315,6 +319,7 @@ impl Metadata {
             DoraParameter::ListInt(_) => "list<int>",
             DoraParameter::ListFloat(_) => "list<float>",
             DoraParameter::ListString(_) => "list<string>",
+            DoraParameter::Timestamp(_) => "timestamp",
         }
     }
 
@@ -342,6 +347,7 @@ impl Metadata {
                     .map(|value| JsonValue::String(value.clone()))
                     .collect(),
             )),
+            DoraParameter::Timestamp(dt) => Ok(JsonValue::String(dt.to_rfc3339())),
         }
     }
 
@@ -426,6 +432,17 @@ impl Metadata {
         }
     }
 
+    pub fn get_timestamp(&self, key: &str) -> EyreResult<String> {
+        let parameter = self.expect_parameter(key)?;
+        match parameter {
+            DoraParameter::Timestamp(dt) => Ok(dt.to_rfc3339()),
+            other => Err(eyre!(
+                "metadata key '{key}' has type '{}', expected 'timestamp'",
+                Metadata::parameter_type_name(other)
+            )),
+        }
+    }
+
     pub fn get_json(&self, key: &str) -> EyreResult<String> {
         let parameter = self.expect_parameter(key)?;
         let json_value = Metadata::parameter_to_json(parameter, key)?;
@@ -479,6 +496,13 @@ impl Metadata {
         self.insert_parameter(key, DoraParameter::ListString(value))
     }
 
+    pub fn set_timestamp(&mut self, key: &str, value: String) -> EyreResult<()> {
+        let dt = DateTime::parse_from_rfc3339(&value)
+            .map_err(|e| eyre!("Failed to parse timestamp '{value}': {e}"))?
+            .with_timezone(&Utc);
+        self.insert_parameter(key, DoraParameter::Timestamp(dt))
+    }
+
     pub fn value_type(&self, key: &str) -> EyreResult<MetadataValueType> {
         let parameter = self.expect_parameter(key)?;
         let value_type = match parameter {
@@ -489,6 +513,7 @@ impl Metadata {
             DoraParameter::ListInt(_) => MetadataValueType::ListInt,
             DoraParameter::ListFloat(_) => MetadataValueType::ListFloat,
             DoraParameter::ListString(_) => MetadataValueType::ListString,
+            DoraParameter::Timestamp(_) => MetadataValueType::Timestamp,
         };
         Ok(value_type)
     }

apis/python/operator/Cargo.toml

@@ -22,3 +22,4 @@ arrow-schema = { workspace = true }
 aligned-vec = "0.5.0"
 futures = "0.3.28"
 futures-concurrency = "7.3.0"
+chrono = { version = "0.4", features = ["serde"] }

apis/python/operator/src/lib.rs

@@ -4,6 +4,7 @@ use std::{
 };
 
 use arrow::pyarrow::ToPyArrow;
+use chrono::{DateTime, Utc};
 use dora_node_api::{
     DoraNode, Event, EventStream, Metadata, MetadataParameters, Parameter, StopCause,
     merged::{MergeExternalSend, MergedEvent},
@@ -234,8 +235,39 @@ pub fn pydict_to_metadata(dict: Option<Bound<'_, PyDict>>) -> Result<MetadataPar
                 let list: Vec<String> = value.extract()?;
                 parameters.insert(key, Parameter::ListString(list))
             } else {
-                println!("could not convert type {value}");
-                parameters.insert(key, Parameter::String(value.str()?.to_string()))
+                // Check if it's a datetime.datetime object
+                let datetime_module = PyModule::import(value.py(), "datetime")
+                    .context("Failed to import datetime module")?;
+                let datetime_class = datetime_module.getattr("datetime")?;
+
+                if value.is_instance(datetime_class.as_ref())? {
+                    // Extract timestamp using timestamp() method
+                    let timestamp_float: f64 = value
+                        .call_method0("timestamp")?
+                        .extract()
+                        .context("Failed to extract timestamp from datetime")?;
+
+                    // Convert to chrono::DateTime<Utc>
+                    // timestamp() returns seconds since epoch as float
+                    // Convert to SystemTime first, then to DateTime<Utc>
+                    let seconds = timestamp_float as i64;
+                    let nanos = ((timestamp_float - seconds as f64) * 1_000_000_000.0) as u64;
+
+                    let system_time = if seconds >= 0 {
+                        UNIX_EPOCH + std::time::Duration::new(seconds as u64, nanos as u32)
+                    } else {
+                        UNIX_EPOCH
+                            .checked_sub(std::time::Duration::new((-seconds) as u64, nanos as u32))
+                            .unwrap_or(UNIX_EPOCH)
+                    };
+
+                    let dt = DateTime::<Utc>::from(system_time);
+
+                    parameters.insert(key, Parameter::Timestamp(dt))
+                } else {
+                    println!("could not convert type {value}");
+                    parameters.insert(key, Parameter::String(value.str()?.to_string()))
+                }
             };
         }
     }
@@ -306,6 +338,26 @@ pub fn metadata_to_pydict<'a>(
             Parameter::ListString(l) => dict
                 .set_item(k, l)
                 .context("Could not insert metadata into python dictionary")?,
+            Parameter::Timestamp(dt) => {
+                // Convert chrono::DateTime<Utc> to Python datetime.datetime
+                let timestamp = dt.timestamp();
+                let microseconds = dt.timestamp_subsec_micros();
+
+                // Get UTC timezone from Python's datetime module
+                let datetime_module =
+                    PyModule::import(py, "datetime").context("Failed to import datetime module")?;
+                let datetime_class = datetime_module.getattr("datetime")?;
+                let utc_timezone = datetime_module.getattr("timezone")?.getattr("utc")?;
+
+                // Create timezone-aware datetime using fromtimestamp
+                let total_seconds = timestamp as f64 + microseconds as f64 / 1_000_000.0;
+                let py_datetime = datetime_class
+                    .call_method1("fromtimestamp", (total_seconds, utc_timezone))
+                    .context("Failed to create Python datetime from timestamp")?;
+
+                dict.set_item(k, py_datetime)
+                    .context("Could not insert timestamp into python dictionary")?
+            }
         }
     }
 

binaries/cli/src/command/topic/echo.rs

@@ -229,6 +229,9 @@ async fn log_to_terminal(
                             Parameter::Float(value) => serde_json::to_string(value).unwrap(),
                             Parameter::ListFloat(value) => serde_json::to_string(value).unwrap(),
                             Parameter::ListString(value) => serde_json::to_string(value).unwrap(),
+                            Parameter::Timestamp(dt) => {
+                                serde_json::to_string(&dt.to_rfc3339()).unwrap()
+                            }
                         };
                         write!(output, "{}:{value}", serde_json::Value::String(k.clone()),)
                             .unwrap();

libraries/message/src/metadata.rs

@@ -1,6 +1,7 @@
 use std::collections::BTreeMap;
 
 use arrow_schema::DataType;
+use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 
 /// Additional data that is sent as part of output messages.
@@ -69,6 +70,7 @@ pub enum Parameter {
     Float(f64),
     ListFloat(Vec<f64>),
     ListString(Vec<String>),
+    Timestamp(DateTime<Utc>),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]