mirror of
https://github.com/langgenius/dify-plugin-daemon.git
synced 2026-07-22 01:35:24 -04:00
Tianyi Jing
1dacb201ff
* feat: add template for datasource plugins Signed-off-by: jingfelix <jingfelix@outlook.com> * fix: description and missing category Signed-off-by: jingfelix <jingfelix@outlook.com> --------- Signed-off-by: jingfelix <jingfelix@outlook.com>
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
from typing import Any, Mapping
|
|
import secrets
|
|
import urllib.parse
|
|
|
|
from dify_plugin.errors.tool import ToolProviderCredentialValidationError, DatasourceOAuthError
|
|
from dify_plugin.interfaces.datasource import DatasourceProvider, DatasourceOAuthCredentials
|
|
|
|
|
|
|
|
class {{ .PluginName | SnakeToCamel }}Provider(DatasourceProvider):
|
|
|
|
def _validate_credentials(self, credentials: Mapping[str, Any]) -> None:
|
|
try:
|
|
"""
|
|
IMPLEMENT YOUR VALIDATION HERE
|
|
"""
|
|
except Exception as e:
|
|
raise ToolProviderCredentialValidationError(str(e))
|
|
|
|
|
|
#########################################################################################
|
|
# If OAuth is supported, uncomment the following functions.
|
|
# Warning: please make sure that the sdk version is 0.5.0 or higher.
|
|
#########################################################################################
|
|
# def _oauth_get_authorization_url(self, redirect_uri: str, system_credentials: Mapping[str, Any]) -> str:
|
|
# """
|
|
# Generate the authorization URL for {{ .PluginName }} OAuth.
|
|
# """
|
|
# try:
|
|
# """
|
|
# IMPLEMENT YOUR AUTHORIZATION URL GENERATION HERE
|
|
# """
|
|
# except Exception as e:
|
|
# raise DatasourceOAuthError(str(e))
|
|
# return ""
|
|
|
|
# def _oauth_get_credentials(
|
|
# self, redirect_uri: str, system_credentials: Mapping[str, Any], request: Request
|
|
# ) -> DatasourceOAuthCredentials:
|
|
# """
|
|
# Exchange code for access_token.
|
|
# """
|
|
# try:
|
|
# """
|
|
# IMPLEMENT YOUR CREDENTIALS EXCHANGE HERE
|
|
# """
|
|
# except Exception as e:
|
|
# raise DatasourceOAuthError(str(e))
|
|
# return DatasourceOAuthCredentials(
|
|
# name="",
|
|
# avatar_url="",
|
|
# expires_at=-1,
|
|
# credentials={},
|
|
# )
|
|
|
|
# def _oauth_refresh_credentials(
|
|
# self, redirect_uri: str, system_credentials: Mapping[str, Any], credentials: Mapping[str, Any]
|
|
# ) -> DatasourceOAuthCredentials:
|
|
# """
|
|
# Refresh the credentials
|
|
# """
|
|
# return DatasourceOAuthCredentials(
|
|
# name="",
|
|
# avatar_url="",
|
|
# expires_at=-1,
|
|
# credentials={},
|
|
# )
|
|
|