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

feat(webhooks): Update Akte webhook handlers to trigger immediate synchronization

Browse Source
This commit is contained in:
bsiggel
2026-03-26 10:16:33 +00:00
parent bf02b1a4e1
commit 52114a3c95
2 changed files with 10 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/steps/crm/akte/webhooks/akte_create_api_step.py
View File

@@ -1,21 +1,17 @@
"""Akte Webhook - Create""" """Akte Webhook - Create"""
import json import json
import time
import datetime
from typing import Any from typing import Any
from motia import FlowContext, http, ApiRequest, ApiResponse from motia import FlowContext, http, ApiRequest, ApiResponse
config = { config = {
"name": "Akte Webhook - Create", "name": "Akte Webhook - Create",
"description": "Empfängt EspoCRM-Create-Webhooks für CAkten und queued Entity-IDs für den Sync", "description": "Empfängt EspoCRM-Create-Webhooks für CAkten und triggert sofort den Sync",
"flows": ["akte-sync"], "flows": ["akte-sync"],
"triggers": [http("POST", "/crm/akte/webhook/create")], "triggers": [http("POST", "/crm/akte/webhook/create")],
"enqueues": [], "enqueues": ["akte.sync"],
} }
PENDING_KEY = "akte:pending_entity_ids"
async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse: async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
try: try:
@@ -37,16 +33,10 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
ctx.logger.warn("⚠️ No entity IDs in payload") ctx.logger.warn("⚠️ No entity IDs in payload")
return ApiResponse(status_code=400, body={"error": "No entity ID found in payload"}) return ApiResponse(status_code=400, body={"error": "No entity ID found in payload"})
from services.redis_client import get_redis_client for eid in entity_ids:
redis_client = get_redis_client(strict=False) await ctx.enqueue({'topic': 'akte.sync', 'data': {'akte_id': eid, 'aktennummer': None}})
if not redis_client:
ctx.logger.error("❌ Redis unavailable")
return ApiResponse(status_code=503, body={"error": "Service unavailable"})
ts = time.time() ctx.logger.info(f"✅ Emitted akte.sync for {len(entity_ids)} ID(s): {entity_ids}")
redis_client.zadd(PENDING_KEY, {eid: ts for eid in entity_ids})
ctx.logger.info(f"✅ Queued {len(entity_ids)} entity ID(s): {entity_ids}")
ctx.logger.info("=" * 60) ctx.logger.info("=" * 60)
return ApiResponse(status_code=200, body={"status": "received", "action": "create", "ids_count": len(entity_ids)}) return ApiResponse(status_code=200, body={"status": "received", "action": "create", "ids_count": len(entity_ids)})

20
src/steps/crm/akte/webhooks/akte_update_api_step.py
View File

@@ -1,21 +1,17 @@
"""Akte Webhook - Update""" """Akte Webhook - Update"""
import json import json
import time
import datetime
from typing import Any from typing import Any
from motia import FlowContext, http, ApiRequest, ApiResponse from motia import FlowContext, http, ApiRequest, ApiResponse
config = { config = {
"name": "Akte Webhook - Update", "name": "Akte Webhook - Update",
"description": "Empfängt EspoCRM-Update-Webhooks für CAkten und queued Entity-IDs für den Sync", "description": "Empfängt EspoCRM-Update-Webhooks für CAkten und triggert sofort den Sync",
"flows": ["akte-sync"], "flows": ["akte-sync"],
"triggers": [http("POST", "/crm/akte/webhook/update")], "triggers": [http("POST", "/crm/akte/webhook/update")],
"enqueues": [], "enqueues": ["akte.sync"],
} }
PENDING_KEY = "akte:pending_entity_ids"
async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse: async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
try: try:
@@ -37,16 +33,10 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
ctx.logger.warn("⚠️ No entity IDs in payload") ctx.logger.warn("⚠️ No entity IDs in payload")
return ApiResponse(status_code=400, body={"error": "No entity ID found in payload"}) return ApiResponse(status_code=400, body={"error": "No entity ID found in payload"})
from services.redis_client import get_redis_client for eid in entity_ids:
redis_client = get_redis_client(strict=False) await ctx.enqueue({'topic': 'akte.sync', 'data': {'akte_id': eid, 'aktennummer': None}})
if not redis_client:
ctx.logger.error("❌ Redis unavailable")
return ApiResponse(status_code=503, body={"error": "Service unavailable"})
ts = time.time() ctx.logger.info(f"✅ Emitted akte.sync for {len(entity_ids)} ID(s): {entity_ids}")
redis_client.zadd(PENDING_KEY, {eid: ts for eid in entity_ids})
ctx.logger.info(f"✅ Queued {len(entity_ids)} entity ID(s): {entity_ids}")
ctx.logger.info("=" * 60) ctx.logger.info("=" * 60)
return ApiResponse(status_code=200, body={"status": "received", "action": "update", "ids_count": len(entity_ids)}) return ApiResponse(status_code=200, body={"status": "received", "action": "update", "ids_count": len(entity_ids)})