Manage Files with the Kiteworks REST API

import requests
from get_access_token import KWOAuthClient

def list_files_in_folder(base_url, access_token, folder_id):
    """Return all files contained within the given folder."""
    url = f"{base_url}/rest/folders/{folder_id}/files"
    headers = {
        "X-Accellion-Version": "28",
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }
    params = {
        # "name": "example.docx",
        # "userId": "uuid0002-...",
        # "orderBy": "name:asc",
        # "offset": 0,
        # "limit": 10,
    }
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    return response.json().get("data", [])

def main():
    base_url = input("Enter Base URL for Kiteworks instance: ")
    client = KWOAuthClient(base_url)
    access_token = client.get_access_token()
    folder_id = input("Enter folder ID: ")
    files = list_files_in_folder(base_url, access_token, folder_id)
    print(files)

if __name__ == "__main__":
    main()

import requests
from urllib.parse import quote
from get_access_token import KWOAuthClient

def get_file_id_by_path(base_url, access_token, kw_path_to_file):
    """Return the file ID for a given Kiteworks file path."""
    headers = {
        "X-Accellion-Version": "28",
        "Authorization": f"Bearer {access_token}",
    }
    encoded_path = quote(kw_path_to_file)
    url = f"{base_url}/rest/search?path={encoded_path}"

    response = requests.get(url, headers=headers)
    response.raise_for_status()
    data = response.json()

    if data.get("files"):
        return data["files"][0]["id"]

    raise ValueError(f"No file found at path: {kw_path_to_file}")

def main():
    base_url = input("Enter Base URL for Kiteworks instance: ")
    client = KWOAuthClient(base_url)
    access_token = client.get_access_token()
    path = input("Enter Kiteworks file path (e.g. MyFolder/Reports/Q1.pdf): ")
    file_id = get_file_id_by_path(base_url, access_token, path)
    print(f"File ID: {file_id}")

if __name__ == "__main__":
    main()

import requests
from urllib.parse import quote
from get_access_token import KWOAuthClient

def download_file(base_url, access_token, kw_path_to_file, dest_file_path):
    """
    Download a file from Kiteworks to a local path.

    kw_path_to_file: Full Kiteworks path ending in the file name,
                     e.g. "MyFolder/Reports/Q1.pdf"
    dest_file_path:  Local destination path, e.g. "/tmp/Q1.pdf"
    """
    headers = {
        "X-Accellion-Version": "28",
        "Authorization": f"Bearer {access_token}",
    }

    # Step 1: Look up the file ID from its path
    encoded_path = quote(kw_path_to_file)
    search_url = f"{base_url}/rest/search?path={encoded_path}"
    search_response = requests.get(search_url, headers=headers)
    search_response.raise_for_status()
    search_data = search_response.json()

    if not search_data.get("files"):
        raise ValueError(f"No file found at path: {kw_path_to_file}")

    file_id = search_data["files"][0]["id"]

    # Step 2: Download the file content
    download_url = f"{base_url}/rest/files/{file_id}/content"
    response = requests.get(download_url, headers=headers)
    response.raise_for_status()

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

    print(f"File downloaded successfully to {dest_file_path}")

def main():
    base_url = input("Enter Base URL for Kiteworks instance: ")
    client = KWOAuthClient(base_url)
    access_token = client.get_access_token()
    kw_path = input("Enter Kiteworks file path (e.g. MyFolder/Reports/Q1.pdf): ")
    dest = input("Enter local destination path (e.g. /tmp/Q1.pdf): ")
    download_file(base_url, access_token, kw_path, dest)

if __name__ == "__main__":
    main()

import os
import json
import requests
from urllib.parse import quote
from get_access_token import KWOAuthClient

def upload_file(base_url, access_token, path_to_file, path_to_upload_folder, new_file_name):
    """
    Upload a local file to a Kiteworks folder.

    path_to_file:          Local file path, e.g. "/tmp/report.pdf"
    path_to_upload_folder: Kiteworks folder path, e.g. "MyFolder/Reports"
    new_file_name:         Name the file will have on Kiteworks
    """
    headers = {
        "X-Accellion-Version": "28",
        "Authorization": f"Bearer {access_token}",
    }

    # Step 1: Resolve the folder ID from its path
    encoded_path = quote(path_to_upload_folder)
    search_url = f"{base_url}/rest/search?path={encoded_path}"
    search_response = requests.get(search_url, headers=headers)
    search_response.raise_for_status()
    search_data = search_response.json()

    if not search_data.get("folders"):
        raise ValueError(f"Folder not found at path: {path_to_upload_folder}")

    folder_id = search_data["folders"][0]["id"]

    # Step 2: Initiate the upload session
    file_size = os.path.getsize(path_to_file)
    initiate_url = f"{base_url}/rest/folders/{folder_id}/actions/initiateUpload"
    initiate_payload = {
        "filename": new_file_name,
        "totalSize": file_size,
        "totalChunks": 1,
    }
    initiate_response = requests.post(initiate_url, headers=headers, json=initiate_payload)
    initiate_response.raise_for_status()
    upload_id = initiate_response.json()["id"]

    # Step 3: Upload the file as a single chunk
    upload_url = f"{base_url}/rest/uploads/{upload_id}?returnEntity=true"
    with open(path_to_file, "rb") as f:
        payload = {
            "compressionMode": (None, "NORMAL"),
            "compressionSize": (None, str(file_size)),
            "originalSize": (None, str(file_size)),
            "index": (None, "1"),
            "lastChunk": (None, "1"),
            "content": (new_file_name, f),
        }
        upload_response = requests.post(upload_url, headers=headers, files=payload)

    upload_response.raise_for_status()
    print("File uploaded successfully.")
    print(json.dumps(upload_response.json(), indent=2))

def main():
    base_url = input("Enter Base URL for Kiteworks instance: ")
    client = KWOAuthClient(base_url)
    access_token = client.get_access_token()
    path_to_file = input("Enter path to local file: ")
    path_to_folder = input("Enter Kiteworks folder path (e.g. MyFolder/Reports): ")
    new_file_name = input("Enter the file name on Kiteworks: ")
    upload_file(base_url, access_token, path_to_file, path_to_folder, new_file_name)

if __name__ == "__main__":
    main()

{
  "id": 7890,
  "error": "OK",
  "totalSize": 2048,
  "totalChunks": 2,
  "uploadedSize": 1024,
  "uploadedChunks": 1,
  "completeOk": 0,
  "uri": "dacfs_upload1/rest/uploads/7890",
  "timestamp": "2020-05-15T08:11:07Z"
}