File Storage API

Upload, download, search, and manage files in Silo file storage buckets. Requires FILE_TOKEN. Supports async bulk operations.

Bases: BaseAPIClient

API client for Silo file storage operations.

This class provides methods for managing files in the Authentic8 Silo platform, including uploading, downloading, searching, modifying, and deleting files.

The file storage API allows you to: - Upload files from local system to Silo storage - Download files from Silo storage to local system - Search for files using various criteria - Modify file properties and metadata - Delete files from storage - Manage file permissions and access controls

Example

from silo_sdk import FileAPI, load_config config = load_config() files = FileAPI(config)

Upload a file

result = files.upload_file( ... bucket_id="my_bucket", ... file_path="/path/to/local/file.pdf", ... name="document.pdf", ... path="/documents/" ... )

Search for files

found_files = files.find_files( ... bucket_id="my_bucket", ... name="*.pdf", ... path="/documents/" ... )

Source code in silo_sdk/storage/file_api.py

class FileAPI(BaseAPIClient):
    """API client for Silo file storage operations.

    This class provides methods for managing files in the Authentic8 Silo platform,
    including uploading, downloading, searching, modifying, and deleting files.

    The file storage API allows you to:
    - Upload files from local system to Silo storage
    - Download files from Silo storage to local system
    - Search for files using various criteria
    - Modify file properties and metadata
    - Delete files from storage
    - Manage file permissions and access controls

    Example:
        >>> from silo_sdk import FileAPI, load_config
        >>> config = load_config()
        >>> files = FileAPI(config)
        >>>
        >>> # Upload a file
        >>> result = files.upload_file(
        ...     bucket_id="my_bucket",
        ...     file_path="/path/to/local/file.pdf",
        ...     name="document.pdf",
        ...     path="/documents/"
        ... )
        >>>
        >>> # Search for files
        >>> found_files = files.find_files(
        ...     bucket_id="my_bucket",
        ...     name="*.pdf",
        ...     path="/documents/"
        ... )
    """

    def __init__(self, config: dict[str, Any]) -> None:
        """Initialize the file storage API client.

        Args:
            config: Configuration dictionary containing API settings

        Raises:
            ConfigurationError: If required configuration is missing
        """
        super().__init__(config)

        # Validate required configuration
        validate_config_for_api(config, ["FILE_TOKEN"])

        self.auth_token = config["FILE_TOKEN"]
        self._config = config  # Store config for later use
        self.logger.info("Initialized FileAPI client")

    def _cleanup_partial_file(self, output_path: Path) -> None:
        """Remove a partially-written file left behind by a failed download.

        Called when a download raises mid-stream (e.g. a dropped connection
        or disk error while writing chunks). The output file was already
        opened in write mode by that point, so any prior contents at
        ``output_path`` are gone regardless — leaving the truncated remnant
        in place would let a caller mistake it for a complete file.

        Args:
            output_path: Path of the file that was being written when the
                download failed.
        """
        try:
            if output_path.exists():
                output_path.unlink()
                self.logger.debug(
                    "Removed partial file after failed download: %s", output_path
                )
        except OSError as cleanup_error:
            self.logger.warning(
                "Failed to remove partial file %s after download error: %s",
                output_path,
                cleanup_error,
            )

    @api_tag(MethodType.API_COMMAND, api_command="putfile")
    def upload_file(
        self,
        bucket_id: str,
        file_path: str | Path,
        name: str,
        path: str = "/",
        content_type: str | None = None,
    ) -> dict[str, Any]:
        """Upload file from local system to Silo storage.

        Uploads a file from the local filesystem to the specified bucket in
        Silo storage with the given name and path.

        Args:
            bucket_id: ID of the destination bucket (required)
            file_path: Path to the local file to upload (required)
            name: Name to give the file in Silo storage (required)
            path: Path in Silo storage to store the file (optional, defaults to "/")
            content_type: MIME content type of the file (optional, auto-detected if not provided)

        Returns:
            Dictionary containing upload result including:
            - file_id: Unique identifier for the uploaded file
            - name: Name of the file in storage
            - path: Storage path of the file
            - size: File size in bytes
            - content_type: MIME type of the file
            - upload_ts: Upload timestamp

        Raises:
            FileAPIError: If file upload fails
            ValidationError: If parameters are invalid

        Example:
            >>> result = api.upload_file(
            ...     bucket_id="documents",
            ...     file_path="/home/user/report.pdf",
            ...     name="quarterly_report.pdf",
            ...     path="/reports/2024/"
            ... )
            >>> print(f"Uploaded file ID: {result['file_id']}")
        """
        # Validate parameters
        if not bucket_id or not isinstance(bucket_id, str):
            raise ValidationError("Bucket ID must be a non-empty string")

        if not name or not isinstance(name, str):
            raise ValidationError("File name must be a non-empty string")

        if not isinstance(path, str):
            raise ValidationError("Path must be a string")

        # Convert to Path object and validate file exists
        file_path = Path(file_path)
        if not file_path.exists():
            raise ValidationError(f"File does not exist: {file_path}")

        if not file_path.is_file():
            raise ValidationError(f"Path is not a file: {file_path}")

        # Auto-detect content type if not provided
        if content_type is None:
            import mimetypes

            content_type, _ = mimetypes.guess_type(str(file_path))
            if content_type is None:
                content_type = "application/octet-stream"

        # Prepare upload data for /putfile/ endpoint
        url = f"{self.base_url.rstrip('/')}/putfile/"

        try:
            self.logger.debug(
                "Uploading file: %s to bucket: %s", file_path.name, bucket_id
            )
            self.logger.debug("Upload URL: %s", url)

            with open(file_path, "rb") as f:
                # Prepare form data for /putfile/ endpoint
                data = {
                    "auth": self.auth_token,
                    "bucket_id": bucket_id,
                    "name": name,
                    "path": path,
                }

                # Add optional parameters
                if content_type:
                    data["content_type"] = content_type

                files = {"file": (file_path.name, f, content_type)}

                # Debug logging
                self.logger.debug("Data keys: %s", list(data.keys()))
                self.logger.debug("Files keys: %s", list(files.keys()))

                # Route through the shared session so retry/proxy/User-Agent
                # configuration from BaseAPIClient is honored. Override
                # Content-Type (set to None to drop the session's default
                # "application/json") so requests can compute the correct
                # multipart/form-data boundary itself.
                response = self.session.post(
                    url,
                    data=data,
                    files=files,
                    timeout=self.timeout,
                    headers={"Content-Type": None},
                )

                # Log response details for debugging
                self.logger.debug("Response status: %s", response.status_code)
                if not response.ok:
                    self.logger.error("Response content: %s", response.text)

                response.raise_for_status()

                result: dict[str, Any] = response.json()
                self.logger.info("Successfully uploaded file: %s", name)
                return result

        except Exception as e:
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to upload file: {e}") from e

    @api_tag(MethodType.API_COMMAND, api_command="getfile")
    def download_file(
        self,
        file_id: str,
        output_path: str | Path,
        create_dirs: bool = True,
    ) -> str:
        """Download file by file_id from Silo storage.

        Downloads a file from Silo storage to the local filesystem using the
        file's unique identifier.

        Args:
            file_id: ID of the file to download
            output_path: Path where to save the downloaded file
            create_dirs: Whether to create parent directories if they don't exist

        Note:
            Wire endpoint: ``POST https://extapi.authentic8.com/getfile/`` with
            form-data fields ``id`` (file ID) and ``auth`` (file token). This is
            **not** a JSON command-array request — it uses a separate endpoint
            and multipart form-data, not the standard ``/api/`` command format.
            The response body is the raw file content (binary stream).

        Returns:
            Path of the downloaded file (as string)

        Raises:
            FileAPIError: If file download fails
            ValidationError: If parameters are invalid

        Example:
            >>> downloaded_path = api.download_file(
            ...     file_id="abc123",
            ...     output_path="/home/user/downloads/report.pdf"
            ... )
            >>> print(f"File downloaded to: {downloaded_path}")
        """
        if not file_id or not isinstance(file_id, str):
            raise ValidationError("File ID must be a non-empty string")

        if not output_path:
            raise ValidationError("Output path must be provided")

        # Convert to Path object and resolve to prevent path traversal
        output_path_raw = str(output_path)
        output_path = Path(output_path).resolve()

        # Warn if the original path contained traversal components
        if ".." in output_path_raw:
            self.logger.warning(
                "Output path contained '..' components, resolved to: %s",
                output_path,
            )

        # Create parent directories if requested
        if create_dirs and output_path.parent != output_path:
            output_path.parent.mkdir(parents=True, exist_ok=True)

        url = f"{self.base_url.rstrip('/')}/getfile/"

        try:
            self.logger.debug("Downloading file ID: %s", file_id)

            data = {
                "id": file_id,
                "auth": self.auth_token,
            }

            # Route through the shared session so retry/proxy/User-Agent
            # configuration from BaseAPIClient is honored. Override
            # Content-Type (set to None to drop the session's default
            # "application/json") so requests encodes the form body correctly.
            response = self.session.post(
                url,
                data=data,
                timeout=self.timeout,
                stream=True,
                headers={"Content-Type": None},
            )
            response.raise_for_status()

            # Write file content
            try:
                with open(output_path, "wb") as f:
                    for chunk in response.iter_content(chunk_size=8192):
                        if chunk:
                            f.write(chunk)
            except Exception:
                self._cleanup_partial_file(output_path)
                raise

            self.logger.info("Successfully downloaded file to: %s", output_path)
            return str(output_path)

        except Exception as e:
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to download file: {e}") from e

    @api_tag(MethodType.API_COMMAND, api_command="findfiles")
    def find_files(
        self,
        bucket_id: str,
        name: str | None = None,
        path: str | None = None,
        file_type: str | None = None,
        created_before: str | None = None,
        created_after: str | None = None,
        size_min: int | None = None,
        size_max: int | None = None,
        limit: int | None = None,
        metadata_filters: dict[str, str] | None = None,
    ) -> list[dict[str, Any]]:
        """Search Silo storage for files by various criteria.

        Searches for files in the specified bucket using various filter criteria
        such as name patterns, paths, file types, creation dates, and file sizes.

        Args:
            bucket_id: ID of the bucket to search in.
            name: Name pattern to search for (supports wildcards).
            path: Path to search in (supports wildcards).
            file_type: MIME type or file extension to filter by. Sent as
                ``"type"`` in the wire format (field name differs from this
                parameter name).
            created_before: Search for files created before this date (ISO
                format). Sent as ``":created_before"`` in the wire format
                (note the colon prefix on the wire field name).
            created_after: Search for files created after this date (ISO
                format). Sent as ``":created_after"`` in the wire format
                (note the colon prefix on the wire field name).
            size_min: Minimum file size in bytes.
            size_max: Maximum file size in bytes.
            limit: Maximum number of results to return.
            metadata_filters: Dictionary of metadata key-value pairs to filter by.

        Note:
            Wire format parameter name differences:

            - Python ``file_type`` → wire ``"type"``
            - Python ``created_before`` → wire ``":created_before"`` (colon prefix)
            - Python ``created_after`` → wire ``":created_after"`` (colon prefix)

            Wire format structure::

                {
                    "command": "findfiles",
                    "bucket_id": "my_bucket",
                    "type": "application/pdf",          # ← file_type → "type"
                    ":created_before": "2026-01-01",    # ← colon prefix
                    ":created_after": "2025-01-01"      # ← colon prefix
                }

        Returns:
            List of dictionaries containing file information including:
            - file_id: Unique file identifier
            - name: File name
            - path: File path in storage
            - size: File size in bytes
            - content_type: MIME type
            - created_ts: Creation timestamp
            - modified_ts: Last modification timestamp
            - metadata: Custom metadata (if any)

        Raises:
            FileAPIError: If file search fails
            ValidationError: If parameters are invalid

        Example:
            >>> files = api.find_files(
            ...     bucket_id="documents",
            ...     name="*.pdf",
            ...     path="/reports/*",
            ...     created_after="2024-01-01T00:00:00Z",
            ...     size_min=1024,
            ...     limit=100
            ... )
            >>> for file in files:
            ...     print(f"{file['name']} - {file['size']} bytes")
        """
        if not bucket_id or not isinstance(bucket_id, str):
            raise ValidationError("Bucket ID must be a non-empty string")

        # Build search payload
        search_params: dict[str, Any] = {
            "command": "findfiles",
            "bucket_id": bucket_id,
        }

        # Add optional search criteria
        if name is not None:
            search_params["name"] = name

        if path is not None:
            search_params["path"] = path

        if file_type is not None:
            search_params["type"] = file_type

        if created_before is not None:
            search_params[":created_before"] = created_before

        if created_after is not None:
            search_params[":created_after"] = created_after

        if size_min is not None:
            if not isinstance(size_min, int) or size_min < 0:
                raise ValidationError("size_min must be a non-negative integer")
            search_params["size_min"] = size_min

        if size_max is not None:
            if not isinstance(size_max, int) or size_max < 0:
                raise ValidationError("size_max must be a non-negative integer")
            search_params["size_max"] = size_max

        if limit is not None:
            if not isinstance(limit, int) or limit <= 0:
                raise ValidationError("limit must be a positive integer")
            search_params["limit"] = limit

        if metadata_filters is not None:
            if not isinstance(metadata_filters, dict):
                raise ValidationError("metadata_filters must be a dictionary")
            # Add metadata filters to search parameters
            for key, value in metadata_filters.items():
                search_params[f"metadata:{key}"] = value

        payload = [search_params]

        try:
            self.logger.debug("Searching for files in bucket: %s", bucket_id)
            response = self._make_api_request("POST", "api/", self.auth_token, payload)

            # Extract results
            if len(response) >= 2 and isinstance(response[1], dict):
                result_entry = response[1]
                if "error" in result_entry:
                    raise FileAPIError(f"API error: {result_entry['error']}")
                result = result_entry.get("result", [])
                if isinstance(result, list):
                    self.logger.info("Found %d files matching criteria", len(result))
                    return result

            self.logger.warning("Unexpected response format for find_files")
            return []

        except Exception as e:
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to search for files: {e}") from e

    @api_tag(MethodType.CONVENIENCE, wraps=["findfiles"])
    def list_files(
        self,
        bucket_id: str,
        limit: int | None = None,
    ) -> list[dict[str, Any]]:
        """List files in a bucket.

        Lists files in the specified bucket with optional limit.
        This is a convenience wrapper around find_files with no filters.

        Args:
            bucket_id: ID of the bucket to list files from
            limit: Maximum number of files to return

        Returns:
            List of dictionaries containing file information

        Raises:
            FileAPIError: If file listing fails
            ValidationError: If parameters are invalid
        """
        return self.find_files(bucket_id=bucket_id, limit=limit)

    @api_tag(MethodType.API_COMMAND, api_command="modifyfile")
    def modify_file(
        self,
        file_id: str,
        name: str | None = None,
        expire_ts: str | None = None,
        path: str | None = None,
        content_type: str | None = None,
    ) -> dict[str, Any]:
        """Change properties of a file by file_id in Silo storage.

        Modifies various properties of an existing file including its name,
        path, expiration time, and content type.

        Args:
            file_id: ID of the file to modify (required)
            name: New name for the file (optional)
            expire_ts: New file expire date (optional)
            path: Path of new location to move file (optional)
            content_type: New content type of modified file (optional)

        Returns:
            Dictionary containing the API response

        Raises:
            FileAPIError: If file modification fails
            ValidationError: If parameters are invalid

        Example:
            >>> updated_file = api.modify_file(
            ...     file_id="abc123",
            ...     name="updated_report.pdf",
            ...     expire_ts="now+3600",
            ...     path="/reports/2024/updated/"
            ... )
            >>> print(f"Updated file: {updated_file}")
        """
        if not file_id or not isinstance(file_id, str):
            raise ValidationError("File ID must be a non-empty string")

        # Build modifyfile command — setauth is injected by _make_api_request
        modify_command: dict[str, Any] = {"command": "modifyfile", "file_id": file_id}

        # Add optional parameters
        if name is not None:
            if not isinstance(name, str) or not name:
                raise ValidationError("Name must be a non-empty string")
            modify_command["name"] = name

        if expire_ts is not None:
            modify_command["expire_ts"] = expire_ts

        if path is not None:
            if not isinstance(path, str):
                raise ValidationError("Path must be a string")
            modify_command["path"] = path

        if content_type is not None:
            if not isinstance(content_type, str):
                raise ValidationError("Content type must be a string")
            modify_command["content_type"] = content_type

        payload = [modify_command]

        try:
            self.logger.debug("Modifying file: %s", file_id)

            # Route through the shared request helper (like every other
            # command) instead of posting directly to the bare base_url,
            # which is missing the "api/" path segment the server expects.
            response = self._make_api_request("POST", "api/", self.auth_token, payload)

            # Check if modification was successful
            if len(response) >= 2 and isinstance(response[1], dict):
                result_entry = response[1]
                if "error" in result_entry:
                    raise FileAPIError(f"API error: {result_entry['error']}")
                modify_result = result_entry.get("result")
                if modify_result and "update file" in str(modify_result):
                    self.logger.info("Successfully modified file: %s", file_id)
                    return {
                        "result": modify_result,
                        "file_id": file_id,
                        "response": response,
                    }

            raise FileAPIError(f"Unexpected response format: {response}")

        except Exception as e:
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to modify file: {e}") from e

    @api_tag(MethodType.API_COMMAND, api_command="deletefile")
    def delete_file(self, file_id: str) -> bool:
        """Delete file in Silo storage by file_id.

        Permanently deletes a file from Silo storage. This action cannot be undone.

        Args:
            file_id: ID of the file to delete

        Returns:
            True if deletion was successful, False otherwise

        Raises:
            FileAPIError: If file deletion fails
            ValidationError: If file_id is invalid

        Example:
            >>> success = api.delete_file("abc123")
            >>> if success:
            ...     print("File deleted successfully")
        """
        if not file_id or not isinstance(file_id, str):
            raise ValidationError("File ID must be a non-empty string")

        payload = [
            {
                "command": "deletefile",
                "file_id": file_id,
            }
        ]

        try:
            self.logger.debug("Deleting file: %s", file_id)
            response = self._make_api_request("POST", "api/", self.auth_token, payload)

            result = self._extract_api_result(response)
            if isinstance(result, str) and "deleted file" in result:
                self.logger.info("Successfully deleted file: %s", file_id)
                return True
            if isinstance(result, dict) and (
                result.get("deleted") or result.get("success")
            ):
                self.logger.info("Successfully deleted file: %s", file_id)
                return True

            self.logger.warning("File deletion may have failed: %s", result)
            return False

        except Exception as e:
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to delete file: {e}") from e

    @api_tag(MethodType.CONVENIENCE, wraps=["findfiles"])
    def get_file_info(
        self, file_id: str, bucket_id: str | None = None
    ) -> dict[str, Any]:
        """Get detailed information about a file.

        Retrieves information about a single file using a file_id-scoped
        findfiles lookup, rather than fetching the entire bucket listing
        and scanning it client-side.

        Args:
            file_id: ID of the file to get information for
            bucket_id: Optional bucket ID to search in (if not provided, searches configured bucket)

        Returns:
            Dictionary containing detailed file information

        Raises:
            FileAPIError: If file information retrieval fails
            ValidationError: If file_id is invalid

        Example:
            >>> file_info = api.get_file_info("abc123")
            >>> print(f"File: {file_info['name']}")
            >>> print(f"Size: {file_info['file_size']} bytes")
            >>> print(f"Created: {file_info['create_ts']}")
        """
        if not file_id or not isinstance(file_id, str):
            raise ValidationError("File ID must be a non-empty string")

        try:
            self.logger.debug("Getting file info for: %s", file_id)

            # Use the provided bucket_id or fall back to configured bucket
            search_bucket_id = bucket_id
            if not search_bucket_id:
                # Try to get from config if available
                search_bucket_id = self._config.get("BUCKET_ID")

            # Scope the findfiles lookup to this specific file_id instead of
            # requesting the entire bucket listing. This avoids the failure
            # mode where a server-side cap on findfiles results causes a
            # file that genuinely exists to be reported as "not found"
            # simply because it fell outside the returned page.
            find_command: dict[str, Any] = {
                "command": "findfiles",
                "file_id": file_id,
            }

            # Add bucket_id if available
            if search_bucket_id:
                find_command["bucket_id"] = search_bucket_id

            payload = [find_command]

            response = self._make_api_request("POST", "api/", self.auth_token, payload)

            # Extract results and find the matching file_id
            if len(response) >= 2 and isinstance(response[1], dict):
                result_entry = response[1]
                if "error" in result_entry:
                    raise FileAPIError(f"API error: {result_entry['error']}")
                files_list = result_entry.get("result", [])
                if isinstance(files_list, list):
                    # Find the file with matching file_id. The server-side
                    # filter should already narrow this to a single entry,
                    # but we still match explicitly in case the wire
                    # implementation returns extra entries.
                    for file_info in files_list:
                        if (
                            isinstance(file_info, dict)
                            and file_info.get("file_id") == file_id
                        ):
                            self.logger.debug("Found file information for: %s", file_id)
                            return file_info

                    # File not found in this bucket
                    bucket_msg = (
                        f" in bucket {search_bucket_id}" if search_bucket_id else ""
                    )
                    raise FileAPIError(f"File with ID {file_id} not found{bucket_msg}")

            raise FileAPIError(f"Unexpected response format: {response}")

        except Exception as e:
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to get file information: {e}") from e

    # Async Download Methods (require httpx)

    @api_tag(MethodType.ASYNC_VARIANT, api_command="getfile", sync_of="download_file")
    async def download_file_with_retry_async(
        self,
        file_id: str,
        output_path: str | Path,
        max_retries: int = 3,
        timeout: int = 60,
        create_dirs: bool = True,
    ) -> str:
        """Asynchronously download file by file_id with retry mechanism.
        Downloads a file from Silo storage with configurable retry logic and
        timeout handling. This async version allows for non-blocking downloads
        and can be used for concurrent file downloads.

        Args:
            file_id: ID of the file to download
            output_path: Path where to save the downloaded file
            max_retries: Positive maximum number of retry attempts (default: 3)
            timeout: Positive request timeout in seconds (default: 60)
            create_dirs: Whether to create parent directories if they don't exist

        Returns:
            Path of the downloaded file (as string)

        Raises:
            FileAPIError: If file download fails
            ValidationError: If parameters are invalid

        Example:
            >>> import asyncio
            >>> async def download_files():
            ...     files = FileAPI(config)
            ...     downloaded_path = await files.download_file_with_retry_async(
            ...         file_id="abc123",
            ...         output_path="/home/user/downloads/report.pdf",
            ...         max_retries=5
            ...     )
            ...     return downloaded_path
            >>> path = asyncio.run(download_files())
        """
        validate_positive_int(max_retries, "max_retries")
        validate_positive_int(timeout, "timeout")

        if not file_id or not isinstance(file_id, str):
            raise ValidationError("File ID must be a non-empty string")

        if not output_path:
            raise ValidationError("Output path must be provided")

        # Convert to Path object
        output_path = Path(output_path)

        # Create parent directories if requested
        if create_dirs and output_path.parent != output_path:
            output_path.parent.mkdir(parents=True, exist_ok=True)

        try:
            self.logger.debug(
                "Starting async download with retry for file ID: %s", file_id
            )

            # Implement async download with retry logic directly
            output_path = Path(output_path)
            if create_dirs and output_path.parent != output_path:
                output_path.parent.mkdir(parents=True, exist_ok=True)

            ssl_context = self.create_ssl_context()
            download_url = f"{self.base_url.rstrip('/')}/getfile/"

            async with httpx.AsyncClient(verify=ssl_context) as client:
                for attempt in range(max_retries):
                    self.logger.debug(
                        f"Download attempt {attempt + 1} for file {file_id}"
                    )
                    try:
                        response = await client.post(
                            download_url,
                            data={"id": file_id, "auth": self.auth_token},
                            timeout=httpx.Timeout(timeout),
                        )
                        response.raise_for_status()
                        content = response.content

                        with open(output_path, "wb") as f:
                            f.write(content)

                        self.logger.info(
                            "Successfully downloaded file to: %s", output_path
                        )
                        return str(output_path)

                    except httpx.HTTPError as e:
                        self.logger.error(
                            f"Error downloading file (attempt {attempt + 1}/{max_retries}): {e}"
                        )
                        if attempt == max_retries - 1:
                            raise FileAPIError(
                                f"Failed to download file after {max_retries} attempts: {e}"
                            ) from e
                        await asyncio.sleep(2**attempt)  # Exponential backoff

            # If we get here, all retries were exhausted without success
            raise FileAPIError(
                f"Failed to download file after {max_retries} attempts"
            )  # pragma: no cover

        except Exception as e:  # pragma: no cover
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to download file with retry: {e}") from e

    @api_tag(MethodType.ASYNC_VARIANT, api_command="getfile", sync_of="download_file")
    async def download_file_chunked_async(
        self,
        file_id: str,
        output_path: str | Path,
        chunk_size: int = 8192,
        timeout: int = 60,
        create_dirs: bool = True,
    ) -> str:
        """Asynchronously download file by file_id using chunked download.
        Downloads a file from Silo storage using streaming/chunked approach,
        which is more memory-efficient for large files. This async version
        allows for non-blocking downloads of large files.

        Args:
            file_id: ID of the file to download
            output_path: Path where to save the downloaded file
            chunk_size: Positive size of chunks to read at a time (default: 8192)
            timeout: Positive request timeout in seconds (default: 60)
            create_dirs: Whether to create parent directories if they don't exist

        Returns:
            Path of the downloaded file (as string)

        Raises:
            FileAPIError: If file download fails
            ValidationError: If parameters are invalid

        Example:
            >>> import asyncio
            >>> async def download_large_file():
            ...     files = FileAPI(config)
            ...     downloaded_path = await files.download_file_chunked_async(
            ...         file_id="large_file_123",
            ...         output_path="/home/user/downloads/large_archive.zip",
            ...         chunk_size=16384
            ...     )
            ...     return downloaded_path
            >>> path = asyncio.run(download_large_file())
        """
        validate_positive_int(chunk_size, "chunk_size")
        validate_positive_int(timeout, "timeout")

        if not file_id or not isinstance(file_id, str):
            raise ValidationError("File ID must be a non-empty string")

        if not output_path:
            raise ValidationError("Output path must be provided")

        # Convert to Path object
        output_path = Path(output_path)

        # Create parent directories if requested
        if create_dirs and output_path.parent != output_path:
            output_path.parent.mkdir(parents=True, exist_ok=True)

        try:
            self.logger.debug(
                "Starting async chunked download for file ID: %s", file_id
            )

            # Implement async chunked download logic directly
            ssl_context = self.create_ssl_context()
            download_url = f"{self.base_url.rstrip('/')}/getfile/"

            async with httpx.AsyncClient(verify=ssl_context) as client:
                try:
                    async with client.stream(
                        "POST",
                        download_url,
                        data={"id": file_id, "auth": self.auth_token},
                        timeout=httpx.Timeout(timeout),
                    ) as response:
                        response.raise_for_status()

                        try:
                            with open(output_path, "wb") as f:
                                async for chunk in response.aiter_bytes(chunk_size):
                                    f.write(chunk)
                        except Exception:
                            self._cleanup_partial_file(output_path)
                            raise

                        self.logger.info(
                            "Successfully downloaded file to: %s", output_path
                        )
                        return str(output_path)

                except httpx.HTTPError as e:
                    raise FileAPIError(f"Failed to download file: {e}") from e

        except Exception as e:
            if isinstance(e, FileAPIError):
                raise
            raise FileAPIError(f"Failed to download file chunked: {e}") from e

    @api_tag(MethodType.ASYNC_VARIANT, api_command="getfile", sync_of="download_file")
    async def bulk_download_async(
        self,
        file_downloads: list[dict[str, Any]],
        max_concurrent: int = 5,
        use_chunked: bool = False,
    ) -> list[str | None]:
        """Download multiple files concurrently.

        Downloads multiple files in parallel with configurable concurrency
        limits to efficiently handle bulk file downloads.

        Args:
            file_downloads: List of download dictionaries, each containing:
                - file_id: ID of the file to download
                - output_path: Path where to save the file
                - max_retries: Optional max retries (default: 3)
                - timeout: Optional timeout (default: 60)
                - chunk_size: Optional chunk size for chunked downloads
            max_concurrent: Positive maximum number of concurrent downloads (default: 5)
            use_chunked: Whether to use chunked download method (default: False)

        Returns:
            List of downloaded file paths (same order as input), None for failed downloads

        Raises:
            ValidationError: If parameters are invalid

        Example:
            >>> import asyncio
            >>> async def bulk_download():
            ...     files = FileAPI(config)
            ...     downloads = [
            ...         {"file_id": "file1", "output_path": "./downloads/file1.pdf"},
            ...         {"file_id": "file2", "output_path": "./downloads/file2.zip"},
            ...     ]
            ...     results = await files.bulk_download_async(
            ...         file_downloads=downloads,
            ...         max_concurrent=3
            ...     )
            ...     return results
            >>> results = asyncio.run(bulk_download())
        """
        import asyncio

        validate_positive_int(max_concurrent, "max_concurrent")

        # Create semaphore to limit concurrent downloads
        semaphore = asyncio.Semaphore(max_concurrent)

        async def download_single_file(
            download_config: dict[str, Any],
        ) -> str | None:
            """Download a single file with semaphore control."""
            async with semaphore:
                try:
                    file_id = download_config["file_id"]
                    output_path = download_config["output_path"]
                    max_retries = download_config.get("max_retries", 3)
                    timeout = download_config.get("timeout", 60)

                    if use_chunked:
                        chunk_size = download_config.get("chunk_size", 8192)
                        return await self.download_file_chunked_async(
                            file_id=file_id,
                            output_path=output_path,
                            chunk_size=chunk_size,
                            timeout=timeout,
                        )
                    else:
                        return await self.download_file_with_retry_async(
                            file_id=file_id,
                            output_path=output_path,
                            max_retries=max_retries,
                            timeout=timeout,
                        )
                except Exception as e:
                    self.logger.error(
                        f"Failed to download {download_config.get('file_id')}: {e}"
                    )
                    return None

        # Process all downloads concurrently
        self.logger.info(f"Starting bulk download of {len(file_downloads)} files")
        results = await asyncio.gather(
            *[download_single_file(download) for download in file_downloads],
            return_exceptions=True,
        )

        # Convert exceptions to None and log errors
        processed_results: list[str | None] = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                self.logger.error(f"Download {i + 1} failed with exception: {result}")
                processed_results.append(None)
            elif isinstance(result, str):
                processed_results.append(result)
            else:
                processed_results.append(None)

        successful_downloads = sum(1 for r in processed_results if r is not None)
        self.logger.info(
            f"Completed bulk download: {successful_downloads}/{len(file_downloads)} "
            "files successful"
        )

        return processed_results

__init__

__init__(config: dict[str, Any]) -> None

Initialize the file storage API client.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Configuration dictionary containing API settings	required

Raises:

Type	Description
ConfigurationError	If required configuration is missing

Source code in silo_sdk/storage/file_api.py

def __init__(self, config: dict[str, Any]) -> None:
    """Initialize the file storage API client.

    Args:
        config: Configuration dictionary containing API settings

    Raises:
        ConfigurationError: If required configuration is missing
    """
    super().__init__(config)

    # Validate required configuration
    validate_config_for_api(config, ["FILE_TOKEN"])

    self.auth_token = config["FILE_TOKEN"]
    self._config = config  # Store config for later use
    self.logger.info("Initialized FileAPI client")

upload_file

upload_file(bucket_id: str, file_path: str | Path, name: str, path: str = '/', content_type: str | None = None) -> dict[str, Any]

Upload file from local system to Silo storage.

Uploads a file from the local filesystem to the specified bucket in Silo storage with the given name and path.

Parameters:

Name	Type	Description	Default
bucket_id	str	ID of the destination bucket (required)	required
file_path	str | Path	Path to the local file to upload (required)	required
name	str	Name to give the file in Silo storage (required)	required
path	str	Path in Silo storage to store the file (optional, defaults to "/")	'/'
content_type	str | None	MIME content type of the file (optional, auto-detected if not provided)	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing upload result including:
dict[str, Any]	file_id: Unique identifier for the uploaded file
dict[str, Any]	name: Name of the file in storage
dict[str, Any]	path: Storage path of the file
dict[str, Any]	size: File size in bytes
dict[str, Any]	content_type: MIME type of the file
dict[str, Any]	upload_ts: Upload timestamp

Raises:

Type	Description
FileAPIError	If file upload fails
ValidationError	If parameters are invalid

Example

result = api.upload_file( ... bucket_id="documents", ... file_path="/home/user/report.pdf", ... name="quarterly_report.pdf", ... path="/reports/2024/" ... ) print(f"Uploaded file ID: {result['file_id']}")

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.API_COMMAND, api_command="putfile")
def upload_file(
    self,
    bucket_id: str,
    file_path: str | Path,
    name: str,
    path: str = "/",
    content_type: str | None = None,
) -> dict[str, Any]:
    """Upload file from local system to Silo storage.

    Uploads a file from the local filesystem to the specified bucket in
    Silo storage with the given name and path.

    Args:
        bucket_id: ID of the destination bucket (required)
        file_path: Path to the local file to upload (required)
        name: Name to give the file in Silo storage (required)
        path: Path in Silo storage to store the file (optional, defaults to "/")
        content_type: MIME content type of the file (optional, auto-detected if not provided)

    Returns:
        Dictionary containing upload result including:
        - file_id: Unique identifier for the uploaded file
        - name: Name of the file in storage
        - path: Storage path of the file
        - size: File size in bytes
        - content_type: MIME type of the file
        - upload_ts: Upload timestamp

    Raises:
        FileAPIError: If file upload fails
        ValidationError: If parameters are invalid

    Example:
        >>> result = api.upload_file(
        ...     bucket_id="documents",
        ...     file_path="/home/user/report.pdf",
        ...     name="quarterly_report.pdf",
        ...     path="/reports/2024/"
        ... )
        >>> print(f"Uploaded file ID: {result['file_id']}")
    """
    # Validate parameters
    if not bucket_id or not isinstance(bucket_id, str):
        raise ValidationError("Bucket ID must be a non-empty string")

    if not name or not isinstance(name, str):
        raise ValidationError("File name must be a non-empty string")

    if not isinstance(path, str):
        raise ValidationError("Path must be a string")

    # Convert to Path object and validate file exists
    file_path = Path(file_path)
    if not file_path.exists():
        raise ValidationError(f"File does not exist: {file_path}")

    if not file_path.is_file():
        raise ValidationError(f"Path is not a file: {file_path}")

    # Auto-detect content type if not provided
    if content_type is None:
        import mimetypes

        content_type, _ = mimetypes.guess_type(str(file_path))
        if content_type is None:
            content_type = "application/octet-stream"

    # Prepare upload data for /putfile/ endpoint
    url = f"{self.base_url.rstrip('/')}/putfile/"

    try:
        self.logger.debug(
            "Uploading file: %s to bucket: %s", file_path.name, bucket_id
        )
        self.logger.debug("Upload URL: %s", url)

        with open(file_path, "rb") as f:
            # Prepare form data for /putfile/ endpoint
            data = {
                "auth": self.auth_token,
                "bucket_id": bucket_id,
                "name": name,
                "path": path,
            }

            # Add optional parameters
            if content_type:
                data["content_type"] = content_type

            files = {"file": (file_path.name, f, content_type)}

            # Debug logging
            self.logger.debug("Data keys: %s", list(data.keys()))
            self.logger.debug("Files keys: %s", list(files.keys()))

            # Route through the shared session so retry/proxy/User-Agent
            # configuration from BaseAPIClient is honored. Override
            # Content-Type (set to None to drop the session's default
            # "application/json") so requests can compute the correct
            # multipart/form-data boundary itself.
            response = self.session.post(
                url,
                data=data,
                files=files,
                timeout=self.timeout,
                headers={"Content-Type": None},
            )

            # Log response details for debugging
            self.logger.debug("Response status: %s", response.status_code)
            if not response.ok:
                self.logger.error("Response content: %s", response.text)

            response.raise_for_status()

            result: dict[str, Any] = response.json()
            self.logger.info("Successfully uploaded file: %s", name)
            return result

    except Exception as e:
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to upload file: {e}") from e

download_file

download_file(file_id: str, output_path: str | Path, create_dirs: bool = True) -> str

Download file by file_id from Silo storage.

Downloads a file from Silo storage to the local filesystem using the file's unique identifier.

Parameters:

Name	Type	Description	Default
file_id	str	ID of the file to download	required
output_path	str | Path	Path where to save the downloaded file	required
create_dirs	bool	Whether to create parent directories if they don't exist	True

Note

Wire endpoint: POST https://extapi.authentic8.com/getfile/ with form-data fields id (file ID) and auth (file token). This is not a JSON command-array request — it uses a separate endpoint and multipart form-data, not the standard /api/ command format. The response body is the raw file content (binary stream).

Returns:

Type	Description
str	Path of the downloaded file (as string)

Raises:

Type	Description
FileAPIError	If file download fails
ValidationError	If parameters are invalid

Example

downloaded_path = api.download_file( ... file_id="abc123", ... output_path="/home/user/downloads/report.pdf" ... ) print(f"File downloaded to: {downloaded_path}")

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.API_COMMAND, api_command="getfile")
def download_file(
    self,
    file_id: str,
    output_path: str | Path,
    create_dirs: bool = True,
) -> str:
    """Download file by file_id from Silo storage.

    Downloads a file from Silo storage to the local filesystem using the
    file's unique identifier.

    Args:
        file_id: ID of the file to download
        output_path: Path where to save the downloaded file
        create_dirs: Whether to create parent directories if they don't exist

    Note:
        Wire endpoint: ``POST https://extapi.authentic8.com/getfile/`` with
        form-data fields ``id`` (file ID) and ``auth`` (file token). This is
        **not** a JSON command-array request — it uses a separate endpoint
        and multipart form-data, not the standard ``/api/`` command format.
        The response body is the raw file content (binary stream).

    Returns:
        Path of the downloaded file (as string)

    Raises:
        FileAPIError: If file download fails
        ValidationError: If parameters are invalid

    Example:
        >>> downloaded_path = api.download_file(
        ...     file_id="abc123",
        ...     output_path="/home/user/downloads/report.pdf"
        ... )
        >>> print(f"File downloaded to: {downloaded_path}")
    """
    if not file_id or not isinstance(file_id, str):
        raise ValidationError("File ID must be a non-empty string")

    if not output_path:
        raise ValidationError("Output path must be provided")

    # Convert to Path object and resolve to prevent path traversal
    output_path_raw = str(output_path)
    output_path = Path(output_path).resolve()

    # Warn if the original path contained traversal components
    if ".." in output_path_raw:
        self.logger.warning(
            "Output path contained '..' components, resolved to: %s",
            output_path,
        )

    # Create parent directories if requested
    if create_dirs and output_path.parent != output_path:
        output_path.parent.mkdir(parents=True, exist_ok=True)

    url = f"{self.base_url.rstrip('/')}/getfile/"

    try:
        self.logger.debug("Downloading file ID: %s", file_id)

        data = {
            "id": file_id,
            "auth": self.auth_token,
        }

        # Route through the shared session so retry/proxy/User-Agent
        # configuration from BaseAPIClient is honored. Override
        # Content-Type (set to None to drop the session's default
        # "application/json") so requests encodes the form body correctly.
        response = self.session.post(
            url,
            data=data,
            timeout=self.timeout,
            stream=True,
            headers={"Content-Type": None},
        )
        response.raise_for_status()

        # Write file content
        try:
            with open(output_path, "wb") as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
        except Exception:
            self._cleanup_partial_file(output_path)
            raise

        self.logger.info("Successfully downloaded file to: %s", output_path)
        return str(output_path)

    except Exception as e:
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to download file: {e}") from e

find_files

find_files(bucket_id: str, name: str | None = None, path: str | None = None, file_type: str | None = None, created_before: str | None = None, created_after: str | None = None, size_min: int | None = None, size_max: int | None = None, limit: int | None = None, metadata_filters: dict[str, str] | None = None) -> list[dict[str, Any]]

Search Silo storage for files by various criteria.

Searches for files in the specified bucket using various filter criteria such as name patterns, paths, file types, creation dates, and file sizes.

Parameters:

Name	Type	Description	Default
bucket_id	str	ID of the bucket to search in.	required
name	str | None	Name pattern to search for (supports wildcards).	None
path	str | None	Path to search in (supports wildcards).	None
file_type	str | None	MIME type or file extension to filter by. Sent as "type" in the wire format (field name differs from this parameter name).	None
created_before	str | None	Search for files created before this date (ISO format). Sent as ":created_before" in the wire format (note the colon prefix on the wire field name).	None
created_after	str | None	Search for files created after this date (ISO format). Sent as ":created_after" in the wire format (note the colon prefix on the wire field name).	None
size_min	int | None	Minimum file size in bytes.	None
size_max	int | None	Maximum file size in bytes.	None
limit	int | None	Maximum number of results to return.	None
metadata_filters	dict[str, str] | None	Dictionary of metadata key-value pairs to filter by.	None

Note

Wire format parameter name differences:

• Python file_type → wire "type"
• Python created_before → wire ":created_before" (colon prefix)
• Python created_after → wire ":created_after" (colon prefix)

Wire format structure::

{
    "command": "findfiles",
    "bucket_id": "my_bucket",
    "type": "application/pdf",          # ← file_type → "type"
    ":created_before": "2026-01-01",    # ← colon prefix
    ":created_after": "2025-01-01"      # ← colon prefix
}

Returns:

Type	Description
list[dict[str, Any]]	List of dictionaries containing file information including:
list[dict[str, Any]]	file_id: Unique file identifier
list[dict[str, Any]]	name: File name
list[dict[str, Any]]	path: File path in storage
list[dict[str, Any]]	size: File size in bytes
list[dict[str, Any]]	content_type: MIME type
list[dict[str, Any]]	created_ts: Creation timestamp
list[dict[str, Any]]	modified_ts: Last modification timestamp
list[dict[str, Any]]	metadata: Custom metadata (if any)

Raises:

Type	Description
FileAPIError	If file search fails
ValidationError	If parameters are invalid

Example

files = api.find_files( ... bucket_id="documents", ... name=".pdf", ... path="/reports/", ... created_after="2024-01-01T00:00:00Z", ... size_min=1024, ... limit=100 ... ) for file in files: ... print(f"{file['name']} - {file['size']} bytes")

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.API_COMMAND, api_command="findfiles")
def find_files(
    self,
    bucket_id: str,
    name: str | None = None,
    path: str | None = None,
    file_type: str | None = None,
    created_before: str | None = None,
    created_after: str | None = None,
    size_min: int | None = None,
    size_max: int | None = None,
    limit: int | None = None,
    metadata_filters: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
    """Search Silo storage for files by various criteria.

    Searches for files in the specified bucket using various filter criteria
    such as name patterns, paths, file types, creation dates, and file sizes.

    Args:
        bucket_id: ID of the bucket to search in.
        name: Name pattern to search for (supports wildcards).
        path: Path to search in (supports wildcards).
        file_type: MIME type or file extension to filter by. Sent as
            ``"type"`` in the wire format (field name differs from this
            parameter name).
        created_before: Search for files created before this date (ISO
            format). Sent as ``":created_before"`` in the wire format
            (note the colon prefix on the wire field name).
        created_after: Search for files created after this date (ISO
            format). Sent as ``":created_after"`` in the wire format
            (note the colon prefix on the wire field name).
        size_min: Minimum file size in bytes.
        size_max: Maximum file size in bytes.
        limit: Maximum number of results to return.
        metadata_filters: Dictionary of metadata key-value pairs to filter by.

    Note:
        Wire format parameter name differences:

        - Python ``file_type`` → wire ``"type"``
        - Python ``created_before`` → wire ``":created_before"`` (colon prefix)
        - Python ``created_after`` → wire ``":created_after"`` (colon prefix)

        Wire format structure::

            {
                "command": "findfiles",
                "bucket_id": "my_bucket",
                "type": "application/pdf",          # ← file_type → "type"
                ":created_before": "2026-01-01",    # ← colon prefix
                ":created_after": "2025-01-01"      # ← colon prefix
            }

    Returns:
        List of dictionaries containing file information including:
        - file_id: Unique file identifier
        - name: File name
        - path: File path in storage
        - size: File size in bytes
        - content_type: MIME type
        - created_ts: Creation timestamp
        - modified_ts: Last modification timestamp
        - metadata: Custom metadata (if any)

    Raises:
        FileAPIError: If file search fails
        ValidationError: If parameters are invalid

    Example:
        >>> files = api.find_files(
        ...     bucket_id="documents",
        ...     name="*.pdf",
        ...     path="/reports/*",
        ...     created_after="2024-01-01T00:00:00Z",
        ...     size_min=1024,
        ...     limit=100
        ... )
        >>> for file in files:
        ...     print(f"{file['name']} - {file['size']} bytes")
    """
    if not bucket_id or not isinstance(bucket_id, str):
        raise ValidationError("Bucket ID must be a non-empty string")

    # Build search payload
    search_params: dict[str, Any] = {
        "command": "findfiles",
        "bucket_id": bucket_id,
    }

    # Add optional search criteria
    if name is not None:
        search_params["name"] = name

    if path is not None:
        search_params["path"] = path

    if file_type is not None:
        search_params["type"] = file_type

    if created_before is not None:
        search_params[":created_before"] = created_before

    if created_after is not None:
        search_params[":created_after"] = created_after

    if size_min is not None:
        if not isinstance(size_min, int) or size_min < 0:
            raise ValidationError("size_min must be a non-negative integer")
        search_params["size_min"] = size_min

    if size_max is not None:
        if not isinstance(size_max, int) or size_max < 0:
            raise ValidationError("size_max must be a non-negative integer")
        search_params["size_max"] = size_max

    if limit is not None:
        if not isinstance(limit, int) or limit <= 0:
            raise ValidationError("limit must be a positive integer")
        search_params["limit"] = limit

    if metadata_filters is not None:
        if not isinstance(metadata_filters, dict):
            raise ValidationError("metadata_filters must be a dictionary")
        # Add metadata filters to search parameters
        for key, value in metadata_filters.items():
            search_params[f"metadata:{key}"] = value

    payload = [search_params]

    try:
        self.logger.debug("Searching for files in bucket: %s", bucket_id)
        response = self._make_api_request("POST", "api/", self.auth_token, payload)

        # Extract results
        if len(response) >= 2 and isinstance(response[1], dict):
            result_entry = response[1]
            if "error" in result_entry:
                raise FileAPIError(f"API error: {result_entry['error']}")
            result = result_entry.get("result", [])
            if isinstance(result, list):
                self.logger.info("Found %d files matching criteria", len(result))
                return result

        self.logger.warning("Unexpected response format for find_files")
        return []

    except Exception as e:
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to search for files: {e}") from e

list_files

list_files(bucket_id: str, limit: int | None = None) -> list[dict[str, Any]]

List files in a bucket.

Lists files in the specified bucket with optional limit. This is a convenience wrapper around find_files with no filters.

Parameters:

Name	Type	Description	Default
bucket_id	str	ID of the bucket to list files from	required
limit	int | None	Maximum number of files to return	None

Returns:

Type	Description
list[dict[str, Any]]	List of dictionaries containing file information

Raises:

Type	Description
FileAPIError	If file listing fails
ValidationError	If parameters are invalid

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["findfiles"])
def list_files(
    self,
    bucket_id: str,
    limit: int | None = None,
) -> list[dict[str, Any]]:
    """List files in a bucket.

    Lists files in the specified bucket with optional limit.
    This is a convenience wrapper around find_files with no filters.

    Args:
        bucket_id: ID of the bucket to list files from
        limit: Maximum number of files to return

    Returns:
        List of dictionaries containing file information

    Raises:
        FileAPIError: If file listing fails
        ValidationError: If parameters are invalid
    """
    return self.find_files(bucket_id=bucket_id, limit=limit)

modify_file

modify_file(file_id: str, name: str | None = None, expire_ts: str | None = None, path: str | None = None, content_type: str | None = None) -> dict[str, Any]

Change properties of a file by file_id in Silo storage.

Modifies various properties of an existing file including its name, path, expiration time, and content type.

Parameters:

Name	Type	Description	Default
file_id	str	ID of the file to modify (required)	required
name	str | None	New name for the file (optional)	None
expire_ts	str | None	New file expire date (optional)	None
path	str | None	Path of new location to move file (optional)	None
content_type	str | None	New content type of modified file (optional)	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing the API response

Raises:

Type	Description
FileAPIError	If file modification fails
ValidationError	If parameters are invalid

Example

updated_file = api.modify_file( ... file_id="abc123", ... name="updated_report.pdf", ... expire_ts="now+3600", ... path="/reports/2024/updated/" ... ) print(f"Updated file: {updated_file}")

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.API_COMMAND, api_command="modifyfile")
def modify_file(
    self,
    file_id: str,
    name: str | None = None,
    expire_ts: str | None = None,
    path: str | None = None,
    content_type: str | None = None,
) -> dict[str, Any]:
    """Change properties of a file by file_id in Silo storage.

    Modifies various properties of an existing file including its name,
    path, expiration time, and content type.

    Args:
        file_id: ID of the file to modify (required)
        name: New name for the file (optional)
        expire_ts: New file expire date (optional)
        path: Path of new location to move file (optional)
        content_type: New content type of modified file (optional)

    Returns:
        Dictionary containing the API response

    Raises:
        FileAPIError: If file modification fails
        ValidationError: If parameters are invalid

    Example:
        >>> updated_file = api.modify_file(
        ...     file_id="abc123",
        ...     name="updated_report.pdf",
        ...     expire_ts="now+3600",
        ...     path="/reports/2024/updated/"
        ... )
        >>> print(f"Updated file: {updated_file}")
    """
    if not file_id or not isinstance(file_id, str):
        raise ValidationError("File ID must be a non-empty string")

    # Build modifyfile command — setauth is injected by _make_api_request
    modify_command: dict[str, Any] = {"command": "modifyfile", "file_id": file_id}

    # Add optional parameters
    if name is not None:
        if not isinstance(name, str) or not name:
            raise ValidationError("Name must be a non-empty string")
        modify_command["name"] = name

    if expire_ts is not None:
        modify_command["expire_ts"] = expire_ts

    if path is not None:
        if not isinstance(path, str):
            raise ValidationError("Path must be a string")
        modify_command["path"] = path

    if content_type is not None:
        if not isinstance(content_type, str):
            raise ValidationError("Content type must be a string")
        modify_command["content_type"] = content_type

    payload = [modify_command]

    try:
        self.logger.debug("Modifying file: %s", file_id)

        # Route through the shared request helper (like every other
        # command) instead of posting directly to the bare base_url,
        # which is missing the "api/" path segment the server expects.
        response = self._make_api_request("POST", "api/", self.auth_token, payload)

        # Check if modification was successful
        if len(response) >= 2 and isinstance(response[1], dict):
            result_entry = response[1]
            if "error" in result_entry:
                raise FileAPIError(f"API error: {result_entry['error']}")
            modify_result = result_entry.get("result")
            if modify_result and "update file" in str(modify_result):
                self.logger.info("Successfully modified file: %s", file_id)
                return {
                    "result": modify_result,
                    "file_id": file_id,
                    "response": response,
                }

        raise FileAPIError(f"Unexpected response format: {response}")

    except Exception as e:
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to modify file: {e}") from e

delete_file

delete_file(file_id: str) -> bool

Delete file in Silo storage by file_id.

Permanently deletes a file from Silo storage. This action cannot be undone.

Parameters:

Name	Type	Description	Default
file_id	str	ID of the file to delete	required

Returns:

Type	Description
bool	True if deletion was successful, False otherwise

Raises:

Type	Description
FileAPIError	If file deletion fails
ValidationError	If file_id is invalid

Example

success = api.delete_file("abc123") if success: ... print("File deleted successfully")

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.API_COMMAND, api_command="deletefile")
def delete_file(self, file_id: str) -> bool:
    """Delete file in Silo storage by file_id.

    Permanently deletes a file from Silo storage. This action cannot be undone.

    Args:
        file_id: ID of the file to delete

    Returns:
        True if deletion was successful, False otherwise

    Raises:
        FileAPIError: If file deletion fails
        ValidationError: If file_id is invalid

    Example:
        >>> success = api.delete_file("abc123")
        >>> if success:
        ...     print("File deleted successfully")
    """
    if not file_id or not isinstance(file_id, str):
        raise ValidationError("File ID must be a non-empty string")

    payload = [
        {
            "command": "deletefile",
            "file_id": file_id,
        }
    ]

    try:
        self.logger.debug("Deleting file: %s", file_id)
        response = self._make_api_request("POST", "api/", self.auth_token, payload)

        result = self._extract_api_result(response)
        if isinstance(result, str) and "deleted file" in result:
            self.logger.info("Successfully deleted file: %s", file_id)
            return True
        if isinstance(result, dict) and (
            result.get("deleted") or result.get("success")
        ):
            self.logger.info("Successfully deleted file: %s", file_id)
            return True

        self.logger.warning("File deletion may have failed: %s", result)
        return False

    except Exception as e:
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to delete file: {e}") from e

get_file_info

get_file_info(file_id: str, bucket_id: str | None = None) -> dict[str, Any]

Get detailed information about a file.

Retrieves information about a single file using a file_id-scoped findfiles lookup, rather than fetching the entire bucket listing and scanning it client-side.

Parameters:

Name	Type	Description	Default
file_id	str	ID of the file to get information for	required
bucket_id	str | None	Optional bucket ID to search in (if not provided, searches configured bucket)	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing detailed file information

Raises:

Type	Description
FileAPIError	If file information retrieval fails
ValidationError	If file_id is invalid

Example

file_info = api.get_file_info("abc123") print(f"File: {file_info['name']}") print(f"Size: {file_info['file_size']} bytes") print(f"Created: {file_info['create_ts']}")

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["findfiles"])
def get_file_info(
    self, file_id: str, bucket_id: str | None = None
) -> dict[str, Any]:
    """Get detailed information about a file.

    Retrieves information about a single file using a file_id-scoped
    findfiles lookup, rather than fetching the entire bucket listing
    and scanning it client-side.

    Args:
        file_id: ID of the file to get information for
        bucket_id: Optional bucket ID to search in (if not provided, searches configured bucket)

    Returns:
        Dictionary containing detailed file information

    Raises:
        FileAPIError: If file information retrieval fails
        ValidationError: If file_id is invalid

    Example:
        >>> file_info = api.get_file_info("abc123")
        >>> print(f"File: {file_info['name']}")
        >>> print(f"Size: {file_info['file_size']} bytes")
        >>> print(f"Created: {file_info['create_ts']}")
    """
    if not file_id or not isinstance(file_id, str):
        raise ValidationError("File ID must be a non-empty string")

    try:
        self.logger.debug("Getting file info for: %s", file_id)

        # Use the provided bucket_id or fall back to configured bucket
        search_bucket_id = bucket_id
        if not search_bucket_id:
            # Try to get from config if available
            search_bucket_id = self._config.get("BUCKET_ID")

        # Scope the findfiles lookup to this specific file_id instead of
        # requesting the entire bucket listing. This avoids the failure
        # mode where a server-side cap on findfiles results causes a
        # file that genuinely exists to be reported as "not found"
        # simply because it fell outside the returned page.
        find_command: dict[str, Any] = {
            "command": "findfiles",
            "file_id": file_id,
        }

        # Add bucket_id if available
        if search_bucket_id:
            find_command["bucket_id"] = search_bucket_id

        payload = [find_command]

        response = self._make_api_request("POST", "api/", self.auth_token, payload)

        # Extract results and find the matching file_id
        if len(response) >= 2 and isinstance(response[1], dict):
            result_entry = response[1]
            if "error" in result_entry:
                raise FileAPIError(f"API error: {result_entry['error']}")
            files_list = result_entry.get("result", [])
            if isinstance(files_list, list):
                # Find the file with matching file_id. The server-side
                # filter should already narrow this to a single entry,
                # but we still match explicitly in case the wire
                # implementation returns extra entries.
                for file_info in files_list:
                    if (
                        isinstance(file_info, dict)
                        and file_info.get("file_id") == file_id
                    ):
                        self.logger.debug("Found file information for: %s", file_id)
                        return file_info

                # File not found in this bucket
                bucket_msg = (
                    f" in bucket {search_bucket_id}" if search_bucket_id else ""
                )
                raise FileAPIError(f"File with ID {file_id} not found{bucket_msg}")

        raise FileAPIError(f"Unexpected response format: {response}")

    except Exception as e:
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to get file information: {e}") from e

download_file_with_retry_async async

download_file_with_retry_async(file_id: str, output_path: str | Path, max_retries: int = 3, timeout: int = 60, create_dirs: bool = True) -> str

Asynchronously download file by file_id with retry mechanism. Downloads a file from Silo storage with configurable retry logic and timeout handling. This async version allows for non-blocking downloads and can be used for concurrent file downloads.

Parameters:

Name	Type	Description	Default
file_id	str	ID of the file to download	required
output_path	str | Path	Path where to save the downloaded file	required
max_retries	int	Positive maximum number of retry attempts (default: 3)	3
timeout	int	Positive request timeout in seconds (default: 60)	60
create_dirs	bool	Whether to create parent directories if they don't exist	True

Returns:

Type	Description
str	Path of the downloaded file (as string)

Raises:

Type	Description
FileAPIError	If file download fails
ValidationError	If parameters are invalid

Example

import asyncio async def download_files(): ... files = FileAPI(config) ... downloaded_path = await files.download_file_with_retry_async( ... file_id="abc123", ... output_path="/home/user/downloads/report.pdf", ... max_retries=5 ... ) ... return downloaded_path path = asyncio.run(download_files())

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.ASYNC_VARIANT, api_command="getfile", sync_of="download_file")
async def download_file_with_retry_async(
    self,
    file_id: str,
    output_path: str | Path,
    max_retries: int = 3,
    timeout: int = 60,
    create_dirs: bool = True,
) -> str:
    """Asynchronously download file by file_id with retry mechanism.
    Downloads a file from Silo storage with configurable retry logic and
    timeout handling. This async version allows for non-blocking downloads
    and can be used for concurrent file downloads.

    Args:
        file_id: ID of the file to download
        output_path: Path where to save the downloaded file
        max_retries: Positive maximum number of retry attempts (default: 3)
        timeout: Positive request timeout in seconds (default: 60)
        create_dirs: Whether to create parent directories if they don't exist

    Returns:
        Path of the downloaded file (as string)

    Raises:
        FileAPIError: If file download fails
        ValidationError: If parameters are invalid

    Example:
        >>> import asyncio
        >>> async def download_files():
        ...     files = FileAPI(config)
        ...     downloaded_path = await files.download_file_with_retry_async(
        ...         file_id="abc123",
        ...         output_path="/home/user/downloads/report.pdf",
        ...         max_retries=5
        ...     )
        ...     return downloaded_path
        >>> path = asyncio.run(download_files())
    """
    validate_positive_int(max_retries, "max_retries")
    validate_positive_int(timeout, "timeout")

    if not file_id or not isinstance(file_id, str):
        raise ValidationError("File ID must be a non-empty string")

    if not output_path:
        raise ValidationError("Output path must be provided")

    # Convert to Path object
    output_path = Path(output_path)

    # Create parent directories if requested
    if create_dirs and output_path.parent != output_path:
        output_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        self.logger.debug(
            "Starting async download with retry for file ID: %s", file_id
        )

        # Implement async download with retry logic directly
        output_path = Path(output_path)
        if create_dirs and output_path.parent != output_path:
            output_path.parent.mkdir(parents=True, exist_ok=True)

        ssl_context = self.create_ssl_context()
        download_url = f"{self.base_url.rstrip('/')}/getfile/"

        async with httpx.AsyncClient(verify=ssl_context) as client:
            for attempt in range(max_retries):
                self.logger.debug(
                    f"Download attempt {attempt + 1} for file {file_id}"
                )
                try:
                    response = await client.post(
                        download_url,
                        data={"id": file_id, "auth": self.auth_token},
                        timeout=httpx.Timeout(timeout),
                    )
                    response.raise_for_status()
                    content = response.content

                    with open(output_path, "wb") as f:
                        f.write(content)

                    self.logger.info(
                        "Successfully downloaded file to: %s", output_path
                    )
                    return str(output_path)

                except httpx.HTTPError as e:
                    self.logger.error(
                        f"Error downloading file (attempt {attempt + 1}/{max_retries}): {e}"
                    )
                    if attempt == max_retries - 1:
                        raise FileAPIError(
                            f"Failed to download file after {max_retries} attempts: {e}"
                        ) from e
                    await asyncio.sleep(2**attempt)  # Exponential backoff

        # If we get here, all retries were exhausted without success
        raise FileAPIError(
            f"Failed to download file after {max_retries} attempts"
        )  # pragma: no cover

    except Exception as e:  # pragma: no cover
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to download file with retry: {e}") from e

download_file_chunked_async async

download_file_chunked_async(file_id: str, output_path: str | Path, chunk_size: int = 8192, timeout: int = 60, create_dirs: bool = True) -> str

Asynchronously download file by file_id using chunked download. Downloads a file from Silo storage using streaming/chunked approach, which is more memory-efficient for large files. This async version allows for non-blocking downloads of large files.

Parameters:

Name	Type	Description	Default
file_id	str	ID of the file to download	required
output_path	str | Path	Path where to save the downloaded file	required
chunk_size	int	Positive size of chunks to read at a time (default: 8192)	8192
timeout	int	Positive request timeout in seconds (default: 60)	60
create_dirs	bool	Whether to create parent directories if they don't exist	True

Returns:

Type	Description
str	Path of the downloaded file (as string)

Raises:

Type	Description
FileAPIError	If file download fails
ValidationError	If parameters are invalid

Example

import asyncio async def download_large_file(): ... files = FileAPI(config) ... downloaded_path = await files.download_file_chunked_async( ... file_id="large_file_123", ... output_path="/home/user/downloads/large_archive.zip", ... chunk_size=16384 ... ) ... return downloaded_path path = asyncio.run(download_large_file())

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.ASYNC_VARIANT, api_command="getfile", sync_of="download_file")
async def download_file_chunked_async(
    self,
    file_id: str,
    output_path: str | Path,
    chunk_size: int = 8192,
    timeout: int = 60,
    create_dirs: bool = True,
) -> str:
    """Asynchronously download file by file_id using chunked download.
    Downloads a file from Silo storage using streaming/chunked approach,
    which is more memory-efficient for large files. This async version
    allows for non-blocking downloads of large files.

    Args:
        file_id: ID of the file to download
        output_path: Path where to save the downloaded file
        chunk_size: Positive size of chunks to read at a time (default: 8192)
        timeout: Positive request timeout in seconds (default: 60)
        create_dirs: Whether to create parent directories if they don't exist

    Returns:
        Path of the downloaded file (as string)

    Raises:
        FileAPIError: If file download fails
        ValidationError: If parameters are invalid

    Example:
        >>> import asyncio
        >>> async def download_large_file():
        ...     files = FileAPI(config)
        ...     downloaded_path = await files.download_file_chunked_async(
        ...         file_id="large_file_123",
        ...         output_path="/home/user/downloads/large_archive.zip",
        ...         chunk_size=16384
        ...     )
        ...     return downloaded_path
        >>> path = asyncio.run(download_large_file())
    """
    validate_positive_int(chunk_size, "chunk_size")
    validate_positive_int(timeout, "timeout")

    if not file_id or not isinstance(file_id, str):
        raise ValidationError("File ID must be a non-empty string")

    if not output_path:
        raise ValidationError("Output path must be provided")

    # Convert to Path object
    output_path = Path(output_path)

    # Create parent directories if requested
    if create_dirs and output_path.parent != output_path:
        output_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        self.logger.debug(
            "Starting async chunked download for file ID: %s", file_id
        )

        # Implement async chunked download logic directly
        ssl_context = self.create_ssl_context()
        download_url = f"{self.base_url.rstrip('/')}/getfile/"

        async with httpx.AsyncClient(verify=ssl_context) as client:
            try:
                async with client.stream(
                    "POST",
                    download_url,
                    data={"id": file_id, "auth": self.auth_token},
                    timeout=httpx.Timeout(timeout),
                ) as response:
                    response.raise_for_status()

                    try:
                        with open(output_path, "wb") as f:
                            async for chunk in response.aiter_bytes(chunk_size):
                                f.write(chunk)
                    except Exception:
                        self._cleanup_partial_file(output_path)
                        raise

                    self.logger.info(
                        "Successfully downloaded file to: %s", output_path
                    )
                    return str(output_path)

            except httpx.HTTPError as e:
                raise FileAPIError(f"Failed to download file: {e}") from e

    except Exception as e:
        if isinstance(e, FileAPIError):
            raise
        raise FileAPIError(f"Failed to download file chunked: {e}") from e

bulk_download_async async

bulk_download_async(file_downloads: list[dict[str, Any]], max_concurrent: int = 5, use_chunked: bool = False) -> list[str | None]

Download multiple files concurrently.

Downloads multiple files in parallel with configurable concurrency limits to efficiently handle bulk file downloads.

Parameters:

Name	Type	Description	Default
file_downloads	list[dict[str, Any]]	List of download dictionaries, each containing: - file_id: ID of the file to download - output_path: Path where to save the file - max_retries: Optional max retries (default: 3) - timeout: Optional timeout (default: 60) - chunk_size: Optional chunk size for chunked downloads	required
max_concurrent	int	Positive maximum number of concurrent downloads (default: 5)	5
use_chunked	bool	Whether to use chunked download method (default: False)	False

Returns:

Type	Description
list[str | None]	List of downloaded file paths (same order as input), None for failed downloads

Raises:

Type	Description
ValidationError	If parameters are invalid

Example

import asyncio async def bulk_download(): ... files = FileAPI(config) ... downloads = [ ... {"file_id": "file1", "output_path": "./downloads/file1.pdf"}, ... {"file_id": "file2", "output_path": "./downloads/file2.zip"}, ... ] ... results = await files.bulk_download_async( ... file_downloads=downloads, ... max_concurrent=3 ... ) ... return results results = asyncio.run(bulk_download())

Source code in silo_sdk/storage/file_api.py

@api_tag(MethodType.ASYNC_VARIANT, api_command="getfile", sync_of="download_file")
async def bulk_download_async(
    self,
    file_downloads: list[dict[str, Any]],
    max_concurrent: int = 5,
    use_chunked: bool = False,
) -> list[str | None]:
    """Download multiple files concurrently.

    Downloads multiple files in parallel with configurable concurrency
    limits to efficiently handle bulk file downloads.

    Args:
        file_downloads: List of download dictionaries, each containing:
            - file_id: ID of the file to download
            - output_path: Path where to save the file
            - max_retries: Optional max retries (default: 3)
            - timeout: Optional timeout (default: 60)
            - chunk_size: Optional chunk size for chunked downloads
        max_concurrent: Positive maximum number of concurrent downloads (default: 5)
        use_chunked: Whether to use chunked download method (default: False)

    Returns:
        List of downloaded file paths (same order as input), None for failed downloads

    Raises:
        ValidationError: If parameters are invalid

    Example:
        >>> import asyncio
        >>> async def bulk_download():
        ...     files = FileAPI(config)
        ...     downloads = [
        ...         {"file_id": "file1", "output_path": "./downloads/file1.pdf"},
        ...         {"file_id": "file2", "output_path": "./downloads/file2.zip"},
        ...     ]
        ...     results = await files.bulk_download_async(
        ...         file_downloads=downloads,
        ...         max_concurrent=3
        ...     )
        ...     return results
        >>> results = asyncio.run(bulk_download())
    """
    import asyncio

    validate_positive_int(max_concurrent, "max_concurrent")

    # Create semaphore to limit concurrent downloads
    semaphore = asyncio.Semaphore(max_concurrent)

    async def download_single_file(
        download_config: dict[str, Any],
    ) -> str | None:
        """Download a single file with semaphore control."""
        async with semaphore:
            try:
                file_id = download_config["file_id"]
                output_path = download_config["output_path"]
                max_retries = download_config.get("max_retries", 3)
                timeout = download_config.get("timeout", 60)

                if use_chunked:
                    chunk_size = download_config.get("chunk_size", 8192)
                    return await self.download_file_chunked_async(
                        file_id=file_id,
                        output_path=output_path,
                        chunk_size=chunk_size,
                        timeout=timeout,
                    )
                else:
                    return await self.download_file_with_retry_async(
                        file_id=file_id,
                        output_path=output_path,
                        max_retries=max_retries,
                        timeout=timeout,
                    )
            except Exception as e:
                self.logger.error(
                    f"Failed to download {download_config.get('file_id')}: {e}"
                )
                return None

    # Process all downloads concurrently
    self.logger.info(f"Starting bulk download of {len(file_downloads)} files")
    results = await asyncio.gather(
        *[download_single_file(download) for download in file_downloads],
        return_exceptions=True,
    )

    # Convert exceptions to None and log errors
    processed_results: list[str | None] = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            self.logger.error(f"Download {i + 1} failed with exception: {result}")
            processed_results.append(None)
        elif isinstance(result, str):
            processed_results.append(result)
        else:
            processed_results.append(None)

    successful_downloads = sum(1 for r in processed_results if r is not None)
    self.logger.info(
        f"Completed bulk download: {successful_downloads}/{len(file_downloads)} "
        "files successful"
    )

    return processed_results