refactor(typing): update handler signatures to use Dict and Any for improved type hinting

This commit is contained in:
bsiggel
2026-03-08 21:24:12 +00:00
parent 2532bd89ee
commit f392ec0f06
11 changed files with 79 additions and 36 deletions

View File

@@ -17,7 +17,7 @@ from calendar_sync_utils import (
import math
import time
from datetime import datetime
from typing import Any
from typing import Any, Dict
from motia import queue, FlowContext
from pydantic import BaseModel, Field
from services.advoware_service import AdvowareService
@@ -33,7 +33,7 @@ config = {
}
async def handler(input_data: dict, ctx: FlowContext):
async def handler(input_data: Dict[str, Any], ctx: FlowContext) -> None:
"""
Handler that fetches all employees, sorts by last sync time,
and emits calendar_sync_employee events for the oldest ones.

View File

@@ -9,6 +9,7 @@ from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from calendar_sync_utils import log_operation
from typing import Dict, Any
from motia import cron, FlowContext
@@ -23,7 +24,7 @@ config = {
}
async def handler(input_data: dict, ctx: FlowContext):
async def handler(input_data: Dict[str, Any], ctx: FlowContext) -> None:
"""Cron handler that triggers the calendar sync cascade."""
try:
log_operation('info', "Calendar Sync Cron: Starting to emit sync-all event", context=ctx)
@@ -37,14 +38,6 @@ async def handler(input_data: dict, ctx: FlowContext):
})
log_operation('info', "Calendar Sync Cron: Emitted sync-all event", context=ctx)
return {
'status': 'completed',
'triggered_by': 'cron'
}
except Exception as e:
log_operation('error', f"Fehler beim Cron-Job: {e}", context=ctx)
return {
'status': 'error',
'error': str(e)
}

View File

@@ -14,6 +14,7 @@ import asyncio
import os
import datetime
from datetime import timedelta
from typing import Dict, Any
import pytz
import backoff
import time
@@ -945,14 +946,14 @@ config = {
}
async def handler(input_data: dict, ctx: FlowContext):
async def handler(input_data: Dict[str, Any], ctx: FlowContext) -> None:
"""Main event handler for calendar sync."""
start_time = time.time()
kuerzel = input_data.get('kuerzel')
if not kuerzel:
log_operation('error', "No kuerzel provided in event", context=ctx)
return {'status': 400, 'body': {'error': 'No kuerzel provided'}}
return
log_operation('info', f"Starting calendar sync for employee {kuerzel}", context=ctx)