Add S3 File Provider

docker/compose.full.yaml

@@ -5,14 +5,16 @@ volumes:
     name: hatchet_config
   hatchet_api_key:
     name: hatchet_api_key
-  postgres_data:
-    name: postgres_data
   hatchet_rabbitmq_data:
     name: hatchet_rabbitmq_data
   hatchet_rabbitmq_conf:
     name: hatchet_rabbitmq_conf
   hatchet_postgres_data:
     name: hatchet_postgres_data
+  minio_data:
+    name: minio_data
+  postgres_data:
+    name: postgres_data
 
 services:
   postgres:
@@ -34,6 +36,24 @@ services:
       postgres
       -c max_connections=1024
 
+  minio:
+    image: minio/minio
+    profiles: [minio]
+    env_file:
+      - ./env/minio.env
+    volumes:
+      - minio_data:/data
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    restart: on-failure
+    command: server /data --console-address ":9001"
+
   hatchet-postgres:
     image: postgres:latest
     env_file:

docker/compose.yaml

@@ -1,6 +1,8 @@
 volumes:
   postgres_data:
     name: postgres_data
+  minio_data:
+    name: minio_data
 
 services:
   postgres:
@@ -22,6 +24,24 @@ services:
       postgres
       -c max_connections=1024
 
+  minio:
+    image: minio/minio
+    profiles: [minio]
+    env_file:
+      - ./env/minio.env
+    volumes:
+      - minio_data:/data
+    ports:
+      - "9000:9000"
+      - "9001:9001"
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    restart: on-failure
+    command: server /data --console-address ":9001"
+
   graph_clustering:
     image: ragtoriches/cluster-prod
     ports:

docker/env/minio.env

@@ -0,0 +1,2 @@
+MINIO_ROOT_USER=minioadmin
+MINIO_ROOT_PASSWORD=minioadmin

py/all_possible_config.toml

@@ -211,6 +211,12 @@ concurrent_request_limit = 256
 ################################################################################
 [file]
 provider = "postgres"
+# If using S3
+bucket_name = ""
+endpoint_url = ""
+region_name = ""
+aws_access_key_id = ""
+aws_secret_access_key = ""
 
 ################################################################################
 # Ingestion Settings (IngestionConfig and nested settings)

py/core/base/__init__.py

@@ -89,18 +89,21 @@
     # Crypto provider
     "CryptoConfig",
     "CryptoProvider",
-    # Email provider
-    "EmailConfig",
-    "EmailProvider",
     # Database providers
     "LimitSettings",
     "DatabaseConfig",
     "DatabaseProvider",
     "Handler",
     "PostgresConfigurationSettings",
+    # Email provider
+    "EmailConfig",
+    "EmailProvider",
     # Embedding provider
     "EmbeddingConfig",
     "EmbeddingProvider",
+    # File provider
+    "FileConfig",
+    "FileProvider",
     # Ingestion provider
     "IngestionConfig",
     "IngestionProvider",

py/core/base/providers/__init__.py

@@ -11,6 +11,7 @@
 )
 from .email import EmailConfig, EmailProvider
 from .embedding import EmbeddingConfig, EmbeddingProvider
+from .file import FileConfig, FileProvider
 from .ingestion import (
     ChunkingStrategy,
     IngestionConfig,
@@ -29,26 +30,29 @@
     "AppConfig",
     "Provider",
     "ProviderConfig",
-    # Ingestion provider
-    "IngestionConfig",
-    "IngestionProvider",
-    "ChunkingStrategy",
     # Crypto provider
     "CryptoConfig",
     "CryptoProvider",
-    # Email provider
-    "EmailConfig",
-    "EmailProvider",
     # Database providers
     "DatabaseConnectionManager",
     "DatabaseConfig",
     "LimitSettings",
     "PostgresConfigurationSettings",
     "DatabaseProvider",
     "Handler",
+    # Email provider
+    "EmailConfig",
+    "EmailProvider",
     # Embedding provider
     "EmbeddingConfig",
     "EmbeddingProvider",
+    # File provider
+    "FileConfig",
+    "FileProvider",
+    # Ingestion provider
+    "IngestionConfig",
+    "IngestionProvider",
+    "ChunkingStrategy",
     # LLM provider
     "CompletionConfig",
     "CompletionProvider",

py/core/base/providers/file.py

@@ -0,0 +1,110 @@
+import logging
+import os
+from abc import ABC, abstractmethod
+from datetime import datetime
+from io import BytesIO
+from typing import BinaryIO, Optional
+from uuid import UUID
+
+from .base import Provider, ProviderConfig
+
+logger = logging.getLogger()
+
+
+class FileConfig(ProviderConfig):
+    """
+    Configuration for file storage providers.
+    """
+
+    provider: Optional[str] = None
+
+    # S3-specific configuration
+    bucket_name: Optional[str] = None
+    aws_access_key_id: Optional[str] = None
+    aws_secret_access_key: Optional[str] = None
+    region_name: Optional[str] = None
+    endpoint_url: Optional[str] = None
+
+    @property
+    def supported_providers(self) -> list[str]:
+        """
+        List of supported file storage providers.
+        """
+        return [
+            "postgres",
+            "s3",
+        ]
+
+    def validate_config(self) -> None:
+        if self.provider not in self.supported_providers:
+            raise ValueError(f"Unsupported file provider: {self.provider}")
+
+        if self.provider == "s3" and (
+            not self.bucket_name and not os.getenv("S3_BUCKET_NAME")
+        ):
+            raise ValueError(
+                "S3 bucket name is required when using S3 provider"
+            )
+
+
+class FileProvider(Provider, ABC):
+    """
+    Base abstract class for file storage providers.
+    """
+
+    def __init__(self, config: FileConfig):
+        if not isinstance(config, FileConfig):
+            raise ValueError(
+                "FileProvider must be initialized with a `FileConfig`."
+            )
+        super().__init__(config)
+        self.config: FileConfig = config
+
+    @abstractmethod
+    async def initialize(self) -> None:
+        """Initialize the file provider."""
+        pass
+
+    @abstractmethod
+    async def store_file(
+        self,
+        document_id: UUID,
+        file_name: str,
+        file_content: BytesIO,
+        file_type: Optional[str] = None,
+    ) -> None:
+        """Store a file."""
+        pass
+
+    @abstractmethod
+    async def retrieve_file(
+        self, document_id: UUID
+    ) -> Optional[tuple[str, BinaryIO, int]]:
+        """Retrieve a file."""
+        pass
+
+    @abstractmethod
+    async def retrieve_files_as_zip(
+        self,
+        document_ids: Optional[list[UUID]] = None,
+        start_date: Optional[datetime] = None,
+        end_date: Optional[datetime] = None,
+    ) -> tuple[str, BinaryIO, int]:
+        """Retrieve multiple files as a zip."""
+        pass
+
+    @abstractmethod
+    async def delete_file(self, document_id: UUID) -> bool:
+        """Delete a file."""
+        pass
+
+    @abstractmethod
+    async def get_files_overview(
+        self,
+        offset: int,
+        limit: int,
+        filter_document_ids: Optional[list[UUID]] = None,
+        filter_file_names: Optional[list[str]] = None,
+    ) -> list[dict]:
+        """Get an overview of stored files."""
+        pass

py/core/main/abstractions.py

@@ -19,9 +19,11 @@
     OpenAICompletionProvider,
     OpenAIEmbeddingProvider,
     PostgresDatabaseProvider,
+    PostgresFileProvider,
     R2RAuthProvider,
     R2RCompletionProvider,
     R2RIngestionProvider,
+    S3FileProvider,
     SendGridEmailProvider,
     SimpleOrchestrationProvider,
     SupabaseAuthProvider,
@@ -59,6 +61,7 @@ class R2RProviders(BaseModel):
         | OpenAIEmbeddingProvider
         | OllamaEmbeddingProvider
     )
+    file: PostgresFileProvider | S3FileProvider
     completion_embedding: (
         LiteLLMEmbeddingProvider
         | OpenAIEmbeddingProvider

py/core/main/api/v3/documents_router.py

@@ -517,7 +517,7 @@ async def create_document(
             }
 
             file_name = file_data["filename"]
-            await self.providers.database.files_handler.store_file(
+            await self.providers.file.store_file(
                 document_id,
                 file_name,
                 file_content,

py/core/main/assembly/factory.py

@@ -12,6 +12,7 @@
     EmailConfig,
     EmbeddingConfig,
     EmbeddingProvider,
+    FileConfig,
     IngestionConfig,
     OCRConfig,
     OrchestrationConfig,
@@ -213,20 +214,40 @@ async def create_database_provider(
         quantization_type = (
             self.config.embedding.quantization_settings.quantization_type
         )
-        if db_config.provider == "postgres":
-            database_provider = PostgresDatabaseProvider(
-                db_config,
-                dimension,
-                crypto_provider=crypto_provider,
-                quantization_type=quantization_type,
-            )
-            await database_provider.initialize()
-            return database_provider
-        else:
+        if db_config.provider != "postgres":
             raise ValueError(
                 f"Database provider {db_config.provider} not supported"
             )
 
+        database_provider = PostgresDatabaseProvider(
+            db_config,
+            dimension,
+            crypto_provider=crypto_provider,
+            quantization_type=quantization_type,
+        )
+        await database_provider.initialize()
+        return database_provider
+
+    @staticmethod
+    def create_file_provider(
+        config: FileConfig, database_provider=None, *args, **kwargs
+    ):
+        if config.provider == "postgres":
+            from core.providers import PostgresFileProvider
+
+            return PostgresFileProvider(
+                config=config,
+                project_name=database_provider.project_name,
+                connection_manager=database_provider.connection_manager,
+            )
+
+        elif config.provider == "s3":
+            from core.providers import S3FileProvider
+
+            return S3FileProvider(config)
+        else:
+            raise ValueError(f"File provider {config.provider} not supported")
+
     @staticmethod
     def create_embedding_provider(
         embedding: EmbeddingConfig, *args, **kwargs
@@ -407,6 +428,11 @@ async def create_providers(
             )
         )
 
+        file_provider = self.create_file_provider(
+            config=self.config.file, database_provider=database_provider
+        )
+        await file_provider.initialize()
+
         ocr_provider = ocr_provider_override or self.create_ocr_provider(
             self.config.ocr
         )
@@ -454,12 +480,13 @@ async def create_providers(
 
         return R2RProviders(
             auth=auth_provider,
+            completion_embedding=completion_embedding_provider,
             database=database_provider,
+            email=email_provider,
             embedding=embedding_provider,
-            completion_embedding=completion_embedding_provider,
+            file=file_provider,
             ingestion=ingestion_provider,
             llm=llm_provider,
-            email=email_provider,
             ocr=ocr_provider,
             orchestration=orchestration_provider,
             scheduler=scheduler_provider,