ollama Interface: comparison OllamaInterface/OllamaClient.py

-:3118d16e526e
+:c471738b75b3
 """
 Public method to ask the 'ollama' server to delete the given model.
 @param model name of the model
 @type str
-"""
+@return flag indicating success
-# TODO: not implemented yet
+@rtype bool
+"""
 ollamaRequest = {
-"name": model,
+"model": model,
 }
-self.__sendRequest("delete", data=ollamaRequest)
+_, status = self.__sendSyncRequest("delete", data=ollamaRequest, delete=True)
+return status == 200  # HTTP status 200 OK
 def list(self):
 """
 Public method to request a list of models available locally from the 'ollama'
 server.
 models = []
 with contextlib.suppress(KeyError):
 for model in response["models"]:
 name = model["name"]
 if name:
-models.append(name.replace(":latest", ""))
+models.append(name)
 self.modelsList.emit(models)
 def listDetails(self):
 """
 Public method to request a list of models available locally from the 'ollama'
 server with some model details.
-"""
-response = self.__sendSyncRequest("tags")
+@return list of dictionaries containing the available models and related data
+@rtype list[dict[str, Any]]
+"""
+response, _ = self.__sendSyncRequest("tags")
 models = []
 if response is not None:
 with contextlib.suppress(KeyError):
 for model in response["models"]:
 return models
 def listRunning(self):
 """
 Public method to request a list of running models from the 'ollama' server.
-"""
-response = self.__sendSyncRequest("ps")
+@return list of dictionaries containing the running models and related data
+@rtype list[dict[str, Any]]
+"""
+response, _ = self.__sendSyncRequest("ps")
 models = []
 if response is not None:
 with contextlib.suppress(KeyError):
 for model in response["models"]:
 @return current client state
 @rtype OllamaClientState
 """
 return self.__state
-def __getServerReply(self, endpoint, data=None):
+def __getServerReply(self, endpoint, data=None, delete=False):
 """
 Private method to send a request to the 'ollama' server and return a reply
 object.
 @param endpoint 'ollama' API endpoint to be contacted
 @type str
 @param data dictionary containing the data to send to the server
 (defaults to None)
 @type dict (optional)
+@param delete flag indicating to send a delete request (defaults to False)
+@type bool (optional)
 @return 'ollama' server reply
 @rtype QNetworkReply
 """
-ollamaUrl =  QUrl(
+ollamaUrl = QUrl(
 "{0}://{1}:{2}/api/{3}".format(
 self.__plugin.getPreferences("OllamaScheme"),
 (
 "127.0.0.1"
 if self.__localServer
 if data is not None:
 request.setHeader(
 QNetworkRequest.KnownHeaders.ContentTypeHeader, "application/json"
 )
 jsonData = json.dumps(data).encode("utf-8")
-reply = self.__networkManager.post(request, jsonData)
+if delete:
+reply = self.__networkManager.sendCustomRequest(
+request, b"DELETE", jsonData
+)
+else:
+reply = self.__networkManager.post(request, jsonData)
 else:
 reply = self.__networkManager.get(request)
 reply.errorOccurred.connect(lambda error: self.__errorOccurred(error, reply))
 return reply
 @param processResponse function handling the received data (defaults to None)
 @type function (optional)
 """
 self.__state = OllamaClientState.Requesting
-##ollamaUrl = QUrl(
-##"{0}://{1}:{2}/api/{3}".format(
-##self.__plugin.getPreferences("OllamaScheme"),
-##(
-##"127.0.0.1"
-##if self.__localServer
-##else self.__plugin.getPreferences("OllamaHost")
-##),
-##(
-##self.__plugin.getPreferences("OllamaLocalPort")
-##if self.__localServer
-##else self.__plugin.getPreferences("OllamaPort")
-##),
-##endpoint,
-##)
-##)
-##request = QNetworkRequest(ollamaUrl)
-##if data is not None:
-##request.setHeader(
-##QNetworkRequest.KnownHeaders.ContentTypeHeader, "application/json"
-##)
-##jsonData = json.dumps(data).encode("utf-8")
-##reply = self.__networkManager.post(request, jsonData)
-##else:
-##reply = self.__networkManager.get(request)
-##
 reply = self.__getServerReply(endpoint=endpoint, data=data)
 reply.finished.connect(lambda: self.__replyFinished(reply))
-##reply.errorOccurred.connect(lambda error: self.__errorOccurred(error, reply))
 reply.readyRead.connect(lambda: self.__processData(reply, processResponse))
 self.__replies.append(reply)
 def __replyFinished(self, reply):
 """
 with contextlib.suppress(json.JSONDecodeError):
 data = json.loads(buffer)
 if data and processResponse:
 processResponse(data)
-def __sendSyncRequest(self, endpoint, data=None):
+def __sendSyncRequest(self, endpoint, data=None, delete=False):
 """
 Private method to send a request to the 'ollama' server and handle its
 responses.
 @param endpoint 'ollama' API endpoint to be contacted
 @type str
 @param data dictionary containing the data to send to the server
 (defaults to None)
 @type dict (optional)
+@param delete flag indicating to send a delete request (defaults to False)
+@type bool (optional)
+@return tuple containing the data sent by the 'ollama' server and the HTTP
+status code
+@rtype tuple of (Any, int)
 """
 self.__state = OllamaClientState.Requesting
-reply = self.__getServerReply(endpoint=endpoint, data=data)
+reply = self.__getServerReply(endpoint=endpoint, data=data, delete=delete)
 while not reply.isFinished():
 QCoreApplication.processEvents()
 QThread.msleep(100)
 reply.deleteLater()
 self.__state = OllamaClientState.Finished
+statusCode = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute)
 if reply.error() == QNetworkReply.NetworkError.NoError:
 buffer = bytes(reply.readAll())
 with contextlib.suppress(json.JSONDecodeError):
 data = json.loads(buffer)
-return data
+return data, statusCode
-return None
+return None, statusCode
 def heartbeat(self):
 """
 Public method to check, if the 'ollama' server has started and is responsive.

Mercurial Repositories > ollama Interface / file comparison

comparison: OllamaInterface/OllamaClient.py

OllamaInterface/OllamaClient.py