Data Store

The Data Store API allows you to browse files and folders within store connectors on the Airia platform. This is useful for exploring the contents of connected data sources like Google Drive, SharePoint, or other file storage systems.

Get Files and Folders

You can browse files and folders in a store connector using the get_files_and_folders method. This returns a list of items with metadata, processing status, and pagination information.

Basic Usage

from airia import AiriaClient

client = AiriaClient(api_key="your_api_key")

# Get files and folders from root level
response = client.data_store.store.get_files_and_folders(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e"
)

# Access the results
if response.files_and_folders:
    for item in response.files_and_folders:
        print(f"{item.name} ({item.type})")

        if item.type == "folder":
            print(f"  Contains {item.file_count} files, {item.folder_count} folders")
        elif item.type == "file":
            print(f"  Size: {item.size} bytes")
            print(f"  MIME Type: {item.mime_type}")
            print(f"  Status: {item.status}")

print(f"\nTotal items: {response.total_count}")

Browse a Specific Folder

Navigate into a specific folder by providing a folder_id:

from airia import AiriaClient

client = AiriaClient(api_key="your_api_key")

# Browse inside a specific folder
response = client.data_store.store.get_files_and_folders(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e",
    folder_id="3292db69-f365-4bb6-a2b6-a017a187fb77"
)

# Process the folder contents
for item in response.files_and_folders:
    print(f"- {item.name}")

Pagination

Use pagination parameters to navigate through large result sets:

from airia import AiriaClient

client = AiriaClient(api_key="your_api_key")

# Get first page of 50 items
response = client.data_store.store.get_files_and_folders(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e",
    page_number=1,
    page_size=50
)

# Calculate total pages
total_pages = (response.total_count + 49) // 50
print(f"Page 1 of {total_pages}")
print(f"Showing {len(response.files_and_folders)} of {response.total_count} items")

Sorting

Sort results by different fields and orders:

from airia import AiriaClient

client = AiriaClient(api_key="your_api_key")

# Sort by last sync time, descending
response = client.data_store.store.get_files_and_folders(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e",
    sort_by="lastSyncAt",
    sort_order="DESC"
)

# Sort by name, ascending
response = client.data_store.store.get_files_and_folders(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e",
    sort_by="name",
    sort_order="ASC"
)

Filtering

Filter results by specific criteria:

from airia import AiriaClient

client = AiriaClient(api_key="your_api_key")

# Filter by type to show only files
response = client.data_store.store.get_files_and_folders(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e",
    filter_by="type",
    filter_value="file"
)

Working with File Processing Status

Access detailed processing status information for files:

from airia import AiriaClient

client = AiriaClient(api_key="your_api_key")

response = client.data_store.store.get_files_and_folders(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e"
)

for item in response.files_and_folders:
    if item.type == "file" and item.ingestion_processing_statuses:
        print(f"\nFile: {item.name}")
        for status in item.ingestion_processing_statuses:
            print(f"  Processing Mode: {status.table_document_processing_mode}")
            print(f"  Status: {status.status}")

            if status.user_errors:
                print(f"  User Errors: {', '.join(status.user_errors)}")
            if status.system_errors:
                print(f"  System Errors: {', '.join(status.system_errors)}")

Async Usage

The async client provides the same functionality with asynchronous operations:

from airia import AiriaAsyncClient

async def browse_store():
    async with AiriaAsyncClient(api_key="your_api_key") as client:
        # Get files and folders asynchronously
        response = await client.data_store.store.get_files_and_folders(
            store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
            project_id="0196162b-1553-71b2-8ebf-44594717936e",
            page_number=1,
            page_size=50,
            sort_by="lastSyncAt",
            sort_order="DESC"
        )

        # Process results
        for item in response.files_and_folders:
            print(f"{item.name} ({item.type})")

        return response

# Run the async function
import asyncio
asyncio.run(browse_store())

Complete Example

Here's a complete example that navigates through folders recursively:

from airia import AiriaClient

client = AiriaClient(api_key="your_api_key")

def browse_folder(store_connector_id, project_id, folder_id=None, indent=0):
    """Recursively browse folders and display their contents."""
    response = client.data_store.store.get_files_and_folders(
        store_connector_id=store_connector_id,
        project_id=project_id,
        folder_id=folder_id,
        sort_by="name",
        sort_order="ASC"
    )

    if not response.files_and_folders:
        return

    for item in response.files_and_folders:
        prefix = "  " * indent

        if item.type == "folder":
            print(f"{prefix}📁 {item.name} ({item.file_count} files, {item.folder_count} folders)")
            # Recursively browse subfolders (uncomment to enable)
            # browse_folder(store_connector_id, project_id, item.id, indent + 1)
        elif item.type == "file":
            size_kb = item.size / 1024 if item.size else 0
            print(f"{prefix}📄 {item.name} ({size_kb:.1f} KB) - {item.status}")

# Browse from root
browse_folder(
    store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
    project_id="0196162b-1553-71b2-8ebf-44594717936e"
)

Error Handling

Handle common errors when browsing store connectors:

from airia import AiriaClient
from airia.exceptions import AiriaAPIError

client = AiriaClient(api_key="your_api_key")

try:
    response = client.data_store.store.get_files_and_folders(
        store_connector_id="02b2ef8b-8bff-42c3-9ddb-e325c893176e",
        project_id="0196162b-1553-71b2-8ebf-44594717936e",
        folder_id="invalid-folder-id"
    )
except AiriaAPIError as e:
    if e.status_code == 404:
        print("Store connector or folder not found")
    elif e.status_code == 403:
        print("Access forbidden - check permissions")
    else:
        print(f"API error: {e.message}")