add comparison of datetime objects with strings

This commit is contained in:
Thomas Peetz
2025-06-04 14:06:18 +02:00
parent cc8a166f5c
commit 4bb7d61f80
+22 -8
View File
@@ -3,7 +3,7 @@ import data from json file to PostgreSQL
"""
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
from datetime import datetime
from datetime import datetime, date
from typing import Any, AnyStr, Dict, List
import os
import json
@@ -62,13 +62,9 @@ def get_ids(items: List[Any]) -> List[AnyStr]:
def update_item(db: Session, import_data: Dict[AnyStr, Any], item: Any, dry_run: bool, log):
for (key, value) in import_data.items():
existing_value = getattr(item, str(key))
update: bool = False
if isinstance(existing_value, datetime):
if str(existing_value) != str(value):
update = True
else:
if existing_value != value:
update = True
update: bool = has_changed(existing_value, value, log)
#if key == 'published_on':
# log.info(f"{type(value)}:{value} != {type(existing_value)}:{existing_value} : {update}")
if update:
log.info(f"update {key}({existing_value}) with {value}")
if not dry_run:
@@ -79,6 +75,24 @@ def update_item(db: Session, import_data: Dict[AnyStr, Any], item: Any, dry_run:
db.add(item)
db.commit()
def has_changed(existing_data: Any, import_data: AnyStr, log) -> bool:
if existing_data is None and import_data == 'None':
return False
if isinstance(existing_data, str):
return existing_data != import_data
if isinstance(existing_data, date):
if len(import_data) > 19:
import_date = datetime.strptime(import_data, "%Y-%m-%d %H:%M:%S.%f")
log.debug(f"{type(existing_data)}:{existing_data} == {import_date} : {existing_data != import_date}")
return existing_data != import_date
if len(import_data) > 10:
import_date = datetime.strptime(import_data, "%Y-%m-%d %H:%M:%S")
log.debug(f"{type(existing_data)}:{existing_data} == {import_date} : {existing_data != import_date}")
return existing_data != import_date
return existing_data.strftime("%Y-%m-%d") != import_data
return existing_data != import_data
def item_import(db: Session, import_data: Dict[AnyStr, Any], dry_run: bool, log):
log.info(f"import {import_data}")
if not dry_run: