Enhance PromptCache and SysPrompt to support versioning for cache operations

Author: juanjcardona13

sysprompt/cache.py

@@ -17,40 +17,43 @@ def __init__(self, default_ttl: int = 3600):
         self._expiration: dict = {}
         self._default_ttl: int = default_ttl
 
-    def get(self, key: str) -> Optional[Any]:
+    def get(self, key: str, version: Optional[str] = None) -> Optional[Any]:
         """
         Get a prompt from the cache.
         """
+        cache_key = self._calculate_key(key, version)
         if self._default_ttl == 0:
             return None  # Cache is disabled
-        if key in self._cache:
-            if self._is_expired(key):
-                self.delete(key)
+        if cache_key in self._cache:
+            if self._is_expired(cache_key):
+                self.delete(cache_key)
                 return None
-            self._cache.move_to_end(key)
-            return self._cache[key]
+            self._cache.move_to_end(cache_key)
+            return self._cache[cache_key]
         return None
 
-    def set(self, key: str, value: Any, ttl: Optional[int] = None):
+    def set(self, key: str, version: Optional[str] = None, value: Any = None, ttl: Optional[int] = None):
         """
         Set a prompt in the cache.
         """
         ttl = ttl if ttl is not None else self._default_ttl
+        cache_key = self._calculate_key(key, version)
         if ttl == 0:
             return  # Don't cache if TTL is 0
-        if key in self._cache:
-            del self._cache[key]
-        self._cache[key] = value
-        self._expiration[key] = time.time() + (ttl or self._default_ttl)
-        self._cache.move_to_end(key)
+        if cache_key in self._cache:
+            del self._cache[cache_key]
+        self._cache[cache_key] = value
+        self._expiration[cache_key] = time.time() + (ttl or self._default_ttl)
+        self._cache.move_to_end(cache_key)
 
-    def delete(self, key: str):
+    def delete(self, key: str, version: Optional[str] = None):
         """
         Delete a prompt from the cache.
         """
-        if key in self._cache:
-            del self._cache[key]
-            del self._expiration[key]
+        cache_key = self._calculate_key(key, version)
+        if cache_key in self._cache:
+            del self._cache[cache_key]
+            del self._expiration[cache_key]
 
     def clear(self):
         """
@@ -71,6 +74,9 @@ def set_default_ttl(self, ttl: int):
         """
         self._default_ttl = ttl
 
+    def _calculate_key(self, key: str, version: Optional[str] = None) -> str:
+        return f"{key}:{version}" if version else key
+
     def __len__(self):
         """
         Gets the number of prompts in the cache.

sysprompt/main.py

@@ -22,8 +22,8 @@ class SysPrompt:
     """
 
     BASE_URL: str = "https://api.sysprompt.com"
-    PROMPT_RETRIEVAL_PATH: str = "sdk/prompts/{code_id}"
-    PROMPT_LOGGING_PATH: str = "sdk/logs/prompt"
+    PROMPT_RETRIEVAL_PATH: str = "sdk/prompts/{code_id}/"
+    PROMPT_LOGGING_PATH: str = "sdk/logs/prompt/"
     API_VERSION: str = "v1"
 
     def __init__(self, api_key: Optional[str], raise_on_error: bool = False, default_cache_ttl: int = 900):
@@ -71,36 +71,37 @@ def preload_prompts(self, code_ids: Optional[List[str]] = None):
         """
         prompts: List[dict] = []
         if code_ids is None:
-            prompts = list(self._api_request('GET', 'sdk/prompts'))
+            prompts = list(self._api_request('GET', 'sdk/prompts/'))
         else:
-            prompts = list(self._api_request('POST', 'sdk/prompts', {'code_ids': code_ids}))
+            prompts = list(self._api_request('POST', 'sdk/prompts/', {'code_ids': code_ids}))
 
         for prompt in prompts:
-            self._prompts_cache.set(prompt['code_id'], prompt)
+            self._prompts_cache.set(prompt['code_id'], None, prompt)
 
-    def get_prompt(self, code_id: str) -> str | dict | list:
+    def get_prompt(self, code_id: str, version: Optional[str] = None) -> str | dict | list:
         """
         Retrieve a prompt from the SysPrompt API.
 
         Args:
             code_id (str): The unique identifier for the prompt.
-
+            version (Optional[str]): The version of the prompt to retrieve, empty for the default version.
         Returns:
             str | dict | list: The prompt data.
 
         Raises:
             SysPromptException: If the API request fails.
         """
-        cached_prompt = self._prompts_cache.get(code_id)
+        cached_prompt = self._prompts_cache.get(code_id, version)
         if cached_prompt is not None:
             return cached_prompt.get('content', [])
 
         try:
             if not code_id:
                 raise SysPromptException("Prompt code ID is required. Got empty string.")
-            prompt = self._api_request('GET', self.PROMPT_RETRIEVAL_PATH.format(code_id=code_id))
+            version_string = f"?version={version}" if version else ""
+            prompt = self._api_request('GET', f"{self.PROMPT_RETRIEVAL_PATH.format(code_id=code_id)}{version_string}")
             if prompt:
-                self._prompts_cache.set(code_id, prompt)
+                self._prompts_cache.set(code_id, version, prompt)
             return prompt.get('content', [])
         except Exception as e:
             if self._raise_on_error:
@@ -113,21 +114,23 @@ def compile(
         self,
         prompt_id: str | None = None,
         params: dict = {},
-        prompt_object: None | str | dict | list = None
+        version: Optional[str] = None,
+        prompt_object: None | str | dict | list = None,
     ) -> str | dict | list:
         """
         Compile a prompt with the given parameters.
         You can either provide a prompt code ID or a prompt object.
 
         Args:
+            prompt_id (str): The unique identifier for the prompt.
             params (dict): The parameters to substitute into the prompt.
-            prompt_code_id (str): The unique identifier for the prompt.
+            version (str): The version of the prompt to retrieve, empty for the default version.
             prompt_object (str | dict | list): The prompt object to compile.
         Returns:
             str | dict | list: The compiled prompt.
         """
 
-        prompt: str | dict | list | None = self.get_prompt(prompt_id) if prompt_id else prompt_object
+        prompt: str | dict | list | None = self.get_prompt(prompt_id, version) if prompt_id else prompt_object
         if not prompt:
             if self._raise_on_error:
                 raise SysPromptException("Prompt not found")
@@ -316,6 +319,7 @@ def _api_request(self, method: str, path: str, data: Optional[dict] = None) -> d
         for attempt in range(max_retries):
             try:
                 url: str = self._get_url(path)
+                print(url)
                 response: requests.Response = requests.request(method, url, json=data, headers=self._get_headers())
                 response.raise_for_status()
                 return response.json()
@@ -375,7 +379,7 @@ def _get_url(self, path: str) -> str:
         Returns:
             str: The full URL.
         """
-        return f"{self.BASE_URL}/{self.API_VERSION}/{path}/"
+        return f"{self.BASE_URL}/{self.API_VERSION}/{path}"
 
     def _get_headers(self) -> dict:
         """