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

Refactor logging in log_operation to support context-based logging and fallback to standard logger

Browse Source
This commit is contained in:
bsiggel
2026-03-01 23:15:14 +00:00
parent 0b8da01b71
commit dc586385df
1 changed files with 24 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
steps/advoware_cal_sync/calendar_sync_utils.py
View File

@@ -18,20 +18,31 @@ logger = logging.getLogger(__name__)
def log_operation(level: str, message: str, context=None, **context_vars): def log_operation(level: str, message: str, context=None, **context_vars):
"""Centralized logging with context, supporting file and console logging.""" """Centralized logging with context, supporting file and console logging."""
context_str = ' '.join(f"{k}={v}" for k, v in context_vars.items() if v is not None) context_str = ' '.join(f"{k}={v}" for k, v in context_vars.items() if v is not None)
full_message = f"[{time.time()}] {message} {context_str}".strip() full_message = f"{message} {context_str}".strip()
# Log via logger # Use ctx.logger if context is available (Motia III FlowContext)
if level == 'info': if context and hasattr(context, 'logger'):
logger.info(full_message) if level == 'info':
elif level == 'warning': context.logger.info(full_message)
logger.warning(full_message) elif level == 'warning':
elif level == 'error': context.logger.warning(full_message)
logger.error(full_message) elif level == 'error':
elif level == 'debug': context.logger.error(full_message)
logger.debug(full_message) elif level == 'debug':
context.logger.debug(full_message)
# Also log to console for journalctl visibility else:
print(f"[{level.upper()}] {full_message}") # Fallback to standard logger
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)
# Also log to console for journalctl visibility
print(f"[{level.upper()}] {full_message}")
async def connect_db(context=None): async def connect_db(context=None):