Difyの​イテレーションで​過去の​結果を​利用する​方​法

from pathlib import Path
import pickle
from typing import Dict, Optional
import uuid
from uuid import UUID

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

import config as CONFIG


app = FastAPI()


class TextData(BaseModel):
    uuid: UUID
    idx: int
    text: str


@app.get("/temporary-id/")
async def generate_uuid():
    """
    新しいUUIDを生成して返すエンドポイント

    Returns:
        Dict[str, str]: 生成されたUUIDを含む辞書
    """
    new_uuid = str(uuid.uuid4())
    return {"uuid": new_uuid}


@app.post("/text/")
def save_text(text_data: TextData):
    """
    文字列を受け取り、辞書のリストに変換してから保存するエンドポイント

    Args:
        text_data (TextData): 受け取る文字列を含むリクエストボディ

    Returns:
        Dict[str, bool]: 処理の成否を示す辞書
    """
    text_file_service = TextCacheService(text_data.uuid)
    success = text_file_service.save_text(text_data.idx, text_data.text)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to save text")
    return {"success": success}


@app.get("/text/")
async def get_text(uuid: UUID, idx: int):
    """
    UUIDとインデックスを指定して保存されているデータを返すエンドポイント

    Args:
        uuid (UUID): ファイル名に使用するUUID
        idx (int): インデックス番号

    Returns:
        Dict[str, str]: 保存されているテキストデータ
    """
    text_file_service = TextCacheService(uuid)
    text = text_file_service.get_text(idx)
    if text is None:
        raise HTTPException(status_code=404, detail="Text not found")
    return {"text": text}


class TextCacheService:
    def __init__(self, uuid: UUID):
        """
        TextCacheServiceクラスの初期化

        Args:
            uuid (UUID): ファイル名に使用するUUID
        """
        self.file_path = Path(CONFIG.DIR_CACHE) / f"{uuid}.pkl"
        self.data = self._load_file()

    def _load_file(self) -> Dict[int, str]:
        """
        ファイルを読み込み、辞書を返す関数

        Returns:
            Dict[int, str]: ファイルから読み込んだ辞書
        """
        if self.file_path.exists():
            with self.file_path.open("rb") as file:
                return pickle.load(file)
        return {}

    def save_text(self, idx: int, text: str) -> bool:
        """
        辞書にテキストを保存し、ファイルに書き込む関数

        Args:
            idx (int): インデックス番号
            text (str): 保存するテキスト

        Returns:
            bool: 処理の成否
        """
        self.data[idx] = text
        return self._save_file()

    def get_text(self, idx: int) -> Optional[str]:
        """
        指定されたインデックスのテキストを取得する関数

        Args:
            idx (int): インデックス番号

        Returns:
            Optional[str]: 保存されているテキスト、存在しない場合はNone
        """
        return self.data.get(idx)

    def _save_file(self) -> bool:
        """
        辞書をファイルに保存する関数

        Returns:
            bool: 処理の成否
        """
        try:
            self.file_path.parent.mkdir(parents=True, exist_ok=True)
            with self.file_path.open("wb") as file:
                pickle.dump(self.data, file)
            return True
        except Exception as e:
            print(f"Error saving file: {e}")
            return False