def find_conversation_by_id()

in packages/blueprints/gen-ai-chatbot/static-assets/chatbot-genai-components/backend/python/app/repositories/conversation.py [0:0]

• 81 lines of code
• 11 McCabe index (conditional complexity)

def find_conversation_by_id(user_id: str, conversation_id: str) -> ConversationModel:
    logger.info(f"Finding conversation: {conversation_id}")
    table = _get_table_client(user_id)
    response = table.query(
        IndexName="SKIndex",
        KeyConditionExpression=Key("SK").eq(compose_conv_id(user_id, conversation_id)),
    )
    if len(response["Items"]) == 0:
        raise RecordNotFoundError(f"No conversation found with id: {conversation_id}")

    # NOTE: conversation is unique
    item = response["Items"][0]
    if item.get("IsLargeMessage", False):
        large_message_path = item["LargeMessagePath"]
        response = s3_client.get_object(
            Bucket=LARGE_MESSAGE_BUCKET, Key=large_message_path
        )
        message_map = json.loads(response["Body"].read().decode("utf-8"))
    else:
        message_map = json.loads(item["MessageMap"])

    conv = ConversationModel(
        id=decompose_conv_id(item["SK"]),
        create_time=float(item["CreateTime"]),
        title=item["Title"],
        total_price=item.get("TotalPrice", 0),
        message_map={
            k: MessageModel(
                role=v["role"],
                content=(
                    [
                        ContentModel(
                            content_type=c["content_type"],
                            body=c["body"],
                            media_type=c["media_type"],
                        )
                        for c in v["content"]
                    ]
                    if type(v["content"]) == list
                    else [
                        # For backward compatibility
                        ContentModel(
                            content_type=v["content"]["content_type"],
                            body=v["content"]["body"],
                            media_type=None,
                        )
                    ]
                ),
                model=v["model"],
                children=v["children"],
                parent=v["parent"],
                create_time=float(v["create_time"]),
                feedback=(
                    FeedbackModel(
                        thumbs_up=v["feedback"]["thumbs_up"],
                        category=v["feedback"]["category"],
                        comment=v["feedback"]["comment"],
                    )
                    if v.get("feedback")
                    else None
                ),
                used_chunks=(
                    [
                        ChunkModel(
                            content=c["content"],
                            content_type=(
                                c["content_type"] if "content_type" in c else "s3"
                            ),
                            source=c["source"],
                            rank=c["rank"],
                        )
                        for c in v["used_chunks"]
                    ]
                    if v.get("used_chunks")
                    else None
                ),
                thinking_log=v.get("thinking_log"),
            )
            for k, v in message_map.items()
        },
        last_message_id=item["LastMessageId"],
        bot_id=item["BotId"] if "BotId" in item else None,
    )
    logger.info(f"Found conversation: {conv}")
    return conv