bsiggel/motia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update logging to use Motia workbench context for visibility

Browse Source
This commit is contained in:
root
2025-10-24 00:12:53 +00:00
parent ddad58faa3
commit 409bea3615
1 changed files with 68 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
bitbylaw/steps/advoware_cal_sync/calendar_sync_event_step.py
View File

@@ -30,18 +30,28 @@ FETCH_TO = f"{current_year + 9}-12-31T23:59:59" # End of 9 years ahead
CALENDAR_SYNC_LOCK_KEY = 'calendar_sync_lock'
def log_operation(level, message, **context):
"""Centralized logging with context."""
context_str = ' '.join(f"{k}={v}" for k, v in context.items() if v is not None)
def log_operation(level, message, context=None, **context_vars):
"""Centralized logging with context, supporting Motia workbench logging."""
context_str = ' '.join(f"{k}={v}" for k, v in context_vars.items() if v is not None)
full_message = f"{message} {context_str}".strip()
if level == 'info':
logger.info(full_message)
elif level == 'warning':
logger.warning(full_message)
elif level == 'error':
logger.error(full_message)
elif level == 'debug':
logger.debug(full_message)
if context:
if level == 'info':
context.logger.info(full_message)
elif level == 'warning':
context.logger.warning(full_message)
elif level == 'error':
context.logger.error(full_message)
elif level == 'debug':
context.logger.debug(full_message)
else:
if level == 'info':
logger.info(full_message)
elif level == 'warning':
logger.warning(full_message)
elif level == 'error':
logger.error(full_message)
elif level == 'debug':
logger.debug(full_message)
async def connect_db():
"""Connect to Postgres DB from Config."""
@@ -571,9 +581,9 @@ def get_timestamps(adv_data, google_data):
return adv_ts, google_ts
async def process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware):
async def process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware, context=None):
"""Phase 1: Process new appointments from Advoware to Google."""
logger.info("Phase 1: Processing new appointments from Advoware")
log_operation('info', "Phase 1: Processing new appointments from Advoware", context=context)
for frnr, app in state['adv_map'].items():
if frnr not in state['db_adv_index']:
try:
@@ -586,14 +596,14 @@ async def process_new_from_advoware(state, conn, service, calendar_id, kuerzel,
""",
kuerzel, int(frnr), event_id
)
logger.info(f"Phase 1: Created new from Advoware: frNr {frnr}, event_id {event_id}")
log_operation('info', f"Phase 1: Created new from Advoware: frNr {frnr}, event_id {event_id}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
except Exception as e:
logger.warning(f"Phase 1: Failed to process new Advoware {frnr}: {e}")
log_operation('warning', f"Phase 1: Failed to process new Advoware {frnr}: {e}", context=context)
async def process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware):
async def process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware, context=None):
"""Phase 2: Process new events from Google to Advoware."""
logger.info("Phase 2: Processing new events from Google")
log_operation('info', "Phase 2: Processing new events from Google", context=context)
for event_id, evt in state['google_map'].items():
# For recurring events, check if the master event (recurringEventId) is already synced
# For regular events, check the event_id directly
@@ -612,15 +622,15 @@ async def process_new_from_google(state, conn, service, calendar_id, kuerzel, ad
""",
kuerzel, int(frnr), event_id
)
logger.info(f"Phase 2: Created new from Google: event_id {event_id}, frNr {frnr}")
log_operation('info', f"Phase 2: Created new from Google: event_id {event_id}, frNr {frnr}", context=context)
else:
logger.warning(f"Phase 2: Skipped DB insert for Google event {event_id}, frNr is None")
log_operation('warning', f"Phase 2: Skipped DB insert for Google event {event_id}, frNr is None", context=context)
except Exception as e:
logger.warning(f"Phase 2: Failed to process new Google {event_id}: {e}")
log_operation('warning', f"Phase 2: Failed to process new Google {event_id}: {e}", context=context)
async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware):
async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware, context=None):
"""Phase 3: Process deleted entries."""
logger.info("Phase 3: Processing deleted entries")
log_operation('info', "Phase 3: Processing deleted entries", context=context)
for row in state['rows']:
frnr = row['advoware_frnr']
event_id = row['google_event_id']
@@ -643,7 +653,7 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
# Both missing - soft delete
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id'])
logger.info(f"Phase 3: Soft deleted sync_id {row['sync_id']} (both missing)")
log_operation('info', f"Phase 3: Soft deleted sync_id {row['sync_id']} (both missing)", context=context)
elif not adv_exists:
# Missing in Advoware
strategy = row['sync_strategy']
@@ -654,10 +664,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
await delete_google_event(service, calendar_id, event_id)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id'])
logger.info(f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}")
log_operation('info', f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
except Exception as e:
logger.warning(f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
elif row['source_system'] == 'google' and row['advoware_write_allowed']:
@@ -667,13 +677,13 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
if new_frnr and str(new_frnr) != 'None':
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET advoware_frnr = $1, sync_status = 'synced', last_sync = $3 WHERE sync_id = $2;", int(new_frnr), row['sync_id'], datetime.datetime.now(BERLIN_TZ))
logger.info(f"Phase 3: Recreated Advoware appointment {new_frnr} for sync_id {row['sync_id']}")
log_operation('info', f"Phase 3: Recreated Advoware appointment {new_frnr} for sync_id {row['sync_id']}", context=context)
else:
logger.warning(f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}, frNr is None")
log_operation('warning', f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}, frNr is None", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
except Exception as e:
logger.warning(f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 3: Failed to recreate Advoware for sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
else:
@@ -682,10 +692,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
await delete_google_event(service, calendar_id, event_id)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id'])
logger.info(f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}")
log_operation('info', f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
except Exception as e:
logger.warning(f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
else:
@@ -694,10 +704,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
await delete_google_event(service, calendar_id, event_id)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id'])
logger.info(f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}")
log_operation('info', f"Phase 3: Propagated delete to Google for sync_id {row['sync_id']}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
except Exception as e:
logger.warning(f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 3: Failed to delete Google for sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
elif not google_exists:
@@ -711,13 +721,13 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
await safe_delete_advoware_appointment(advoware, frnr, row['advoware_write_allowed'])
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id'])
logger.info(f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}")
log_operation('info', f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}", context=context)
except Exception as e:
logger.warning(f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
else:
logger.warning(f"Phase 3: Cannot delete in Advoware for sync_id {row['sync_id']}, write not allowed")
log_operation('warning', f"Phase 3: Cannot delete in Advoware for sync_id {row['sync_id']}, write not allowed", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
elif row['source_system'] == 'advoware':
@@ -726,10 +736,10 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
new_event_id = await create_google_event(service, calendar_id, standardize_appointment_data(state['adv_map'][str(frnr)], 'advoware'))
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET google_event_id = $1, sync_status = 'synced', last_sync = $3 WHERE sync_id = $2;", new_event_id, row['sync_id'], datetime.datetime.now(BERLIN_TZ))
logger.info(f"Phase 3: Recreated Google event {new_event_id} for sync_id {row['sync_id']}")
log_operation('info', f"Phase 3: Recreated Google event {new_event_id} for sync_id {row['sync_id']}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
except Exception as e:
logger.warning(f"Phase 3: Failed to recreate Google for sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 3: Failed to recreate Google for sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
else:
@@ -738,15 +748,15 @@ async def process_deleted_entries(state, conn, service, calendar_id, kuerzel, ad
await safe_delete_advoware_appointment(advoware, frnr, row['advoware_write_allowed'])
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET deleted = TRUE, sync_status = 'synced' WHERE sync_id = $1;", row['sync_id'])
logger.info(f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}")
log_operation('info', f"Phase 3: Propagated delete to Advoware for sync_id {row['sync_id']}", context=context)
except Exception as e:
logger.warning(f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 3: Failed to delete Advoware for sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
async def process_updates(state, conn, service, calendar_id, kuerzel, advoware, context=None):
"""Phase 4: Process updates for existing entries."""
logger.info("Phase 4: Processing updates for existing entries")
log_operation('info', "Phase 4: Processing updates for existing entries", context=context)
# Track which master events we've already processed to avoid duplicate updates
processed_master_events = set()
@@ -793,32 +803,32 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
await update_google_event(service, calendar_id, event_id, adv_std)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ))
logger.info(f"Phase 4: Updated Google event {event_id} from Advoware frNr {frnr}")
log_operation('info', f"Phase 4: Updated Google event {event_id} from Advoware frNr {frnr}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
elif google_ts and google_ts > row['last_sync']:
logger.warning(f"Phase 4: Unauthorized change in Google event {event_id}, resetting to Advoware frNr {frnr}")
log_operation('warning', f"Phase 4: Unauthorized change in Google event {event_id}, resetting to Advoware frNr {frnr}", context=context)
await update_google_event(service, calendar_id, event_id, adv_std)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ))
logger.info(f"Phase 4: Reset Google event {event_id} to Advoware frNr {frnr}")
log_operation('info', f"Phase 4: Reset Google event {event_id} to Advoware frNr {frnr}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
elif row['source_system'] == 'google' and row['advoware_write_allowed']:
# Check for changes in source (Google) or unauthorized changes in target (Advoware)
google_ts_str = google_data.get('updated', '')
google_ts = datetime.datetime.fromisoformat(google_ts_str.rstrip('Z')).astimezone(BERLIN_TZ) if google_ts_str else None
adv_ts = BERLIN_TZ.localize(datetime.datetime.fromisoformat(adv_data['zuletztGeaendertAm']))
logger.debug(f"Phase 4: Checking sync_id {row['sync_id']}: adv_ts={adv_ts}, google_ts={google_ts}, last_sync={row['last_sync']}")
log_operation('debug', f"Phase 4: Checking sync_id {row['sync_id']}: adv_ts={adv_ts}, google_ts={google_ts}, last_sync={row['last_sync']}", context=context)
if google_ts and google_ts > row['last_sync']:
await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'], row['employee_kuerzel'])
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ))
logger.info(f"Phase 4: Updated Advoware frNr {frnr} from Google event {event_id}")
log_operation('info', f"Phase 4: Updated Advoware frNr {frnr} from Google event {event_id}", context=context)
elif adv_ts > row['last_sync']:
logger.warning(f"Phase 4: Unauthorized change in Advoware frNr {frnr}, resetting to Google event {event_id}")
log_operation('warning', f"Phase 4: Unauthorized change in Advoware frNr {frnr}, resetting to Google event {event_id}", context=context)
await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'], row['employee_kuerzel'])
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], datetime.datetime.now(BERLIN_TZ))
logger.info(f"Phase 4: Reset Advoware frNr {frnr} to Google event {event_id}")
log_operation('info', f"Phase 4: Reset Advoware frNr {frnr} to Google event {event_id}", context=context)
elif strategy == 'last_change_wins':
adv_ts = await get_advoware_timestamp(advoware, frnr)
google_ts_str = google_data.get('updated', '')
@@ -830,10 +840,10 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
await safe_update_advoware_appointment(advoware, frnr, google_std, row['advoware_write_allowed'], row['employee_kuerzel'])
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'synced', last_sync = $2 WHERE sync_id = $1;", row['sync_id'], max(adv_ts, google_ts))
logger.info(f"Phase 4: Updated based on last_change_wins for sync_id {row['sync_id']}")
log_operation('info', f"Phase 4: Updated based on last_change_wins for sync_id {row['sync_id']}", context=context)
await asyncio.sleep(0.1) # Small delay to avoid rate limits
except Exception as e:
logger.warning(f"Phase 4: Failed to update sync_id {row['sync_id']}: {e}")
log_operation('warning', f"Phase 4: Failed to update sync_id {row['sync_id']}: {e}", context=context)
async with conn.transaction():
await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
"""Main event handler for calendar sync."""
@@ -904,7 +914,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
for row in state['rows']:
if row['google_event_id']:
state['db_google_index'][row['google_event_id']] = row
log_operation('debug', "Reloaded indexes", rows=len(state['rows']), adv=len(state['db_adv_index']), google=len(state['db_google_index']))
log_operation('debug', "Reloaded indexes", context=context, rows=len(state['rows']), adv=len(state['db_adv_index']), google=len(state['db_google_index']))
async def reload_api_maps():
"""Reload API maps after creating new events in phases."""
@@ -912,16 +922,16 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
state['adv_map'] = {str(app['frNr']): app for app in state['adv_appointments'] if app.get('frNr')}
state['google_events'] = await fetch_google_events(service, calendar_id)
state['google_map'] = {evt['id']: evt for evt in state['google_events']}
log_operation('debug', "Reloaded API maps", adv=len(state['adv_map']), google=len(state['google_map']))
log_operation('debug', "Reloaded API maps", context=context, adv=len(state['adv_map']), google=len(state['google_map']))
# Initial fetch
log_operation('info', "Fetching fresh data from APIs")
log_operation('info', "Fetching fresh data from APIs", context=context)
await reload_api_maps()
await reload_db_indexes()
log_operation('info', "Fetched existing sync rows", count=len(state['rows']))
log_operation('info', "Fetched existing sync rows", context=context, count=len(state['rows']))
# Phase 1: New from Advoware => Google
await process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware)
await process_new_from_advoware(state, conn, service, calendar_id, kuerzel, advoware, context)
# Reload indexes after Phase 1 changes
await reload_db_indexes()
@@ -929,7 +939,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
await reload_api_maps()
# Phase 2: New from Google => Advoware
await process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware)
await process_new_from_google(state, conn, service, calendar_id, kuerzel, advoware, context)
# Reload indexes after Phase 2 changes
await reload_db_indexes()
@@ -937,7 +947,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
await reload_api_maps()
# Phase 3: Identify deleted entries
await process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware)
await process_deleted_entries(state, conn, service, calendar_id, kuerzel, advoware, context)
# Reload indexes after Phase 3 changes
await reload_db_indexes()
@@ -945,7 +955,7 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware):
await reload_api_maps()
# Phase 4: Update existing entries if changed
await process_updates(state, conn, service, calendar_id, kuerzel, advoware)
await process_updates(state, conn, service, calendar_id, kuerzel, advoware, context)
# Update last_sync timestamps
logger.debug("Updated last_sync timestamps")