Add Prompt dataclass with initial methods

scripts/api_generator.py

@@ -44,6 +44,8 @@
     "/e2e_fetch_trace/",
     "/e2e_fetch_span_score/",
     "/e2e_fetch_trace_scorer_span_score/",
+    "/prompts/insert/",
+    "/prompts/fetch/",
 ]
 
 

scripts/openapi_transform.py

@@ -42,6 +42,8 @@
     "/projects/resolve/",
     "/e2e_fetch_trace/",
     "/e2e_fetch_span_score/",
+    "/prompts/insert/",
+    "/prompts/fetch/",
 ]
 
 

src/judgeval/api/__init__.py

@@ -199,6 +199,28 @@ def upload_custom_scorer(
             payload,
         )
 
+    def prompts_insert(self, payload: PromptInsertRequest) -> PromptInsertResponse:
+        return self._request(
+            "POST",
+            url_for("/prompts/insert/"),
+            payload,
+        )
+
+    def prompts_fetch(
+        self, name: str, commit_id: Optional[str] = None, tag: Optional[str] = None
+    ) -> PromptFetchResponse:
+        query_params = {}
+        query_params["name"] = name
+        if commit_id is not None:
+            query_params["commit_id"] = commit_id
+        if tag is not None:
+            query_params["tag"] = tag
+        return self._request(
+            "GET",
+            url_for("/prompts/fetch/"),
+            query_params,
+        )
+
     def projects_resolve(
         self, payload: ResolveProjectNameRequest
     ) -> ResolveProjectNameResponse:
@@ -410,6 +432,30 @@ async def upload_custom_scorer(
             payload,
         )
 
+    async def prompts_insert(
+        self, payload: PromptInsertRequest
+    ) -> PromptInsertResponse:
+        return await self._request(
+            "POST",
+            url_for("/prompts/insert/"),
+            payload,
+        )
+
+    async def prompts_fetch(
+        self, name: str, commit_id: Optional[str] = None, tag: Optional[str] = None
+    ) -> PromptFetchResponse:
+        query_params = {}
+        query_params["name"] = name
+        if commit_id is not None:
+            query_params["commit_id"] = commit_id
+        if tag is not None:
+            query_params["tag"] = tag
+        return await self._request(
+            "GET",
+            url_for("/prompts/fetch/"),
+            query_params,
+        )
+
     async def projects_resolve(
         self, payload: ResolveProjectNameRequest
     ) -> ResolveProjectNameResponse:

src/judgeval/api/api_types.py

@@ -1,6 +1,6 @@
 # generated by datamodel-codegen:
 #   filename:  .openapi.json
-#   timestamp: 2025-09-12T16:54:35+00:00
+#   timestamp: 2025-09-17T19:04:58+00:00
 
 from __future__ import annotations
 from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
@@ -77,6 +77,26 @@ class CustomScorerTemplateResponse(TypedDict):
     message: str
 
 
+class PromptInsertRequest(TypedDict):
+    name: str
+    prompt: str
+    tags: List[str]
+
+
+class PromptInsertResponse(TypedDict):
+    commit_id: str
+    parent_commit_id: NotRequired[Optional[str]]
+
+
+class PromptFetchResponse(TypedDict):
+    name: str
+    prompt: str
+    tags: List[str]
+    commit_id: str
+    parent_commit_id: NotRequired[Optional[str]]
+    created_at: str
+
+
 class ResolveProjectNameRequest(TypedDict):
     project_name: str
 

src/judgeval/data/judgment_types.py

@@ -1,6 +1,6 @@
 # generated by datamodel-codegen:
 #   filename:  .openapi.json
-#   timestamp: 2025-09-12T16:54:34+00:00
+#   timestamp: 2025-09-17T19:04:57+00:00
 
 from __future__ import annotations
 from typing import Annotated, Any, Dict, List, Optional, Union
@@ -79,6 +79,26 @@ class CustomScorerTemplateResponse(BaseModel):
     message: Annotated[str, Field(title="Message")]
 
 
+class PromptInsertRequest(BaseModel):
+    name: Annotated[str, Field(title="Name")]
+    prompt: Annotated[str, Field(title="Prompt")]
+    tags: Annotated[List[str], Field(title="Tags")]
+
+
+class PromptInsertResponse(BaseModel):
+    commit_id: Annotated[str, Field(title="Commit Id")]
+    parent_commit_id: Annotated[Optional[str], Field(title="Parent Commit Id")] = None
+
+
+class PromptFetchResponse(BaseModel):
+    name: Annotated[str, Field(title="Name")]
+    prompt: Annotated[str, Field(title="Prompt")]
+    tags: Annotated[List[str], Field(title="Tags")]
+    commit_id: Annotated[str, Field(title="Commit Id")]
+    parent_commit_id: Annotated[Optional[str], Field(title="Parent Commit Id")] = None
+    created_at: Annotated[str, Field(title="Created At")]
+
+
 class ResolveProjectNameRequest(BaseModel):
     project_name: Annotated[str, Field(title="Project Name")]
 

src/judgeval/prompts/prompt.py

@@ -0,0 +1,96 @@
+from typing import List, Optional
+import os
+from judgeval.api import JudgmentSyncClient
+from judgeval.exceptions import JudgmentAPIError
+from dataclasses import dataclass, field
+import re
+from string import Template
+
+
+def push_prompt(
+    name: str,
+    prompt: str,
+    tags: List[str],
+    judgment_api_key: str = os.getenv("JUDGMENT_API_KEY") or "",
+    organization_id: str = os.getenv("JUDGMENT_ORG_ID") or "",
+) -> tuple[str, Optional[str]]:
+    client = JudgmentSyncClient(judgment_api_key, organization_id)
+    try:
+        r = client.prompts_insert(
+            payload={"name": name, "prompt": prompt, "tags": tags}
+        )
+        return r["commit_id"], r["parent_commit_id"]
+    except JudgmentAPIError as e:
+        raise JudgmentAPIError(
+            status_code=e.status_code,
+            detail=f"Failed to save prompt: {e.detail}",
+            response=e.response,
+        )
+
+
+def fetch_prompt(
+    name: str,
+    commit_id: Optional[str] = None,
+    tag: Optional[str] = None,
+    judgment_api_key: str = os.getenv("JUDGMENT_API_KEY") or "",
+    organization_id: str = os.getenv("JUDGMENT_ORG_ID") or "",
+):
+    client = JudgmentSyncClient(judgment_api_key, organization_id)
+    try:
+        prompt_config = client.prompts_fetch(name, commit_id, tag)
+        return prompt_config
+    except JudgmentAPIError as e:
+        raise JudgmentAPIError(
+            status_code=e.status_code,
+            detail=f"Failed to fetch prompt '{name}': {e.detail}",
+            response=e.response,
+        )
+
+
+@dataclass
+class Prompt:
+    name: str
+    prompt: str
+    tags: List[str]
+    commit_id: str
+    parent_commit_id: Optional[str] = None
+    _template: Template = field(init=False, repr=False)
+
+    def __post_init__(self):
+        template_str = re.sub(r"\{\{(\w+)\}\}", r"$\1", self.prompt)
+        self._template = Template(template_str)
+
+    @classmethod
+    def create(cls, name: str, prompt: str, tags: Optional[List[str]] = None):
+        if not tags:
+            tags = []
+        commit_id, parent_commit_id = push_prompt(name, prompt, tags)
+        return cls(
+            name=name,
+            prompt=prompt,
+            tags=tags,
+            commit_id=commit_id,
+            parent_commit_id=parent_commit_id,
+        )
+
+    @classmethod
+    def get(cls, name: str, commit_id: Optional[str] = None, tag: Optional[str] = None):
+        if commit_id is not None and tag is not None:
+            raise ValueError(
+                "You cannot fetch a prompt by both commit_id and tag at the same time"
+            )
+        prompt_config = fetch_prompt(name, commit_id, tag)
+        return cls(
+            name=prompt_config["name"],
+            prompt=prompt_config["prompt"],
+            tags=prompt_config["tags"],
+            commit_id=prompt_config["commit_id"],
+            parent_commit_id=prompt_config["parent_commit_id"],
+        )
+
+    def compile(self, **kwargs) -> str:
+        try:
+            return self._template.substitute(**kwargs)
+        except KeyError as e:
+            missing_var = str(e).strip("'")
+            raise ValueError(f"Missing required variable: {missing_var}")

src/judgeval/scorers/judgeval_scorers/api_scorers/prompt_scorer.py

@@ -32,12 +32,6 @@ def push_prompt_scorer(
             }
         )
     except JudgmentAPIError as e:
-        if e.status_code == 500:
-            raise JudgmentAPIError(
-                status_code=e.status_code,
-                detail=f"The server is temporarily unavailable. Please try your request again in a few moments. Error details: {e.detail}",
-                response=e.response,
-            )
         raise JudgmentAPIError(
             status_code=e.status_code,
             detail=f"Failed to save prompt scorer: {e.detail}",
@@ -58,12 +52,6 @@ def fetch_prompt_scorer(
         scorer_config.pop("updated_at")
         return scorer_config
     except JudgmentAPIError as e:
-        if e.status_code == 500:
-            raise JudgmentAPIError(
-                status_code=e.status_code,
-                detail=f"The server is temporarily unavailable. Please try your request again in a few moments. Error details: {e.detail}",
-                response=e.response,
-            )
         raise JudgmentAPIError(
             status_code=e.status_code,
             detail=f"Failed to fetch prompt scorer '{name}': {e.detail}",