Return a chat session with ordered messages.
(user_id: str, session_id: str)
| 447 | |
| 448 | |
| 449 | def get_chat_session(user_id: str, session_id: str) -> Optional[Dict[str, Any]]: |
| 450 | """Return a chat session with ordered messages.""" |
| 451 | normalized_user_id = _normalize_identifier(user_id) |
| 452 | normalized_session_id = _normalize_identifier(session_id) |
| 453 | if not normalized_user_id or not normalized_session_id: |
| 454 | raise ValueError("user_id and session_id are required") |
| 455 | |
| 456 | ensure_chat_tables() |
| 457 | conn = get_connection() |
| 458 | cursor = conn.cursor() |
| 459 | cursor.execute( |
| 460 | """ |
| 461 | SELECT session_id, user_id, title, created_at, updated_at, message_count |
| 462 | FROM chat_sessions |
| 463 | WHERE user_id = ? AND session_id = ? |
| 464 | """, |
| 465 | (normalized_user_id, normalized_session_id), |
| 466 | ) |
| 467 | session_row = cursor.fetchone() |
| 468 | if not session_row: |
| 469 | conn.close() |
| 470 | return None |
| 471 | cursor.execute( |
| 472 | """ |
| 473 | SELECT id, session_id, user_id, role, content, metadata, created_at |
| 474 | FROM chat_messages |
| 475 | WHERE user_id = ? AND session_id = ? |
| 476 | ORDER BY id ASC |
| 477 | """, |
| 478 | (normalized_user_id, normalized_session_id), |
| 479 | ) |
| 480 | messages = [] |
| 481 | for row in cursor.fetchall(): |
| 482 | message = dict(row) |
| 483 | message["metadata"] = _load_json_metadata(message.get("metadata")) |
| 484 | messages.append(message) |
| 485 | conn.close() |
| 486 | return {"session": _chat_session_row(session_row), "messages": messages} |
| 487 | |
| 488 | |
| 489 | def delete_chat_session(user_id: str, session_id: str) -> bool: |
nothing calls this directly
no test coverage detected