quixportal 2.0.1

A Python library for creating fsspec filesystems from JSON blob storage configurations

Quix Blob Storage

A Python library for creating fsspec filesystems from JSON blob storage configurations.

Overview

This library provides a clean, type-safe interface for creating fsspec filesystems from blob storage configurations. It reads configuration from environment variables or direct input and returns appropriate filesystem instances. The library supports multiple cloud storage providers and includes comprehensive error handling and logging.

Features

• Multi-provider support: S3, S3-compatible services (MinIO), Azure Blob Storage, Google Cloud Storage, Local filesystem
• Type-safe configuration: Pydantic models for configuration validation
• Multiple configuration sources: Environment variables, dictionaries, JSON strings
• Clean API: FilesystemFactory class for advanced usage, convenience functions for simple cases
• Comprehensive logging: Structured logging with credential masking
• Error handling: Custom exceptions with meaningful error messages
• Connection validation: Optional connection testing during filesystem creation
• Production-ready: Built with reliability and observability in mind

Installation

# Install with uv
uv add quixportal

# Install with pip
pip install quixportal

# Install with specific provider extras
uv add "quixportal[s3,azure,gcp]"

Quick Start

Simple Usage (Environment Variable)

import os
from quixportal import get_filesystem

# Set configuration in environment variable
os.environ["Quix__BlobStorage__Connection__Json"] = '''{
    "provider": "S3",
    "s3Compatible": {
        "bucketName": "my-bucket",
        "accessKeyId": "your-access-key",
        "secretAccessKey": "your-secret-key",
        "region": "us-west-2",
        "serviceUrl": "https://s3.amazonaws.com"
    }
}'''

# Create filesystem using convenience function
fs = get_filesystem()

# Use the filesystem
files = fs.ls("my-bucket/")
with fs.open("my-bucket/file.txt", "r") as f:
    content = f.read()

Advanced Usage (FilesystemFactory)

from quixportal import FilesystemFactory, ConfigurationError

# Create factory instance
factory = FilesystemFactory(enable_connection_testing=True)

# Configuration dictionary
config = {
    "provider": "S3",
    "s3Compatible": {
        "bucketName": "my-bucket",
        "accessKeyId": "your-access-key",
        "secretAccessKey": "your-secret-key",
        "region": "us-west-2",
        "serviceUrl": "https://s3.amazonaws.com"
    }
}

try:
    # Create filesystem from configuration dict
    fs = factory.get_filesystem_from_config(config)

    # Or from JSON string
    import json
    fs = factory.get_filesystem_from_json(json.dumps(config))

    # Or from environment variable
    fs = factory.get_filesystem()

except ConfigurationError as e:
    print(f"Configuration error: {e}")

Configuration

Environment Variable Configuration

Set the Quix__BlobStorage__Connection__Json environment variable:

export Quix__BlobStorage__Connection__Json='{"provider": "S3", "s3_compatible": {...}}'

Configuration Schema

S3 / S3-Compatible (including MinIO)

{
  "provider": "S3",
  "s3Compatible": {
    "bucketName": "my-bucket",
    "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
    "secretAccessKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    "region": "us-west-2",
    "serviceUrl": "https://s3.amazonaws.com"
  }
}

Azure Blob Storage

{
  "provider": "Azure",
  "azureBlobStorage": {
    "accountName": "mystorageaccount",
    "key": "YXp1cmUgYWNjb3VudCBrZXk=",
    "containerName": "mycontainer"
  }
}

MinIO Example

{
  "provider": "Minio",
  "s3Compatible": {
    "bucketName": "test-bucket",
    "accessKeyId": "minioadmin",
    "secretAccessKey": "minioadmin",
    "region": "us-east-1",
    "serviceUrl": "http://localhost:9000"
  }
}

Local Filesystem Storage

The Local provider uses fsspec's DirFileSystem to provide a blob-storage-like interface for local filesystem operations. All file operations are automatically scoped to the configured directory, providing a consistent API across all storage providers.

{
  "provider": "Local",
  "localStorage": {
    "directoryPath": "/data/storage"
  }
}

Key Features:

• Automatic Path Scoping: All operations are relative to the configured directory
• Blob Storage Interface: Use relative paths like "data/file.txt" instead of absolute paths
• Full fsspec Feature Set: Supports advanced operations like glob(), find(), copy(), move(), etc.
• Consistent API: Same interface as cloud storage providers for easy switching

Example Usage:

# Configuration for local storage
config = {
    "provider": "Local", 
    "localStorage": {"directoryPath": "/data/app-storage"}
}

fs = get_filesystem_from_config(config)

# All operations are scoped to /data/app-storage
fs.write("logs/app.log", "Log entry")  # Creates /data/app-storage/logs/app.log
files = fs.glob("**/*.log")            # Find all .log files recursively
fs.copy("data.txt", "backup/data.txt") # Copy within the scoped directory

Supported Providers

• S3: Amazon S3
• S3Compatible: Generic S3-compatible services
• Minio: MinIO object storage
• Azure: Azure Blob Storage
• Gcp: Google Cloud Storage
• Local: Local filesystem storage (fsspec-based with automatic path scoping)

Generating JSON Connection Strings

The library provides helper functions to programmatically generate JSON connection strings that are compatible with the Quix__BlobStorage__Connection__Json environment variable.

Quick Generation with generate_connection_json

from quixportal.storage import generate_connection_json

# Generate S3 connection JSON
s3_json = generate_connection_json(
    provider="S3",
    bucket_name="my-bucket",
    access_key_id="AKIAIOSFODNN7EXAMPLE",
    secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    region="us-west-2",
)

# Generate MinIO connection JSON
minio_json = generate_connection_json(
    provider="Minio",
    bucket_name="test-bucket",
    access_key_id="minioadmin",
    secret_access_key="minioadmin",
    service_url="http://localhost:9000",
)

# Generate Azure connection JSON
azure_json = generate_connection_json(
    provider="Azure",
    container_name="data-container",
    account_name="mystorageaccount",
    key="connection-string-or-key",
)

# Generate Local storage connection JSON
local_json = generate_connection_json(
    provider="Local",
    directory_path="/data/storage",
)

# Set as environment variable
import os
os.environ["Quix__BlobStorage__Connection__Json"] = s3_json

Builder Functions for Typed Configuration

For more control and type safety, use the builder functions:

from quixportal.storage import (
    create_s3_config,
    create_minio_config,
    create_azure_config,
    create_local_config,
    config_to_json,
    config_to_dict,
)

# Create and validate an S3 configuration
config = create_s3_config(
    bucket_name="my-bucket",
    access_key_id="AKIAIOSFODNN7EXAMPLE",
    secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    region="us-west-2",
    service_url="https://s3.amazonaws.com",
)

# Convert to JSON string for environment variable
json_string = config_to_json(config)
print(json_string)

# Or convert to dictionary for programmatic use
config_dict = config_to_dict(config)

Available Builder Functions

Function	Provider	Required Parameters
create_s3_config()	S3	bucket_name, access_key_id, secret_access_key
create_s3_compatible_config()	S3Compatible	bucket_name, access_key_id, secret_access_key, service_url
create_minio_config()	Minio	bucket_name, access_key_id, secret_access_key, service_url
create_azure_config()	Azure	container_name, account_name, key
create_local_config()	Local	directory_path

Development

This project uses uv for dependency management.

# Install dependencies
uv sync --dev

# Run tests
uv run pytest

# Run linting
uv run ruff check
uv run mypy .

# Format code
uv run ruff format

Support

For questions and support, please contact [e-mail].