| Name | chuk-virtual-fs JSON |
| Version |
0.2.1
JSON |
| download |
| home_page | None |
| Summary | A secure, modular virtual filesystem designed for AI agent sandboxes |
| upload_time | 2025-10-12 01:24:28 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.11 |
| license | MIT |
| keywords |
filesystem
virtual
sandbox
ai
security
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# chuk-virtual-fs: Modular Virtual Filesystem Library
A powerful, flexible virtual filesystem library for Python with advanced features, multiple storage providers, and robust security.
## 🌟 Key Features
### 🔧 Modular Design
- Pluggable storage providers
- Flexible filesystem abstraction
- Supports multiple backend implementations
### 💾 Storage Providers
- **Memory Provider**: In-memory filesystem for quick testing and lightweight use
- **SQLite Provider**: Persistent storage with SQLite database backend
- **Pyodide Provider**: Web browser filesystem integration
- **S3 Provider**: Cloud storage with AWS S3 or S3-compatible services
- **E2B Sandbox Provider**: Remote sandbox environment filesystem
- Easy to extend with custom providers
### 🔒 Advanced Security
- Multiple predefined security profiles
- Customizable access controls
- Path and file type restrictions
- Quota management
- Security violation tracking
### 🚀 Advanced Capabilities
- **Streaming Operations**: Memory-efficient streaming for large files with:
- Real-time progress tracking callbacks
- Atomic write safety (temp file + atomic move)
- Automatic error recovery and cleanup
- Support for both sync and async callbacks
- **Virtual Mounts**: Unix-like mounting system to combine multiple providers
- Snapshot and versioning support
- Template-based filesystem setup
- Flexible path resolution
- Comprehensive file and directory operations
- CLI tools for bucket management
## 📦 Installation
### From PyPI
```bash
pip install chuk-virtual-fs
```
### With Optional Dependencies
```bash
# Install with S3 support
pip install "chuk-virtual-fs[s3]"
# Using uv
uv pip install -e ".[s3]"
# Add S3 dependency to existing project
uv add . --optional s3
```
### For Development
```bash
# Clone the repository
git clone https://github.com/yourusername/chuk-virtual-fs.git
cd chuk-virtual-fs
# Install in development mode with all dependencies
pip install -e ".[dev,s3,e2b]"
# Using uv
uv pip install -e ".[dev,s3,e2b]"
```
## 🚀 Quick Start
### Basic Usage (Async)
The library uses async/await for all operations:
```python
from chuk_virtual_fs import AsyncVirtualFileSystem
import asyncio
async def main():
# Use async context manager
async with AsyncVirtualFileSystem(provider="memory") as fs:
# Create directories
await fs.mkdir("/home/user/documents")
# Write to a file
await fs.write_file("/home/user/documents/hello.txt", "Hello, Virtual World!")
# Read from a file
content = await fs.read_text("/home/user/documents/hello.txt")
print(content) # Outputs: Hello, Virtual World!
# List directory contents
files = await fs.ls("/home/user/documents")
print(files) # Outputs: ['hello.txt']
# Change directory
await fs.cd("/home/user/documents")
print(fs.pwd()) # Outputs: /home/user/documents
# Copy and move operations
await fs.cp("hello.txt", "hello_copy.txt")
await fs.mv("hello_copy.txt", "/home/user/hello_moved.txt")
# Find files matching pattern
results = await fs.find("*.txt", path="/home", recursive=True)
print(results) # Finds all .txt files under /home
# Run the async function
asyncio.run(main())
```
> **Note**: The library also provides a synchronous `VirtualFileSystem` alias for backward compatibility, but the async API (`AsyncVirtualFileSystem`) is recommended for new code and required for streaming and mount operations.
## 💾 Storage Providers
### Available Providers
The virtual filesystem supports multiple storage providers:
- **Memory**: In-memory storage (default)
- **SQLite**: SQLite database storage
- **S3**: AWS S3 or S3-compatible storage
- **Pyodide**: Native integration with Pyodide environment
- **E2B**: E2B Sandbox environments
### Using the S3 Provider
The S3 provider allows you to use AWS S3 or S3-compatible storage (like Tigris Storage) as the backend for your virtual filesystem.
#### Installation
```bash
# Install with S3 support
pip install "chuk-virtual-fs[s3]"
# Or with uv
uv pip install "chuk-virtual-fs[s3]"
```
#### Configuration
Create a `.env` file with your S3 credentials:
```ini
# AWS credentials for S3 provider
AWS_ACCESS_KEY_ID=your_access_key
AWS_SECRET_ACCESS_KEY=your_secret_key
AWS_REGION=us-east-1
# For S3-compatible storage (e.g., Tigris Storage)
AWS_ENDPOINT_URL_S3=https://your-endpoint.example.com
S3_BUCKET_NAME=your-bucket-name
```
#### Example Usage
```python
from dotenv import load_dotenv
from chuk_virtual_fs import VirtualFileSystem
# Load environment variables
load_dotenv()
# Create filesystem with S3 provider
fs = VirtualFileSystem("s3",
bucket_name="your-bucket-name",
prefix="your-prefix", # Optional namespace in bucket
endpoint_url="https://your-endpoint.example.com") # For S3-compatible storage
# Use the filesystem as normal
fs.mkdir("/projects")
fs.write_file("/projects/notes.txt", "Virtual filesystem backed by S3")
# List directory contents
print(fs.ls("/projects"))
```
### E2B Sandbox Provider Example
```python
import os
from dotenv import load_dotenv
# Load E2B API credentials from .env file
load_dotenv()
# Ensure E2B API key is set
if not os.getenv("E2B_API_KEY"):
raise ValueError("E2B_API_KEY must be set in .env file")
from chuk_virtual_fs import VirtualFileSystem
# Create a filesystem in an E2B sandbox
# API key will be automatically used from environment variables
fs = VirtualFileSystem("e2b", root_dir="/home/user/sandbox")
# Create project structure
fs.mkdir("/projects")
fs.mkdir("/projects/python")
# Write a Python script
fs.write_file("/projects/python/hello.py", 'print("Hello from E2B sandbox!")')
# List directory contents
print(fs.ls("/projects/python"))
# Execute code in the sandbox (if supported)
if hasattr(fs.provider, 'sandbox') and hasattr(fs.provider.sandbox, 'run_code'):
result = fs.provider.sandbox.run_code(
fs.read_file("/projects/python/hello.py")
)
print(result.logs)
```
#### E2B Authentication
To use the E2B Sandbox Provider, you need to:
1. Install the E2B SDK:
```bash
pip install e2b-code-interpreter
```
2. Create a `.env` file in your project root:
```
E2B_API_KEY=your_e2b_api_key_here
```
3. Make sure to add `.env` to your `.gitignore` to keep credentials private.
Note: You can obtain an E2B API key from the [E2B platform](https://e2b.dev).
## 🛡️ Security Features
The virtual filesystem provides robust security features to protect against common vulnerabilities and limit resource usage.
### Security Profiles
```python
from chuk_virtual_fs import VirtualFileSystem
# Create a filesystem with strict security
fs = VirtualFileSystem(
security_profile="strict",
security_max_file_size=1024 * 1024, # 1MB max file size
security_allowed_paths=["/home", "/tmp"]
)
# Attempt to write to a restricted path
fs.write_file("/etc/sensitive", "This will fail")
# Get security violations
violations = fs.get_security_violations()
```
### Available Security Profiles
- **default**: Standard security with moderate restrictions
- **strict**: High security with tight constraints
- **readonly**: Completely read-only, no modifications allowed
- **untrusted**: Highly restrictive environment for untrusted code
- **testing**: Relaxed security for development and testing
### Security Features
- File size and total storage quotas
- Path traversal protection
- Deny/allow path and pattern rules
- Security violation logging
- Read-only mode
## 🛠️ CLI Tools
### S3 Bucket Management CLI
The package includes a CLI tool for managing S3 buckets:
```bash
# List all buckets
python s3_bucket_cli.py list
# Create a new bucket
python s3_bucket_cli.py create my-bucket
# Show bucket information
python s3_bucket_cli.py info my-bucket --show-top 5
# List objects in a bucket
python s3_bucket_cli.py ls my-bucket --prefix data/
# Clear all objects in a bucket or prefix
python s3_bucket_cli.py clear my-bucket --prefix tmp/
# Delete a bucket (must be empty)
python s3_bucket_cli.py delete my-bucket
# Copy objects between buckets or prefixes
python s3_bucket_cli.py copy source-bucket dest-bucket --source-prefix data/ --dest-prefix backup/
```
## 📋 Advanced Features
### Snapshots
Create and restore filesystem snapshots:
```python
from chuk_virtual_fs import VirtualFileSystem
from chuk_virtual_fs.snapshot_manager import SnapshotManager
fs = VirtualFileSystem()
snapshot_mgr = SnapshotManager(fs)
# Create initial content
fs.mkdir("/home/user")
fs.write_file("/home/user/file.txt", "Original content")
# Create a snapshot
snapshot_id = snapshot_mgr.create_snapshot("initial_state", "Initial filesystem setup")
# Modify content
fs.write_file("/home/user/file.txt", "Modified content")
fs.write_file("/home/user/new_file.txt", "New file")
# List available snapshots
snapshots = snapshot_mgr.list_snapshots()
for snap in snapshots:
print(f"{snap['name']}: {snap['description']}")
# Restore to initial state
snapshot_mgr.restore_snapshot("initial_state")
# Verify restore
print(fs.read_file("/home/user/file.txt")) # Outputs: Original content
print(fs.get_node_info("/home/user/new_file.txt")) # Outputs: None
# Export a snapshot
snapshot_mgr.export_snapshot("initial_state", "/tmp/snapshot.json")
```
### Templates
Load filesystem structures from templates:
```python
from chuk_virtual_fs import VirtualFileSystem
from chuk_virtual_fs.template_loader import TemplateLoader
fs = VirtualFileSystem()
template_loader = TemplateLoader(fs)
# Define a template
project_template = {
"directories": [
"/projects/app",
"/projects/app/src",
"/projects/app/docs"
],
"files": [
{
"path": "/projects/app/README.md",
"content": "# ${project_name}\n\n${project_description}"
},
{
"path": "/projects/app/src/main.py",
"content": "def main():\n print('Hello from ${project_name}!')"
}
]
}
# Apply the template with variables
template_loader.apply_template(project_template, variables={
"project_name": "My App",
"project_description": "A sample project created with the virtual filesystem"
})
```
### Streaming Operations
Handle large files efficiently with streaming support, progress tracking, and atomic write safety:
```python
from chuk_virtual_fs import AsyncVirtualFileSystem
async def main():
async with AsyncVirtualFileSystem(provider="memory") as fs:
# Stream write with progress tracking
async def data_generator():
for i in range(1000):
yield f"Line {i}: {'x' * 1000}\n".encode()
# Track upload progress
def progress_callback(bytes_written, total_bytes):
if bytes_written % (100 * 1024) < 1024: # Every 100KB
print(f"Uploaded {bytes_written / 1024:.1f} KB...")
# Write large file with progress reporting and atomic safety
await fs.stream_write(
"/large_file.txt",
data_generator(),
progress_callback=progress_callback
)
# Stream read - process chunks as they arrive
total_bytes = 0
async for chunk in fs.stream_read("/large_file.txt", chunk_size=8192):
total_bytes += len(chunk)
# Process chunk without loading entire file
print(f"Processed {total_bytes} bytes")
# Run with asyncio
import asyncio
asyncio.run(main())
```
#### Progress Reporting
Track upload/download progress with callbacks:
```python
async def upload_with_progress():
async with AsyncVirtualFileSystem(provider="s3", bucket_name="my-bucket") as fs:
# Progress tracking with sync callback
def track_progress(bytes_written, total_bytes):
percent = (bytes_written / total_bytes * 100) if total_bytes > 0 else 0
print(f"Progress: {percent:.1f}% ({bytes_written:,} bytes)")
# Or use async callback
async def async_track_progress(bytes_written, total_bytes):
# Can perform async operations here
await update_progress_db(bytes_written, total_bytes)
# Stream large file with progress tracking
async def generate_data():
for i in range(10000):
yield f"Record {i}\n".encode()
await fs.stream_write(
"/exports/large_dataset.csv",
generate_data(),
progress_callback=track_progress # or async_track_progress
)
```
#### Atomic Write Safety
All streaming writes use atomic operations to prevent file corruption:
```python
async def safe_streaming():
async with AsyncVirtualFileSystem(provider="filesystem", root_path="/data") as fs:
# Streaming write is automatically atomic:
# 1. Writes to temporary file (.tmp_*)
# 2. Atomically moves to final location on success
# 3. Auto-cleanup of temp files on failure
try:
await fs.stream_write("/critical_data.json", data_stream())
# File appears atomically - never partially written
except Exception as e:
# On failure, no partial file exists
# Temp files are automatically cleaned up
print(f"Upload failed safely: {e}")
```
#### Provider-Specific Features
Different providers implement atomic writes differently:
| Provider | Atomic Write Method | Progress Support |
|----------|-------------------|------------------|
| **Memory** | Temp buffer → swap | ✅ Yes |
| **Filesystem** | Temp file → `os.replace()` (OS-level atomic) | ✅ Yes |
| **SQLite** | Temp file → atomic move | ✅ Yes |
| **S3** | Multipart upload (inherently atomic) | ✅ Yes |
| **E2B Sandbox** | Temp file → `mv` command (atomic) | ✅ Yes |
**Key Features:**
- Memory-efficient processing of large files
- Real-time progress tracking with callbacks
- Atomic write safety prevents corruption
- Automatic temp file cleanup on errors
- Customizable chunk sizes
- Works with all storage providers
- Perfect for streaming uploads/downloads
- Both sync and async callback support
### Virtual Mounts
Combine multiple storage providers in a single filesystem:
```python
from chuk_virtual_fs import AsyncVirtualFileSystem
async def main():
async with AsyncVirtualFileSystem(
provider="memory",
enable_mounts=True
) as fs:
# Mount S3 bucket at /cloud
await fs.mount(
"/cloud",
provider="s3",
bucket_name="my-bucket",
endpoint_url="https://my-endpoint.com"
)
# Mount local filesystem at /local
await fs.mount(
"/local",
provider="filesystem",
root_path="/tmp/storage"
)
# Now use paths transparently across providers
await fs.write_file("/cloud/data.txt", "Stored in S3")
await fs.write_file("/local/cache.txt", "Stored locally")
await fs.write_file("/memory.txt", "Stored in memory")
# List all active mounts
mounts = fs.list_mounts()
for mount in mounts:
print(f"{mount['mount_point']}: {mount['provider']}")
# Copy between providers seamlessly
await fs.cp("/cloud/data.txt", "/local/backup.txt")
# Unmount when done
await fs.unmount("/cloud")
import asyncio
asyncio.run(main())
```
**Key Features:**
- Unix-like mount system
- Transparent path routing to correct provider
- Combine cloud, local, and in-memory storage
- Read-only mount support
- Seamless cross-provider operations (copy, move)
## 📖 API Reference
### Core Methods
#### Basic Operations
- `mkdir(path)`: Create a directory
- `touch(path)`: Create an empty file
- `write_file(path, content)`: Write content to a file
- `read_file(path)`: Read content from a file
- `ls(path)`: List directory contents
- `cd(path)`: Change current directory
- `pwd()`: Get current directory
- `rm(path)`: Remove a file or directory
- `cp(source, destination)`: Copy a file or directory
- `mv(source, destination)`: Move a file or directory
- `find(path, recursive)`: Find files and directories
- `search(path, pattern, recursive)`: Search for files matching a pattern
- `get_node_info(path)`: Get information about a node
- `get_fs_info()`: Get comprehensive filesystem information
#### Streaming Operations
- `stream_write(path, stream, chunk_size=8192, progress_callback=None, **metadata)`: Write from async iterator
- `progress_callback`: Optional callback function `(bytes_written, total_bytes) -> None`
- Supports both sync and async callbacks
- Atomic write safety with automatic temp file cleanup
- `stream_read(path, chunk_size=8192)`: Read as async iterator
#### Mount Management
- `mount(mount_point, provider, **provider_kwargs)`: Mount a provider at a path
- `unmount(mount_point)`: Unmount a provider
- `list_mounts()`: List all active mounts
## 🔍 Use Cases
- **Large File Processing**: Stream large files (GB+) without memory constraints
- Real-time progress tracking for user feedback
- Atomic writes prevent corruption on network failures
- Perfect for video uploads, data exports, log processing
- **Multi-Provider Storage**: Combine local, cloud, and in-memory storage seamlessly
- **Cloud Data Pipelines**: Stream data between S3, local storage, and processing systems
- Monitor upload/download progress
- Automatic retry and recovery with atomic operations
- Development sandboxing and isolated code execution
- Educational environments and web-based IDEs
- Reproducible computing environments
- Testing and simulation with multiple storage backends
- Cloud storage abstraction for provider-agnostic applications
## 💡 Requirements
- Python 3.8+
- Optional dependencies:
- `sqlite3` for SQLite provider
- `boto3` for S3 provider
- `e2b-code-interpreter` for E2B sandbox provider
## 🤝 Contributing
Contributions are welcome! Please submit pull requests or open issues on our GitHub repository.
## 📄 License
MIT License
## 🚨 Disclaimer
This library provides a flexible virtual filesystem abstraction. Always validate and sanitize inputs in production environments.
Raw data
{
"_id": null,
"home_page": null,
"name": "chuk-virtual-fs",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "filesystem, virtual, sandbox, ai, security",
"author": null,
"author_email": "Chris Hay <your.email@example.com>",
"download_url": "https://files.pythonhosted.org/packages/b5/87/f4db6589a8bea9236f9486212b6ba221b77a5961b968af495271f22f0ad3/chuk_virtual_fs-0.2.1.tar.gz",
"platform": null,
"description": "# chuk-virtual-fs: Modular Virtual Filesystem Library\n\nA powerful, flexible virtual filesystem library for Python with advanced features, multiple storage providers, and robust security.\n\n## \ud83c\udf1f Key Features\n\n### \ud83d\udd27 Modular Design\n- Pluggable storage providers\n- Flexible filesystem abstraction\n- Supports multiple backend implementations\n\n### \ud83d\udcbe Storage Providers\n- **Memory Provider**: In-memory filesystem for quick testing and lightweight use\n- **SQLite Provider**: Persistent storage with SQLite database backend\n- **Pyodide Provider**: Web browser filesystem integration\n- **S3 Provider**: Cloud storage with AWS S3 or S3-compatible services\n- **E2B Sandbox Provider**: Remote sandbox environment filesystem\n- Easy to extend with custom providers\n\n### \ud83d\udd12 Advanced Security\n- Multiple predefined security profiles\n- Customizable access controls\n- Path and file type restrictions\n- Quota management\n- Security violation tracking\n\n### \ud83d\ude80 Advanced Capabilities\n- **Streaming Operations**: Memory-efficient streaming for large files with:\n - Real-time progress tracking callbacks\n - Atomic write safety (temp file + atomic move)\n - Automatic error recovery and cleanup\n - Support for both sync and async callbacks\n- **Virtual Mounts**: Unix-like mounting system to combine multiple providers\n- Snapshot and versioning support\n- Template-based filesystem setup\n- Flexible path resolution\n- Comprehensive file and directory operations\n- CLI tools for bucket management\n\n## \ud83d\udce6 Installation\n\n### From PyPI\n\n```bash\npip install chuk-virtual-fs\n```\n\n### With Optional Dependencies\n\n```bash\n# Install with S3 support\npip install \"chuk-virtual-fs[s3]\"\n\n# Using uv\nuv pip install -e \".[s3]\"\n\n# Add S3 dependency to existing project\nuv add . --optional s3\n```\n\n### For Development\n\n```bash\n# Clone the repository\ngit clone https://github.com/yourusername/chuk-virtual-fs.git\ncd chuk-virtual-fs\n\n# Install in development mode with all dependencies\npip install -e \".[dev,s3,e2b]\"\n\n# Using uv\nuv pip install -e \".[dev,s3,e2b]\"\n```\n\n## \ud83d\ude80 Quick Start\n\n### Basic Usage (Async)\n\nThe library uses async/await for all operations:\n\n```python\nfrom chuk_virtual_fs import AsyncVirtualFileSystem\nimport asyncio\n\nasync def main():\n # Use async context manager\n async with AsyncVirtualFileSystem(provider=\"memory\") as fs:\n\n # Create directories\n await fs.mkdir(\"/home/user/documents\")\n\n # Write to a file\n await fs.write_file(\"/home/user/documents/hello.txt\", \"Hello, Virtual World!\")\n\n # Read from a file\n content = await fs.read_text(\"/home/user/documents/hello.txt\")\n print(content) # Outputs: Hello, Virtual World!\n\n # List directory contents\n files = await fs.ls(\"/home/user/documents\")\n print(files) # Outputs: ['hello.txt']\n\n # Change directory\n await fs.cd(\"/home/user/documents\")\n print(fs.pwd()) # Outputs: /home/user/documents\n\n # Copy and move operations\n await fs.cp(\"hello.txt\", \"hello_copy.txt\")\n await fs.mv(\"hello_copy.txt\", \"/home/user/hello_moved.txt\")\n\n # Find files matching pattern\n results = await fs.find(\"*.txt\", path=\"/home\", recursive=True)\n print(results) # Finds all .txt files under /home\n\n# Run the async function\nasyncio.run(main())\n```\n\n> **Note**: The library also provides a synchronous `VirtualFileSystem` alias for backward compatibility, but the async API (`AsyncVirtualFileSystem`) is recommended for new code and required for streaming and mount operations.\n\n## \ud83d\udcbe Storage Providers\n\n### Available Providers\n\nThe virtual filesystem supports multiple storage providers:\n\n- **Memory**: In-memory storage (default)\n- **SQLite**: SQLite database storage\n- **S3**: AWS S3 or S3-compatible storage\n- **Pyodide**: Native integration with Pyodide environment\n- **E2B**: E2B Sandbox environments\n\n### Using the S3 Provider\n\nThe S3 provider allows you to use AWS S3 or S3-compatible storage (like Tigris Storage) as the backend for your virtual filesystem.\n\n#### Installation\n\n```bash\n# Install with S3 support\npip install \"chuk-virtual-fs[s3]\"\n\n# Or with uv\nuv pip install \"chuk-virtual-fs[s3]\"\n```\n\n#### Configuration\n\nCreate a `.env` file with your S3 credentials:\n\n```ini\n# AWS credentials for S3 provider\nAWS_ACCESS_KEY_ID=your_access_key\nAWS_SECRET_ACCESS_KEY=your_secret_key\nAWS_REGION=us-east-1\n\n# For S3-compatible storage (e.g., Tigris Storage)\nAWS_ENDPOINT_URL_S3=https://your-endpoint.example.com\nS3_BUCKET_NAME=your-bucket-name\n```\n\n#### Example Usage\n\n```python\nfrom dotenv import load_dotenv\nfrom chuk_virtual_fs import VirtualFileSystem\n\n# Load environment variables\nload_dotenv()\n\n# Create filesystem with S3 provider\nfs = VirtualFileSystem(\"s3\", \n bucket_name=\"your-bucket-name\",\n prefix=\"your-prefix\", # Optional namespace in bucket\n endpoint_url=\"https://your-endpoint.example.com\") # For S3-compatible storage\n\n# Use the filesystem as normal\nfs.mkdir(\"/projects\")\nfs.write_file(\"/projects/notes.txt\", \"Virtual filesystem backed by S3\")\n\n# List directory contents\nprint(fs.ls(\"/projects\"))\n```\n\n### E2B Sandbox Provider Example\n\n```python\nimport os\nfrom dotenv import load_dotenv\n\n# Load E2B API credentials from .env file\nload_dotenv()\n\n# Ensure E2B API key is set\nif not os.getenv(\"E2B_API_KEY\"):\n raise ValueError(\"E2B_API_KEY must be set in .env file\")\n\nfrom chuk_virtual_fs import VirtualFileSystem\n\n# Create a filesystem in an E2B sandbox\n# API key will be automatically used from environment variables\nfs = VirtualFileSystem(\"e2b\", root_dir=\"/home/user/sandbox\")\n\n# Create project structure\nfs.mkdir(\"/projects\")\nfs.mkdir(\"/projects/python\")\n\n# Write a Python script\nfs.write_file(\"/projects/python/hello.py\", 'print(\"Hello from E2B sandbox!\")')\n\n# List directory contents\nprint(fs.ls(\"/projects/python\"))\n\n# Execute code in the sandbox (if supported)\nif hasattr(fs.provider, 'sandbox') and hasattr(fs.provider.sandbox, 'run_code'):\n result = fs.provider.sandbox.run_code(\n fs.read_file(\"/projects/python/hello.py\")\n )\n print(result.logs)\n```\n\n#### E2B Authentication\n\nTo use the E2B Sandbox Provider, you need to:\n\n1. Install the E2B SDK:\n ```bash\n pip install e2b-code-interpreter\n ```\n\n2. Create a `.env` file in your project root:\n ```\n E2B_API_KEY=your_e2b_api_key_here\n ```\n\n3. Make sure to add `.env` to your `.gitignore` to keep credentials private.\n\nNote: You can obtain an E2B API key from the [E2B platform](https://e2b.dev).\n\n## \ud83d\udee1\ufe0f Security Features\n\nThe virtual filesystem provides robust security features to protect against common vulnerabilities and limit resource usage.\n\n### Security Profiles\n\n```python\nfrom chuk_virtual_fs import VirtualFileSystem\n\n# Create a filesystem with strict security\nfs = VirtualFileSystem(\n security_profile=\"strict\",\n security_max_file_size=1024 * 1024, # 1MB max file size\n security_allowed_paths=[\"/home\", \"/tmp\"]\n)\n\n# Attempt to write to a restricted path\nfs.write_file(\"/etc/sensitive\", \"This will fail\")\n\n# Get security violations\nviolations = fs.get_security_violations()\n```\n\n### Available Security Profiles\n\n- **default**: Standard security with moderate restrictions\n- **strict**: High security with tight constraints\n- **readonly**: Completely read-only, no modifications allowed\n- **untrusted**: Highly restrictive environment for untrusted code\n- **testing**: Relaxed security for development and testing\n\n### Security Features\n\n- File size and total storage quotas\n- Path traversal protection\n- Deny/allow path and pattern rules\n- Security violation logging\n- Read-only mode\n\n## \ud83d\udee0\ufe0f CLI Tools\n\n### S3 Bucket Management CLI\n\nThe package includes a CLI tool for managing S3 buckets:\n\n```bash\n# List all buckets\npython s3_bucket_cli.py list\n\n# Create a new bucket\npython s3_bucket_cli.py create my-bucket\n\n# Show bucket information\npython s3_bucket_cli.py info my-bucket --show-top 5\n\n# List objects in a bucket\npython s3_bucket_cli.py ls my-bucket --prefix data/\n\n# Clear all objects in a bucket or prefix\npython s3_bucket_cli.py clear my-bucket --prefix tmp/\n\n# Delete a bucket (must be empty)\npython s3_bucket_cli.py delete my-bucket\n\n# Copy objects between buckets or prefixes\npython s3_bucket_cli.py copy source-bucket dest-bucket --source-prefix data/ --dest-prefix backup/\n```\n\n## \ud83d\udccb Advanced Features\n\n### Snapshots\n\nCreate and restore filesystem snapshots:\n\n```python\nfrom chuk_virtual_fs import VirtualFileSystem\nfrom chuk_virtual_fs.snapshot_manager import SnapshotManager\n\nfs = VirtualFileSystem()\nsnapshot_mgr = SnapshotManager(fs)\n\n# Create initial content\nfs.mkdir(\"/home/user\")\nfs.write_file(\"/home/user/file.txt\", \"Original content\")\n\n# Create a snapshot\nsnapshot_id = snapshot_mgr.create_snapshot(\"initial_state\", \"Initial filesystem setup\")\n\n# Modify content\nfs.write_file(\"/home/user/file.txt\", \"Modified content\")\nfs.write_file(\"/home/user/new_file.txt\", \"New file\")\n\n# List available snapshots\nsnapshots = snapshot_mgr.list_snapshots()\nfor snap in snapshots:\n print(f\"{snap['name']}: {snap['description']}\")\n\n# Restore to initial state\nsnapshot_mgr.restore_snapshot(\"initial_state\")\n\n# Verify restore\nprint(fs.read_file(\"/home/user/file.txt\")) # Outputs: Original content\nprint(fs.get_node_info(\"/home/user/new_file.txt\")) # Outputs: None\n\n# Export a snapshot\nsnapshot_mgr.export_snapshot(\"initial_state\", \"/tmp/snapshot.json\")\n```\n\n### Templates\n\nLoad filesystem structures from templates:\n\n```python\nfrom chuk_virtual_fs import VirtualFileSystem\nfrom chuk_virtual_fs.template_loader import TemplateLoader\n\nfs = VirtualFileSystem()\ntemplate_loader = TemplateLoader(fs)\n\n# Define a template\nproject_template = {\n \"directories\": [\n \"/projects/app\",\n \"/projects/app/src\",\n \"/projects/app/docs\"\n ],\n \"files\": [\n {\n \"path\": \"/projects/app/README.md\",\n \"content\": \"# ${project_name}\\n\\n${project_description}\"\n },\n {\n \"path\": \"/projects/app/src/main.py\",\n \"content\": \"def main():\\n print('Hello from ${project_name}!')\"\n }\n ]\n}\n\n# Apply the template with variables\ntemplate_loader.apply_template(project_template, variables={\n \"project_name\": \"My App\",\n \"project_description\": \"A sample project created with the virtual filesystem\"\n})\n```\n\n### Streaming Operations\n\nHandle large files efficiently with streaming support, progress tracking, and atomic write safety:\n\n```python\nfrom chuk_virtual_fs import AsyncVirtualFileSystem\n\nasync def main():\n async with AsyncVirtualFileSystem(provider=\"memory\") as fs:\n\n # Stream write with progress tracking\n async def data_generator():\n for i in range(1000):\n yield f\"Line {i}: {'x' * 1000}\\n\".encode()\n\n # Track upload progress\n def progress_callback(bytes_written, total_bytes):\n if bytes_written % (100 * 1024) < 1024: # Every 100KB\n print(f\"Uploaded {bytes_written / 1024:.1f} KB...\")\n\n # Write large file with progress reporting and atomic safety\n await fs.stream_write(\n \"/large_file.txt\",\n data_generator(),\n progress_callback=progress_callback\n )\n\n # Stream read - process chunks as they arrive\n total_bytes = 0\n async for chunk in fs.stream_read(\"/large_file.txt\", chunk_size=8192):\n total_bytes += len(chunk)\n # Process chunk without loading entire file\n\n print(f\"Processed {total_bytes} bytes\")\n\n# Run with asyncio\nimport asyncio\nasyncio.run(main())\n```\n\n#### Progress Reporting\n\nTrack upload/download progress with callbacks:\n\n```python\nasync def upload_with_progress():\n async with AsyncVirtualFileSystem(provider=\"s3\", bucket_name=\"my-bucket\") as fs:\n\n # Progress tracking with sync callback\n def track_progress(bytes_written, total_bytes):\n percent = (bytes_written / total_bytes * 100) if total_bytes > 0 else 0\n print(f\"Progress: {percent:.1f}% ({bytes_written:,} bytes)\")\n\n # Or use async callback\n async def async_track_progress(bytes_written, total_bytes):\n # Can perform async operations here\n await update_progress_db(bytes_written, total_bytes)\n\n # Stream large file with progress tracking\n async def generate_data():\n for i in range(10000):\n yield f\"Record {i}\\n\".encode()\n\n await fs.stream_write(\n \"/exports/large_dataset.csv\",\n generate_data(),\n progress_callback=track_progress # or async_track_progress\n )\n```\n\n#### Atomic Write Safety\n\nAll streaming writes use atomic operations to prevent file corruption:\n\n```python\nasync def safe_streaming():\n async with AsyncVirtualFileSystem(provider=\"filesystem\", root_path=\"/data\") as fs:\n\n # Streaming write is automatically atomic:\n # 1. Writes to temporary file (.tmp_*)\n # 2. Atomically moves to final location on success\n # 3. Auto-cleanup of temp files on failure\n\n try:\n await fs.stream_write(\"/critical_data.json\", data_stream())\n # File appears atomically - never partially written\n except Exception as e:\n # On failure, no partial file exists\n # Temp files are automatically cleaned up\n print(f\"Upload failed safely: {e}\")\n```\n\n#### Provider-Specific Features\n\nDifferent providers implement atomic writes differently:\n\n| Provider | Atomic Write Method | Progress Support |\n|----------|-------------------|------------------|\n| **Memory** | Temp buffer \u2192 swap | \u2705 Yes |\n| **Filesystem** | Temp file \u2192 `os.replace()` (OS-level atomic) | \u2705 Yes |\n| **SQLite** | Temp file \u2192 atomic move | \u2705 Yes |\n| **S3** | Multipart upload (inherently atomic) | \u2705 Yes |\n| **E2B Sandbox** | Temp file \u2192 `mv` command (atomic) | \u2705 Yes |\n\n**Key Features:**\n- Memory-efficient processing of large files\n- Real-time progress tracking with callbacks\n- Atomic write safety prevents corruption\n- Automatic temp file cleanup on errors\n- Customizable chunk sizes\n- Works with all storage providers\n- Perfect for streaming uploads/downloads\n- Both sync and async callback support\n\n### Virtual Mounts\n\nCombine multiple storage providers in a single filesystem:\n\n```python\nfrom chuk_virtual_fs import AsyncVirtualFileSystem\n\nasync def main():\n async with AsyncVirtualFileSystem(\n provider=\"memory\",\n enable_mounts=True\n ) as fs:\n\n # Mount S3 bucket at /cloud\n await fs.mount(\n \"/cloud\",\n provider=\"s3\",\n bucket_name=\"my-bucket\",\n endpoint_url=\"https://my-endpoint.com\"\n )\n\n # Mount local filesystem at /local\n await fs.mount(\n \"/local\",\n provider=\"filesystem\",\n root_path=\"/tmp/storage\"\n )\n\n # Now use paths transparently across providers\n await fs.write_file(\"/cloud/data.txt\", \"Stored in S3\")\n await fs.write_file(\"/local/cache.txt\", \"Stored locally\")\n await fs.write_file(\"/memory.txt\", \"Stored in memory\")\n\n # List all active mounts\n mounts = fs.list_mounts()\n for mount in mounts:\n print(f\"{mount['mount_point']}: {mount['provider']}\")\n\n # Copy between providers seamlessly\n await fs.cp(\"/cloud/data.txt\", \"/local/backup.txt\")\n\n # Unmount when done\n await fs.unmount(\"/cloud\")\n\nimport asyncio\nasyncio.run(main())\n```\n\n**Key Features:**\n- Unix-like mount system\n- Transparent path routing to correct provider\n- Combine cloud, local, and in-memory storage\n- Read-only mount support\n- Seamless cross-provider operations (copy, move)\n\n## \ud83d\udcd6 API Reference\n\n### Core Methods\n\n#### Basic Operations\n- `mkdir(path)`: Create a directory\n- `touch(path)`: Create an empty file\n- `write_file(path, content)`: Write content to a file\n- `read_file(path)`: Read content from a file\n- `ls(path)`: List directory contents\n- `cd(path)`: Change current directory\n- `pwd()`: Get current directory\n- `rm(path)`: Remove a file or directory\n- `cp(source, destination)`: Copy a file or directory\n- `mv(source, destination)`: Move a file or directory\n- `find(path, recursive)`: Find files and directories\n- `search(path, pattern, recursive)`: Search for files matching a pattern\n- `get_node_info(path)`: Get information about a node\n- `get_fs_info()`: Get comprehensive filesystem information\n\n#### Streaming Operations\n- `stream_write(path, stream, chunk_size=8192, progress_callback=None, **metadata)`: Write from async iterator\n - `progress_callback`: Optional callback function `(bytes_written, total_bytes) -> None`\n - Supports both sync and async callbacks\n - Atomic write safety with automatic temp file cleanup\n- `stream_read(path, chunk_size=8192)`: Read as async iterator\n\n#### Mount Management\n- `mount(mount_point, provider, **provider_kwargs)`: Mount a provider at a path\n- `unmount(mount_point)`: Unmount a provider\n- `list_mounts()`: List all active mounts\n\n## \ud83d\udd0d Use Cases\n\n- **Large File Processing**: Stream large files (GB+) without memory constraints\n - Real-time progress tracking for user feedback\n - Atomic writes prevent corruption on network failures\n - Perfect for video uploads, data exports, log processing\n- **Multi-Provider Storage**: Combine local, cloud, and in-memory storage seamlessly\n- **Cloud Data Pipelines**: Stream data between S3, local storage, and processing systems\n - Monitor upload/download progress\n - Automatic retry and recovery with atomic operations\n- Development sandboxing and isolated code execution\n- Educational environments and web-based IDEs\n- Reproducible computing environments\n- Testing and simulation with multiple storage backends\n- Cloud storage abstraction for provider-agnostic applications\n\n## \ud83d\udca1 Requirements\n\n- Python 3.8+\n- Optional dependencies:\n - `sqlite3` for SQLite provider\n - `boto3` for S3 provider\n - `e2b-code-interpreter` for E2B sandbox provider\n\n## \ud83e\udd1d Contributing\n\nContributions are welcome! Please submit pull requests or open issues on our GitHub repository.\n\n## \ud83d\udcc4 License\n\nMIT License\n\n## \ud83d\udea8 Disclaimer\n\nThis library provides a flexible virtual filesystem abstraction. Always validate and sanitize inputs in production environments.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A secure, modular virtual filesystem designed for AI agent sandboxes",
"version": "0.2.1",
"project_urls": {
"Bug Tracker": "https://github.com/chrishayuk/chuk-virtual-fs/issues",
"Documentation": "https://github.com/chrishayuk/chuk-virtual-fs#readme",
"Homepage": "https://github.com/chrishayuk/chuk-virtual-fs"
},
"split_keywords": [
"filesystem",
" virtual",
" sandbox",
" ai",
" security"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "28597f638cc346fefb93e8d6601b437eceaf365c24bf02da096eacab78f54ae0",
"md5": "a3902896e333302fd0d457746ef7f4c6",
"sha256": "132a31a5963e7cb25337aef20407a7533624b320849a7094e59086af77e86951"
},
"downloads": -1,
"filename": "chuk_virtual_fs-0.2.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a3902896e333302fd0d457746ef7f4c6",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 100852,
"upload_time": "2025-10-12T01:24:26",
"upload_time_iso_8601": "2025-10-12T01:24:26.905456Z",
"url": "https://files.pythonhosted.org/packages/28/59/7f638cc346fefb93e8d6601b437eceaf365c24bf02da096eacab78f54ae0/chuk_virtual_fs-0.2.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "b587f4db6589a8bea9236f9486212b6ba221b77a5961b968af495271f22f0ad3",
"md5": "cffcbb34b52e311f0fc64facb3a9ad19",
"sha256": "1534144306f24c93fc3613f2d97e61dda18e7e1671a0dba87da3be7e0ba12bdb"
},
"downloads": -1,
"filename": "chuk_virtual_fs-0.2.1.tar.gz",
"has_sig": false,
"md5_digest": "cffcbb34b52e311f0fc64facb3a9ad19",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 179825,
"upload_time": "2025-10-12T01:24:28",
"upload_time_iso_8601": "2025-10-12T01:24:28.598469Z",
"url": "https://files.pythonhosted.org/packages/b5/87/f4db6589a8bea9236f9486212b6ba221b77a5961b968af495271f22f0ad3/chuk_virtual_fs-0.2.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-12 01:24:28",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "chrishayuk",
"github_project": "chuk-virtual-fs",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "chuk-virtual-fs"
}
