File size: 1,598 Bytes
ed3fa91 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 |
import asyncio
import cloudinary
import cloudinary.uploader
import cloudinary.api
from phi.tools import Toolkit
from phi.utils.log import logger
class CloudinaryTool(Toolkit):
def __init__(
self,
cloud_name: str,
api_key: str,
api_secret: str,
):
super().__init__(name="cloudinary_tool")
# Initialize Cloudinary
cloudinary.config(
cloud_name=cloud_name,
api_key=api_key,
api_secret=api_secret
)
# Registering methods to make them accessible via the toolkit
self.register(self.upload_file)
def upload_file(self, file_path: str) -> str:
"""
Uploads a file to Cloudinary.
Args:
file_path (str): The path to the file to upload.
Returns:
dict: A dictionary containing the upload response details.
Example:
>>> upload_file('/path/to/file.pdf')
"""
logger.info(f"Uploading file to Cloudinary: {file_path}")
try:
response = cloudinary.uploader.upload(file_path, resource_type="raw")
logger.info(f"Upload response: {response}")
return f"{response}"
except Exception as e:
logger.warning(f"Failed to upload file: {e}")
return f"Error: {e}"
# # Example usage
# cloudinary_tool = CloudinaryTool(
# cloud_name="your_cloud_name",
# api_key="your_api_key",
# api_secret="your_api_secret"
# )
#
# upload_response = cloudinary_tool.upload_file('/path/to/your/file.txt')
# print(upload_response)
|