(
df_to_join: dd.DataFrame,
timestamp_field: str,
created_timestamp_column: Optional[str] = None,
)
| 1104 | |
| 1105 | |
| 1106 | def _normalize_timestamp( |
| 1107 | df_to_join: dd.DataFrame, |
| 1108 | timestamp_field: str, |
| 1109 | created_timestamp_column: Optional[str] = None, |
| 1110 | ) -> dd.DataFrame: |
| 1111 | df_to_join_types = df_to_join.dtypes |
| 1112 | timestamp_field_type = df_to_join_types[timestamp_field] |
| 1113 | |
| 1114 | if created_timestamp_column: |
| 1115 | created_timestamp_column_type = df_to_join_types[created_timestamp_column] |
| 1116 | |
| 1117 | # TODO: need to figure out why the value of timestamp_field_type.tz is pytz.UTC |
| 1118 | if not hasattr(timestamp_field_type, "tz") or timestamp_field_type.tz != pytz.UTC: |
| 1119 | # if you are querying for the event timestamp field, we have to deduplicate |
| 1120 | if len(df_to_join[timestamp_field].shape) > 1: |
| 1121 | df_to_join, dups = _df_column_uniquify(df_to_join) |
| 1122 | df_to_join = df_to_join.drop(columns=dups) |
| 1123 | |
| 1124 | # Make sure all timestamp fields are tz-aware. We default tz-naive fields to UTC |
| 1125 | df_to_join[timestamp_field] = df_to_join[timestamp_field].apply( |
| 1126 | lambda x: x if x.tzinfo else x.replace(tzinfo=timezone.utc), |
| 1127 | meta=(timestamp_field, "datetime64[ns, UTC]"), |
| 1128 | ) |
| 1129 | |
| 1130 | # TODO: need to figure out why the value of created_timestamp_column_type.tz is pytz.UTC |
| 1131 | if created_timestamp_column and ( |
| 1132 | not hasattr(created_timestamp_column_type, "tz") |
| 1133 | or created_timestamp_column_type.tz != pytz.UTC |
| 1134 | ): |
| 1135 | if len(df_to_join[created_timestamp_column].shape) > 1: |
| 1136 | # if you are querying for the created timestamp field, we have to deduplicate |
| 1137 | df_to_join, dups = _df_column_uniquify(df_to_join) |
| 1138 | df_to_join = df_to_join.drop(columns=dups) |
| 1139 | |
| 1140 | df_to_join[created_timestamp_column] = df_to_join[ |
| 1141 | created_timestamp_column |
| 1142 | ].apply( |
| 1143 | lambda x: x if x.tzinfo else x.replace(tzinfo=timezone.utc), |
| 1144 | meta=(timestamp_field, "datetime64[ns, UTC]"), |
| 1145 | ) |
| 1146 | |
| 1147 | return df_to_join.persist() |
| 1148 | |
| 1149 | |
| 1150 | def _filter_ttl( |
no test coverage detected