Browse Source

improve type hints using typing.Literal and add type annotations (#25641)

Co-authored-by: EchterTimo <EchterTimo@users.noreply.github.com>
tags/1.9.0
Timo 1 month ago
parent
commit
9234a2293d
No account linked to committer's email address
1 changed files with 18 additions and 18 deletions
  1. 18
    18
      sdks/python-client/dify_client/client.py

+ 18
- 18
sdks/python-client/dify_client/client.py View File

import json import json
from typing import Literal
import requests import requests




self.api_key = api_key self.api_key = api_key
self.base_url = base_url self.base_url = base_url


def _send_request(self, method, endpoint, json=None, params=None, stream=False):
def _send_request(self, method: str, endpoint: str, json: dict | None = None, params: dict | None = None, stream: bool = False):
headers = { headers = {
"Authorization": f"Bearer {self.api_key}", "Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json", "Content-Type": "application/json",


return response return response


def message_feedback(self, message_id, rating, user):
def message_feedback(self, message_id: str, rating: Literal["like", "dislike"], user: str):
data = {"rating": rating, "user": user} data = {"rating": rating, "user": user}
return self._send_request("POST", f"/messages/{message_id}/feedbacks", data) return self._send_request("POST", f"/messages/{message_id}/feedbacks", data)


def get_application_parameters(self, user):
def get_application_parameters(self, user: str):
params = {"user": user} params = {"user": user}
return self._send_request("GET", "/parameters", params=params) return self._send_request("GET", "/parameters", params=params)


def file_upload(self, user, files):
def file_upload(self, user: str, files: dict):
data = {"user": user} data = {"user": user}
return self._send_request_with_files( return self._send_request_with_files(
"POST", "/files/upload", data=data, files=files "POST", "/files/upload", data=data, files=files
data = {"text": text, "user": user, "streaming": streaming} data = {"text": text, "user": user, "streaming": streaming}
return self._send_request("POST", "/text-to-audio", json=data) return self._send_request("POST", "/text-to-audio", json=data)


def get_meta(self, user):
def get_meta(self, user: str):
params = {"user": user} params = {"user": user}
return self._send_request("GET", "/meta", params=params) return self._send_request("GET", "/meta", params=params)




class CompletionClient(DifyClient): class CompletionClient(DifyClient):
def create_completion_message(self, inputs, response_mode, user, files=None):
def create_completion_message(self, inputs: dict, response_mode: Literal["blocking", "streaming"], user: str, files: dict | None = None):
data = { data = {
"inputs": inputs, "inputs": inputs,
"response_mode": response_mode, "response_mode": response_mode,
inputs: dict, inputs: dict,
query: str, query: str,
user: str, user: str,
response_mode: str = "blocking",
response_mode: Literal["blocking", "streaming"] = "blocking",
conversation_id: str | None = None, conversation_id: str | None = None,
files: dict | None = None, files: dict | None = None,
): ):


class WorkflowClient(DifyClient): class WorkflowClient(DifyClient):
def run( def run(
self, inputs: dict, response_mode: str = "streaming", user: str = "abc-123"
self, inputs: dict, response_mode: Literal["blocking", "streaming"] = "streaming", user: str = "abc-123"
): ):
data = {"inputs": inputs, "response_mode": response_mode, "user": user} data = {"inputs": inputs, "response_mode": response_mode, "user": user}
return self._send_request("POST", "/workflows/run", data) return self._send_request("POST", "/workflows/run", data)
class KnowledgeBaseClient(DifyClient): class KnowledgeBaseClient(DifyClient):
def __init__( def __init__(
self, self,
api_key,
api_key: str,
base_url: str = "https://api.dify.ai/v1", base_url: str = "https://api.dify.ai/v1",
dataset_id: str | None = None, dataset_id: str | None = None,
): ):
return self._send_request("POST", url, json=data, **kwargs) return self._send_request("POST", url, json=data, **kwargs)


def update_document_by_text( def update_document_by_text(
self, document_id, name, text, extra_params: dict | None = None, **kwargs
self, document_id: str, name: str, text: str, extra_params: dict | None = None, **kwargs
): ):
""" """
Update a document by text. Update a document by text.
return self._send_request("POST", url, json=data, **kwargs) return self._send_request("POST", url, json=data, **kwargs)


def create_document_by_file( def create_document_by_file(
self, file_path, original_document_id=None, extra_params: dict | None = None
self, file_path: str, original_document_id: str | None = None, extra_params: dict | None = None
): ):
""" """
Create a document by file. Create a document by file.
) )


def update_document_by_file( def update_document_by_file(
self, document_id, file_path, extra_params: dict | None = None
self, document_id: str, file_path: str, extra_params: dict | None = None
): ):
""" """
Update a document by file. Update a document by file.
url = f"/datasets/{self._get_dataset_id()}" url = f"/datasets/{self._get_dataset_id()}"
return self._send_request("DELETE", url) return self._send_request("DELETE", url)


def delete_document(self, document_id):
def delete_document(self, document_id: str):
""" """
Delete a document. Delete a document.


url = f"/datasets/{self._get_dataset_id()}/documents" url = f"/datasets/{self._get_dataset_id()}/documents"
return self._send_request("GET", url, params=params, **kwargs) return self._send_request("GET", url, params=params, **kwargs)


def add_segments(self, document_id, segments, **kwargs):
def add_segments(self, document_id: str, segments: list[dict], **kwargs):
""" """
Add segments to a document. Add segments to a document.




def query_segments( def query_segments(
self, self,
document_id,
document_id: str,
keyword: str | None = None, keyword: str | None = None,
status: str | None = None, status: str | None = None,
**kwargs, **kwargs,
params.update(kwargs["params"]) params.update(kwargs["params"])
return self._send_request("GET", url, params=params, **kwargs) return self._send_request("GET", url, params=params, **kwargs)


def delete_document_segment(self, document_id, segment_id):
def delete_document_segment(self, document_id: str, segment_id: str):
""" """
Delete a segment from a document. Delete a segment from a document.


url = f"/datasets/{self._get_dataset_id()}/documents/{document_id}/segments/{segment_id}" url = f"/datasets/{self._get_dataset_id()}/documents/{document_id}/segments/{segment_id}"
return self._send_request("DELETE", url) return self._send_request("DELETE", url)


def update_document_segment(self, document_id, segment_id, segment_data, **kwargs):
def update_document_segment(self, document_id: str, segment_id: str, segment_data: dict, **kwargs):
""" """
Update a segment in a document. Update a segment in a document.



Loading…
Cancel
Save