va/int-layer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix lint errors

Browse Source
This commit is contained in:
Anibal Angulo
2026-02-20 04:59:56 +00:00
committed by Anibal Angulo
parent 52a674959c
commit faf1f13049
19 changed files with 442 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
src/capa_de_integracion/services/redis_service.py
View File

@@ -1,11 +1,13 @@
"""Redis service for caching conversation sessions and notifications."""
import json
import logging
from datetime import datetime
from datetime import UTC, datetime
from redis.asyncio import Redis
from capa_de_integracion.config import Settings
from capa_de_integracion.models import ConversationSession
from capa_de_integracion.models import ConversationEntry, ConversationSession
from capa_de_integracion.models.notification import Notification, NotificationSession
logger = logging.getLogger(__name__)
@@ -31,7 +33,9 @@ class RedisService:
decode_responses=True,
)
logger.info(
f"Connected to Redis at {self.settings.redis_host}:{self.settings.redis_port}",
"Connected to Redis at %s:%s",
self.settings.redis_host,
self.settings.redis_port,
)
async def close(self) -> None:
@@ -66,29 +70,26 @@ class RedisService:
phone_key = self._phone_to_session_key(session_id_or_phone)
mapped_session_id = await self.redis.get(phone_key)
if mapped_session_id:
# Found phone mapping, get the actual session
session_id = mapped_session_id
else:
# Try as direct session ID
session_id = session_id_or_phone
# Use mapped session ID if found, otherwise use input directly
session_id = mapped_session_id or session_id_or_phone
# Get session by ID
key = self._session_key(session_id)
data = await self.redis.get(key)
if not data:
logger.debug(f"Session not found in Redis: {session_id_or_phone}")
logger.debug("Session not found in Redis: %s", session_id_or_phone)
return None
try:
session_dict = json.loads(data)
session = ConversationSession.model_validate(session_dict)
logger.debug(f"Retrieved session from Redis: {session_id}")
return session
except Exception as e:
logger.exception(f"Error deserializing session {session_id}: {e!s}")
logger.debug("Retrieved session from Redis: %s", session_id)
except Exception:
logger.exception("Error deserializing session %s:", session_id)
return None
else:
return session
async def save_session(self, session: ConversationSession) -> bool:
"""Save conversation session to Redis with TTL.
@@ -99,7 +100,7 @@ class RedisService:
msg = "Redis client not connected"
raise RuntimeError(msg)
key = self._session_key(session.sessionId)
key = self._session_key(session.session_id)
phone_key = self._phone_to_session_key(session.telefono)
try:
@@ -108,15 +109,18 @@ class RedisService:
await self.redis.setex(key, self.session_ttl, data)
# Save phone-to-session mapping
await self.redis.setex(phone_key, self.session_ttl, session.sessionId)
await self.redis.setex(phone_key, self.session_ttl, session.session_id)
logger.debug(
f"Saved session to Redis: {session.sessionId} for phone: {session.telefono}",
"Saved session to Redis: %s for phone: %s",
session.session_id,
session.telefono,
)
return True
except Exception as e:
logger.exception(f"Error saving session {session.sessionId} to Redis: {e!s}")
except Exception:
logger.exception("Error saving session %s to Redis:", session.session_id)
return False
else:
return True
async def delete_session(self, session_id: str) -> bool:
"""Delete conversation session from Redis."""
@@ -128,11 +132,12 @@ class RedisService:
try:
result = await self.redis.delete(key)
logger.debug(f"Deleted session from Redis: {session_id}")
return result > 0
except Exception as e:
logger.exception(f"Error deleting session {session_id} from Redis: {e!s}")
logger.debug("Deleted session from Redis: %s", session_id)
except Exception:
logger.exception("Error deleting session %s from Redis:", session_id)
return False
else:
return result > 0
async def exists(self, session_id: str) -> bool:
"""Check if session exists in Redis."""
@@ -149,7 +154,7 @@ class RedisService:
"""Generate Redis key for conversation messages."""
return f"conversation:messages:{session_id}"
async def save_message(self, session_id: str, message) -> bool:
async def save_message(self, session_id: str, message: ConversationEntry) -> bool:
"""Save a conversation message to Redis sorted set.
Messages are stored in a sorted set with timestamp as score.
@@ -179,13 +184,15 @@ class RedisService:
# Set TTL on the messages key to match session TTL
await self.redis.expire(key, self.session_ttl)
logger.debug(f"Saved message to Redis: {session_id}")
return True
except Exception as e:
logger.debug("Saved message to Redis: %s", session_id)
except Exception:
logger.exception(
f"Error saving message to Redis for session {session_id}: {e!s}",
"Error saving message to Redis for session %s:",
session_id,
)
return False
else:
return True
async def get_messages(self, session_id: str) -> list:
"""Retrieve all conversation messages for a session from Redis.
@@ -210,7 +217,7 @@ class RedisService:
message_strings = await self.redis.zrange(key, 0, -1)
if not message_strings:
logger.debug(f"No messages found in Redis for session: {session_id}")
logger.debug("No messages found in Redis for session: %s", session_id)
return []
# Parse JSON strings to dictionaries
@@ -218,19 +225,23 @@ class RedisService:
for msg_str in message_strings:
try:
messages.append(json.loads(msg_str))
except json.JSONDecodeError as e:
logger.exception(f"Error parsing message JSON: {e!s}")
except json.JSONDecodeError:
logger.exception("Error parsing message JSON:")
continue
logger.debug(
f"Retrieved {len(messages)} messages from Redis for session: {session_id}",
"Retrieved %s messages from Redis for session: %s",
len(messages),
session_id,
)
return messages
except Exception as e:
except Exception:
logger.exception(
f"Error retrieving messages from Redis for session {session_id}: {e!s}",
"Error retrieving messages from Redis for session %s:",
session_id,
)
return []
else:
return messages
# ====== Notification Methods ======
@@ -273,17 +284,17 @@ class RedisService:
updated_session = NotificationSession(
sessionId=notification_session_id,
telefono=phone_number,
fechaCreacion=existing_session.fechaCreacion,
ultimaActualizacion=datetime.now(),
fechaCreacion=existing_session.fecha_creacion,
ultimaActualizacion=datetime.now(UTC),
notificaciones=updated_notifications,
)
else:
# Create new session
updated_session = NotificationSession(
sessionId=notification_session_id,
session_id=notification_session_id,
telefono=phone_number,
fechaCreacion=datetime.now(),
ultimaActualizacion=datetime.now(),
fecha_creacion=datetime.now(UTC),
ultima_actualizacion=datetime.now(UTC),
notificaciones=[new_entry],
)
@@ -296,7 +307,7 @@ class RedisService:
msg = "Redis client not connected"
raise RuntimeError(msg)
key = self._notification_key(session.sessionId)
key = self._notification_key(session.session_id)
phone_key = self._phone_to_notification_key(session.telefono)
try:
@@ -305,15 +316,17 @@ class RedisService:
await self.redis.setex(key, self.notification_ttl, data)
# Save phone-to-session mapping
await self.redis.setex(phone_key, self.notification_ttl, session.sessionId)
await self.redis.setex(phone_key, self.notification_ttl, session.session_id)
logger.debug(f"Cached notification session: {session.sessionId}")
return True
except Exception as e:
logger.debug("Cached notification session: %s", session.session_id)
except Exception:
logger.exception(
f"Error caching notification session {session.sessionId}: {e!s}",
"Error caching notification session %s:",
session.session_id,
)
return False
else:
return True
async def get_notification_session(
self, session_id: str,
@@ -327,19 +340,21 @@ class RedisService:
data = await self.redis.get(key)
if not data:
logger.debug(f"Notification session not found in Redis: {session_id}")
logger.debug("Notification session not found in Redis: %s", session_id)
return None
try:
session_dict = json.loads(data)
session = NotificationSession.model_validate(session_dict)
logger.info(f"Notification session {session_id} retrieved from Redis")
return session
except Exception as e:
logger.info("Notification session %s retrieved from Redis", session_id)
except Exception:
logger.exception(
f"Error deserializing notification session {session_id}: {e!s}",
"Error deserializing notification session %s:",
session_id,
)
return None
else:
return session
async def get_notification_id_for_phone(self, phone: str) -> str | None:
"""Get notification session ID for a phone number."""
@@ -351,7 +366,7 @@ class RedisService:
session_id = await self.redis.get(key)
if session_id:
logger.info(f"Session ID {session_id} found for phone")
logger.info("Session ID %s found for phone", session_id)
else:
logger.debug("Session ID not found for phone")
@@ -367,12 +382,14 @@ class RedisService:
phone_key = self._phone_to_notification_key(phone_number)
try:
logger.info(f"Deleting notification session for phone {phone_number}")
logger.info("Deleting notification session for phone %s", phone_number)
await self.redis.delete(notification_key)
await self.redis.delete(phone_key)
return True
except Exception as e:
except Exception:
logger.exception(
f"Error deleting notification session for phone {phone_number}: {e!s}",
"Error deleting notification session for phone %s:",
phone_number,
)
return False
else:
return True