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

Refactor and enhance logging in webhook handlers and Redis client

Browse Source 
- Translated comments and docstrings from German to English for better clarity.
- Improved logging consistency across various webhook handlers for create, delete, and update operations.
- Centralized logging functionality by utilizing a dedicated logger utility.
- Added new enums for file and XAI sync statuses in models.
- Updated Redis client factory to use a centralized logger and improved error handling.
- Enhanced API responses to include more descriptive messages and status codes.
This commit is contained in:
bsiggel
2026-03-08 21:50:34 +00:00
parent f392ec0f06
commit a0cf845877
17 changed files with 300 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
steps/advoware_cal_sync/calendar_sync_api_step.py
View File

@@ -7,7 +7,7 @@ Supports syncing a single employee or all employees.
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from calendar_sync_utils import get_redis_client, set_employee_lock, log_operation
from calendar_sync_utils import get_redis_client, set_employee_lock, get_logger
from motia import http, ApiRequest, ApiResponse, FlowContext
@@ -38,10 +38,10 @@ async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
kuerzel = body.get('kuerzel')
if not kuerzel:
return ApiResponse(
status=400,
status_code=400,
body={
'error': 'kuerzel required',
'message': 'Bitte kuerzel im Body angeben'
'message': 'Please provide kuerzel in body'
}
)
@@ -49,7 +49,7 @@ async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
if kuerzel_upper == 'ALL':
# Emit sync-all event
log_operation('info', "Calendar Sync API: Emitting sync-all event", context=ctx)
ctx.logger.info("Calendar Sync API: Emitting sync-all event")
await ctx.enqueue({
"topic": "calendar_sync_all",
"data": {
@@ -57,10 +57,10 @@ async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
}
})
return ApiResponse(
status=200,
status_code=200,
body={
'status': 'triggered',
'message': 'Calendar sync wurde für alle Mitarbeiter ausgelöst',
'message': 'Calendar sync triggered for all employees',
'triggered_by': 'api'
}
)
@@ -69,9 +69,9 @@ async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
redis_client = get_redis_client(ctx)
if not set_employee_lock(redis_client, kuerzel_upper, 'api', ctx):
log_operation('info', f"Calendar Sync API: Sync already active for {kuerzel_upper}, skipping", context=ctx)
ctx.logger.info(f"Calendar Sync API: Sync already active for {kuerzel_upper}, skipping")
return ApiResponse(
status=409,
status_code=409,
body={
'status': 'conflict',
'message': f'Calendar sync already active for {kuerzel_upper}',
@@ -80,7 +80,7 @@ async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
}
)
log_operation('info', f"Calendar Sync API called for {kuerzel_upper}", context=ctx)
ctx.logger.info(f"Calendar Sync API called for {kuerzel_upper}")
# Lock successfully set, now emit event
await ctx.enqueue({
@@ -92,19 +92,19 @@ async def handler(request: ApiRequest, ctx: FlowContext) -> ApiResponse:
})
return ApiResponse(
status=200,
status_code=200,
body={
'status': 'triggered',
'message': f'Calendar sync was triggered for {kuerzel_upper}',
'message': f'Calendar sync triggered for {kuerzel_upper}',
'kuerzel': kuerzel_upper,
'triggered_by': 'api'
}
)
except Exception as e:
log_operation('error', f"Error in API trigger: {e}", context=ctx)
ctx.logger.error(f"Error in API trigger: {e}")
return ApiResponse(
status=500,
status_code=500,
body={
'error': 'Internal server error',
'details': str(e)

70
steps/advoware_cal_sync/calendar_sync_utils.py
View File

@@ -3,50 +3,24 @@ Calendar Sync Utilities
Shared utility functions for calendar synchronization between Google Calendar and Advoware.
"""
import logging
import asyncpg
import os
import redis
import time
from typing import Optional, Any, List
from googleapiclient.discovery import build
from google.oauth2 import service_account
# Configure logging
logger = logging.getLogger(__name__)
from services.logging_utils import get_service_logger
def log_operation(level: str, message: str, context=None, **context_vars):
"""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)
full_message = f"{message} {context_str}".strip()
# Use ctx.logger if context is available (Motia III FlowContext)
if context and hasattr(context, 'logger'):
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:
# 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}")
def get_logger(context=None):
"""Get logger for calendar sync operations"""
return get_service_logger('calendar_sync', context)
async def connect_db(context=None):
"""Connect to Postgres DB from environment variables."""
logger = get_logger(context)
try:
conn = await asyncpg.connect(
host=os.getenv('POSTGRES_HOST', 'localhost'),
@@ -57,12 +31,13 @@ async def connect_db(context=None):
)
return conn
except Exception as e:
log_operation('error', f"Failed to connect to DB: {e}", context=context)
logger.error(f"Failed to connect to DB: {e}")
raise
async def get_google_service(context=None):
"""Initialize Google Calendar service."""
logger = get_logger(context)
try:
service_account_path = os.getenv('GOOGLE_CALENDAR_SERVICE_ACCOUNT_PATH', 'service-account.json')
if not os.path.exists(service_account_path):
@@ -75,48 +50,53 @@ async def get_google_service(context=None):
service = build('calendar', 'v3', credentials=creds)
return service
except Exception as e:
log_operation('error', f"Failed to initialize Google service: {e}", context=context)
logger.error(f"Failed to initialize Google service: {e}")
raise
def get_redis_client(context=None):
def get_redis_client(context=None) -> redis.Redis:
"""Initialize Redis client for calendar sync operations."""
logger = get_logger(context)
try:
redis_client = redis.Redis(
host=os.getenv('REDIS_HOST', 'localhost'),
port=int(os.getenv('REDIS_PORT', '6379')),
db=int(os.getenv('REDIS_DB_CALENDAR_SYNC', '2')),
socket_timeout=int(os.getenv('REDIS_TIMEOUT_SECONDS', '5'))
socket_timeout=int(os.getenv('REDIS_TIMEOUT_SECONDS', '5')),
decode_responses=True
)
return redis_client
except Exception as e:
log_operation('error', f"Failed to initialize Redis client: {e}", context=context)
logger.error(f"Failed to initialize Redis client: {e}")
raise
async def get_advoware_employees(advoware, context=None):
async def get_advoware_employees(advoware, context=None) -> List[Any]:
"""Fetch list of employees from Advoware."""
logger = get_logger(context)
try:
result = await advoware.api_call('api/v1/advonet/Mitarbeiter', method='GET', params={'aktiv': 'true'})
employees = result if isinstance(result, list) else []
log_operation('info', f"Fetched {len(employees)} Advoware employees", context=context)
logger.info(f"Fetched {len(employees)} Advoware employees")
return employees
except Exception as e:
log_operation('error', f"Failed to fetch Advoware employees: {e}", context=context)
logger.error(f"Failed to fetch Advoware employees: {e}")
raise
def set_employee_lock(redis_client, kuerzel: str, triggered_by: str, context=None) -> bool:
def set_employee_lock(redis_client: redis.Redis, kuerzel: str, triggered_by: str, context=None) -> bool:
"""Set lock for employee sync operation."""
logger = get_logger(context)
employee_lock_key = f'calendar_sync_lock_{kuerzel}'
if redis_client.set(employee_lock_key, triggered_by, ex=1800, nx=True) is None:
log_operation('info', f"Sync already active for {kuerzel}, skipping", context=context)
logger.info(f"Sync already active for {kuerzel}, skipping")
return False
return True
def clear_employee_lock(redis_client, kuerzel: str, context=None):
def clear_employee_lock(redis_client: redis.Redis, kuerzel: str, context=None) -> None:
"""Clear lock for employee sync operation and update last-synced timestamp."""
logger = get_logger(context)
try:
employee_lock_key = f'calendar_sync_lock_{kuerzel}'
employee_last_synced_key = f'calendar_sync_last_synced_{kuerzel}'
@@ -128,6 +108,6 @@ def clear_employee_lock(redis_client, kuerzel: str, context=None):
# Delete the lock
redis_client.delete(employee_lock_key)
log_operation('debug', f"Cleared lock and updated last-synced for {kuerzel} to {current_time}", context=context)
logger.debug(f"Cleared lock and updated last-synced for {kuerzel} to {current_time}")
except Exception as e:
log_operation('warning', f"Failed to clear lock and update last-synced for {kuerzel}: {e}", context=context)
logger.warning(f"Failed to clear lock and update last-synced for {kuerzel}: {e}")