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

refactor(logging): remove unused logger instances and enhance error logging in webhook steps

Browse Source
This commit is contained in:
bsiggel
2026-03-08 22:21:08 +00:00
parent 1c765d1eec
commit 6c3cf3ca91
14 changed files with 72 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
services/adressen_mapper.py
View File

@@ -7,9 +7,6 @@ Basierend auf ADRESSEN_SYNC_ANALYSE.md Abschnitt 12.
from typing import Dict, Any, Optional
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class AdressenMapper:

2
services/adressen_sync.py
View File

@@ -26,8 +26,6 @@ from services.espocrm import EspoCRMAPI
from services.adressen_mapper import AdressenMapper
from services.notification_utils import NotificationManager
logger = logging.getLogger(__name__)
class AdressenSync:
"""Sync-Klasse für Adressen zwischen EspoCRM und Advoware"""

70
services/advoware.py
View File

@@ -72,6 +72,11 @@ class AdvowareAPI:
self._session: Optional[aiohttp.ClientSession] = None
def _log(self, message: str, level: str = 'info') -> None:
"""Internal logging helper"""
log_func = getattr(self.logger, level, self.logger.info)
log_func(message)
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession()
@@ -98,8 +103,8 @@ class AdvowareAPI:
signature = hmac.new(api_key_bytes, message, hashlib.sha512)
return base64.b64encode(signature.digest()).decode('utf-8')
def _fetch_new_access_token(self) -> str:
"""Fetch new access token from Advoware Auth API"""
async def _fetch_new_access_token(self) -> str:
"""Fetch new access token from Advoware Auth API (async)"""
self.logger.info("Fetching new access token from Advoware")
nonce = str(uuid.uuid4())
@@ -122,40 +127,41 @@ class AdvowareAPI:
self.logger.debug(f"Token request: AppID={self.app_id}, User={self.user}")
# Using synchronous requests for token fetch (called from sync context)
# TODO: Convert to async in future version
import requests
# Async token fetch using aiohttp
session = await self._get_session()
try:
response = requests.post(
async with session.post(
ADVOWARE_CONFIG.auth_url,
json=data,
headers=headers,
timeout=self.api_timeout_seconds
)
self.logger.debug(f"Token response status: {response.status_code}")
if response.status_code == 401:
raise AdvowareAuthError(
"Authentication failed - check credentials",
status_code=401
)
response.raise_for_status()
except requests.Timeout:
timeout=aiohttp.ClientTimeout(total=self.api_timeout_seconds)
) as response:
self.logger.debug(f"Token response status: {response.status}")
if response.status == 401:
raise AdvowareAuthError(
"Authentication failed - check credentials",
status_code=401
)
if response.status >= 400:
error_text = await response.text()
raise AdvowareAPIError(
f"Token request failed ({response.status}): {error_text}",
status_code=response.status
)
result = await response.json()
except asyncio.TimeoutError:
raise AdvowareTimeoutError(
"Token request timed out",
status_code=408
)
except requests.RequestException as e:
raise AdvowareAPIError(
f"Token request failed: {str(e)}",
status_code=getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None
)
except aiohttp.ClientError as e:
raise AdvowareAPIError(f"Token request failed: {str(e)}")
result = response.json()
access_token = result.get("access_token")
if not access_token:
@@ -173,7 +179,7 @@ class AdvowareAPI:
return access_token
def get_access_token(self, force_refresh: bool = False) -> str:
async def get_access_token(self, force_refresh: bool = False) -> str:
"""
Get valid access token (from cache or fetch new).
@@ -187,11 +193,11 @@ class AdvowareAPI:
if not self.redis_client:
self.logger.info("No Redis available, fetching new token")
return self._fetch_new_access_token()
return await self._fetch_new_access_token()
if force_refresh:
self.logger.info("Force refresh requested, fetching new token")
return self._fetch_new_access_token()
return await self._fetch_new_access_token()
# Check cache
cached_token = self.redis_client.get(ADVOWARE_CONFIG.token_cache_key)
@@ -210,7 +216,7 @@ class AdvowareAPI:
self.logger.debug(f"Error reading cached token: {e}")
self.logger.info("Cached token expired or invalid, fetching new")
return self._fetch_new_access_token()
return await self._fetch_new_access_token()
async def api_call(
self,
@@ -254,7 +260,7 @@ class AdvowareAPI:
# Get auth token
try:
token = self.get_access_token()
token = await self.get_access_token()
except AdvowareAuthError:
raise
except Exception as e:
@@ -282,7 +288,7 @@ class AdvowareAPI:
# Handle 401 - retry with fresh token
if response.status == 401:
self.logger.warning("401 Unauthorized, refreshing token")
token = self.get_access_token(force_refresh=True)
token = await self.get_access_token(force_refresh=True)
effective_headers['Authorization'] = f'Bearer {token}'
async with session.request(

3
services/bankverbindungen_mapper.py
View File

@@ -6,9 +6,6 @@ Transformiert Bankverbindungen zwischen den beiden Systemen
from typing import Dict, Any, Optional, List
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class BankverbindungenMapper:

21
services/document_sync_utils.py
View File

@@ -20,6 +20,12 @@ MAX_SYNC_RETRIES = 5
# Retry backoff: Wartezeit zwischen Retries (in Minuten)
RETRY_BACKOFF_MINUTES = [1, 5, 15, 60, 240] # 1min, 5min, 15min, 1h, 4h
# Legacy file status values (for backward compatibility)
# These are old German and English status values that may still exist in the database
LEGACY_NEW_STATUS_VALUES = {'neu', 'Neu', 'New'}
LEGACY_CHANGED_STATUS_VALUES = {'geändert', 'Geändert', 'Changed'}
LEGACY_SYNCED_STATUS_VALUES = {'synced', 'Synced', 'synchronized', 'Synchronized'}
class DocumentSync(BaseSyncUtils):
"""Utility class for document synchronization with xAI"""
@@ -185,16 +191,11 @@ class DocumentSync(BaseSyncUtils):
# ═══════════════════════════════════════════════════════════════
# PRIORITY CHECK 2: fileStatus "new" or "changed"
# ═══════════════════════════════════════════════════════════════
if datei_status in [
FileStatus.NEW.value,
FileStatus.CHANGED.value,
'neu', # Legacy German values
'geändert', # Legacy German values
'Neu', # Case variations
'Geändert',
'New',
'Changed'
]:
# Check for standard enum values and legacy values
is_new = (datei_status == FileStatus.NEW.value or datei_status in LEGACY_NEW_STATUS_VALUES)
is_changed = (datei_status == FileStatus.CHANGED.value or datei_status in LEGACY_CHANGED_STATUS_VALUES)
if is_new or is_changed:
self._log(f"🆕 fileStatus: '{datei_status}' → xAI sync REQUIRED")
if target_collections:

2
services/espocrm.py
View File

@@ -17,8 +17,6 @@ from services.redis_client import get_redis_client
from services.config import ESPOCRM_CONFIG, API_CONFIG
from services.logging_utils import get_service_logger
logger = logging.getLogger(__name__)
class EspoCRMAPI:
"""

2
services/espocrm_mapper.py
View File

@@ -18,8 +18,6 @@ from services.models import (
from services.exceptions import ValidationError
from services.config import FEATURE_FLAGS
logger = logging.getLogger(__name__)
class BeteiligteMapper:
"""Mapper für CBeteiligte (EspoCRM) ↔ Beteiligte (Advoware)"""

2
services/kommunikation_sync_utils.py
View File

@@ -24,8 +24,6 @@ from services.kommunikation_mapper import (
from services.advoware_service import AdvowareService
from services.espocrm import EspoCRMAPI
logger = logging.getLogger(__name__)
class KommunikationSyncManager:
"""Manager für Kommunikation-Synchronisation"""

1
steps/advoware_proxy/advoware_api_proxy_post_step.py
View File

@@ -69,4 +69,3 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
status=500,
body={'error': 'Internal server error', 'details': str(e)}
)
)

4
steps/vmh/webhook/beteiligte_create_api_step.py
View File

@@ -71,7 +71,11 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
except Exception as e:
ctx.logger.error("=" * 80)
ctx.logger.error("❌ ERROR: VMH CREATE WEBHOOK")
ctx.logger.error("=" * 80)
ctx.logger.error(f"Error: {e}")
ctx.logger.error(f"Entity IDs attempted: {list(entity_ids) if 'entity_ids' in locals() else 'N/A'}")
ctx.logger.error(f"Full Payload: {json.dumps(request.body, indent=2, ensure_ascii=False)}")
ctx.logger.error(f"Timestamp: {datetime.datetime.now().isoformat()}")
ctx.logger.error("=" * 80)
return ApiResponse(
status=500,

4
steps/vmh/webhook/beteiligte_update_api_step.py
View File

@@ -71,7 +71,11 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
except Exception as e:
ctx.logger.error("=" * 80)
ctx.logger.error("❌ ERROR: VMH UPDATE WEBHOOK")
ctx.logger.error("=" * 80)
ctx.logger.error(f"Error: {e}")
ctx.logger.error(f"Entity IDs attempted: {list(entity_ids) if 'entity_ids' in locals() else 'N/A'}")
ctx.logger.error(f"Full Payload: {json.dumps(request.body, indent=2, ensure_ascii=False)}")
ctx.logger.error(f"Timestamp: {datetime.datetime.now().isoformat()}")
ctx.logger.error("=" * 80)
return ApiResponse(
status=500,

6
steps/vmh/webhook/document_create_api_step.py
View File

@@ -1,5 +1,6 @@
"""VMH Webhook - Document Create"""
import json
import datetime
from typing import Any
from motia import FlowContext, http, ApiRequest, ApiResponse
@@ -74,8 +75,11 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
except Exception as e:
ctx.logger.error("=" * 80)
ctx.logger.error("❌ ERROR: DOCUMENT CREATE WEBHOOK")
ctx.logger.error("=" * 80)
ctx.logger.error(f"Error: {e}")
ctx.logger.error(f"Payload: {request.body}")
ctx.logger.error(f"Entity IDs attempted: {list(entity_ids) if 'entity_ids' in locals() else 'N/A'}")
ctx.logger.error(f"Full Payload: {json.dumps(request.body, indent=2, ensure_ascii=False)}")
ctx.logger.error(f"Timestamp: {datetime.datetime.now().isoformat()}")
ctx.logger.error("=" * 80)
return ApiResponse(

6
steps/vmh/webhook/document_delete_api_step.py
View File

@@ -1,5 +1,6 @@
"""VMH Webhook - Document Delete"""
import json
import datetime
from typing import Any
from motia import FlowContext, http, ApiRequest, ApiResponse
@@ -74,8 +75,11 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
except Exception as e:
ctx.logger.error("=" * 80)
ctx.logger.error("❌ ERROR: DOCUMENT DELETE WEBHOOK")
ctx.logger.error("=" * 80)
ctx.logger.error(f"Error: {e}")
ctx.logger.error(f"Payload: {request.body}")
ctx.logger.error(f"Entity IDs attempted: {list(entity_ids) if 'entity_ids' in locals() else 'N/A'}")
ctx.logger.error(f"Full Payload: {json.dumps(request.body, indent=2, ensure_ascii=False)}")
ctx.logger.error(f"Timestamp: {datetime.datetime.now().isoformat()}")
ctx.logger.error("=" * 80)
return ApiResponse(

6
steps/vmh/webhook/document_update_api_step.py
View File

@@ -1,5 +1,6 @@
"""VMH Webhook - Document Update"""
import json
import datetime
from typing import Any
from motia import FlowContext, http, ApiRequest, ApiResponse
@@ -74,8 +75,11 @@ async def handler(request: ApiRequest, ctx: FlowContext[Any]) -> ApiResponse:
except Exception as e:
ctx.logger.error("=" * 80)
ctx.logger.error("❌ ERROR: DOCUMENT UPDATE WEBHOOK")
ctx.logger.error("=" * 80)
ctx.logger.error(f"Error: {e}")
ctx.logger.error(f"Payload: {request.body}")
ctx.logger.error(f"Entity IDs attempted: {list(entity_ids) if 'entity_ids' in locals() else 'N/A'}")
ctx.logger.error(f"Full Payload: {json.dumps(request.body, indent=2, ensure_ascii=False)}")
ctx.logger.error(f"Timestamp: {datetime.datetime.now().isoformat()}")
ctx.logger.error("=" * 80)
return ApiResponse(