provider.py 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163
  1. import os
  2. import boto3
  3. from botocore.exceptions import ClientError
  4. import shutil
  5. from typing import BinaryIO, Tuple, Optional, Union
  6. from open_webui.constants import ERROR_MESSAGES
  7. from open_webui.config import (
  8. STORAGE_PROVIDER,
  9. S3_ACCESS_KEY_ID,
  10. S3_SECRET_ACCESS_KEY,
  11. S3_BUCKET_NAME,
  12. S3_REGION_NAME,
  13. S3_ENDPOINT_URL,
  14. UPLOAD_DIR,
  15. )
  16. import boto3
  17. from botocore.exceptions import ClientError
  18. from typing import BinaryIO, Tuple, Optional
  19. class StorageProvider:
  20. def __init__(self, provider: Optional[str] = None):
  21. self.storage_provider: str = provider or STORAGE_PROVIDER
  22. self.s3_client = None
  23. self.s3_bucket_name: Optional[str] = None
  24. if self.storage_provider == "s3":
  25. self._initialize_s3()
  26. def _initialize_s3(self) -> None:
  27. """Initializes the S3 client and bucket name if using S3 storage."""
  28. self.s3_client = boto3.client(
  29. "s3",
  30. region_name=S3_REGION_NAME,
  31. endpoint_url=S3_ENDPOINT_URL,
  32. aws_access_key_id=S3_ACCESS_KEY_ID,
  33. aws_secret_access_key=S3_SECRET_ACCESS_KEY,
  34. )
  35. self.bucket_name = S3_BUCKET_NAME
  36. def _upload_to_s3(self, file: BinaryIO, filename: str) -> Tuple[bytes, str]:
  37. """Handles uploading of the file to S3 storage."""
  38. if not self.s3_client:
  39. raise RuntimeError("S3 Client is not initialized.")
  40. try:
  41. self.s3_client.upload_fileobj(file, self.bucket_name, filename)
  42. return file.read(), f"s3://{self.bucket_name}/{filename}"
  43. except ClientError as e:
  44. raise RuntimeError(f"Error uploading file to S3: {e}")
  45. def _upload_to_local(self, contents: bytes, filename: str) -> Tuple[bytes, str]:
  46. """Handles uploading of the file to local storage."""
  47. file_path = f"{UPLOAD_DIR}/{filename}"
  48. with open(file_path, "wb") as f:
  49. f.write(contents)
  50. return contents, file_path
  51. def _get_file_from_s3(self, file_path: str) -> str:
  52. """Handles downloading of the file from S3 storage."""
  53. if not self.s3_client:
  54. raise RuntimeError("S3 Client is not initialized.")
  55. try:
  56. bucket_name, key = file_path.split("//")[1].split("/")
  57. local_file_path = f"{UPLOAD_DIR}/{key}"
  58. self.s3_client.download_file(bucket_name, key, local_file_path)
  59. return local_file_path
  60. except ClientError as e:
  61. raise RuntimeError(f"Error downloading file from S3: {e}")
  62. def _get_file_from_local(self, file_path: str) -> str:
  63. """Handles downloading of the file from local storage."""
  64. return file_path
  65. def _delete_from_s3(self, filename: str) -> None:
  66. """Handles deletion of the file from S3 storage."""
  67. if not self.s3_client:
  68. raise RuntimeError("S3 Client is not initialized.")
  69. try:
  70. self.s3_client.delete_object(Bucket=self.bucket_name, Key=filename)
  71. except ClientError as e:
  72. raise RuntimeError(f"Error deleting file from S3: {e}")
  73. def _delete_from_local(self, filename: str) -> None:
  74. """Handles deletion of the file from local storage."""
  75. file_path = f"{UPLOAD_DIR}/{filename}"
  76. if os.path.isfile(file_path):
  77. os.remove(file_path)
  78. else:
  79. print(f"File {file_path} not found in local storage.")
  80. def _delete_all_from_s3(self) -> None:
  81. """Handles deletion of all files from S3 storage."""
  82. if not self.s3_client:
  83. raise RuntimeError("S3 Client is not initialized.")
  84. try:
  85. response = self.s3_client.list_objects_v2(Bucket=self.bucket_name)
  86. if "Contents" in response:
  87. for content in response["Contents"]:
  88. self.s3_client.delete_object(
  89. Bucket=self.bucket_name, Key=content["Key"]
  90. )
  91. except ClientError as e:
  92. raise RuntimeError(f"Error deleting all files from S3: {e}")
  93. def _delete_all_from_local(self) -> None:
  94. """Handles deletion of all files from local storage."""
  95. if os.path.exists(UPLOAD_DIR):
  96. for filename in os.listdir(UPLOAD_DIR):
  97. file_path = os.path.join(UPLOAD_DIR, filename)
  98. try:
  99. if os.path.isfile(file_path) or os.path.islink(file_path):
  100. os.unlink(file_path) # Remove the file or link
  101. elif os.path.isdir(file_path):
  102. shutil.rmtree(file_path) # Remove the directory
  103. except Exception as e:
  104. print(f"Failed to delete {file_path}. Reason: {e}")
  105. else:
  106. print(f"Directory {UPLOAD_DIR} not found in local storage.")
  107. def upload_file(self, file: BinaryIO, filename: str) -> Tuple[bytes, str]:
  108. """Uploads a file either to S3 or the local file system."""
  109. contents = file.read()
  110. if not contents:
  111. raise ValueError(ERROR_MESSAGES.EMPTY_CONTENT)
  112. if self.storage_provider == "s3":
  113. return self._upload_to_s3(file, filename)
  114. return self._upload_to_local(contents, filename)
  115. def get_file(self, file_path: str) -> str:
  116. """Downloads a file either from S3 or the local file system and returns the file path."""
  117. if self.storage_provider == "s3":
  118. return self._get_file_from_s3(file_path)
  119. return self._get_file_from_local(file_path)
  120. def delete_file(self, filename: str) -> None:
  121. """Deletes a file either from S3 or the local file system."""
  122. if self.storage_provider == "s3":
  123. self._delete_from_s3(filename)
  124. # Always delete from local storage
  125. self._delete_from_local(filename)
  126. def delete_all_files(self) -> None:
  127. """Deletes all files from the storage."""
  128. if self.storage_provider == "s3":
  129. self._delete_all_from_s3()
  130. # Always delete from local storage
  131. self._delete_all_from_local()
  132. Storage = StorageProvider(provider=STORAGE_PROVIDER)