| Name | twat-fs JSON |
| Version |
1.8.1
JSON |
| download |
| home_page | None |
| Summary | File system utilities for twat with support for multiple upload providers |
| upload_time | 2025-02-15 05:50:23 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.10 |
| license | None |
| keywords |
dropbox
fal
file-upload
s3
twat
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# twat-fs
File system utilities for twat, focusing on robust and extensible file upload capabilities with multiple provider support.
## Rationale
`twat-fs` provides a unified interface for uploading files to various storage providers while addressing common challenges:
* **Provider Flexibility**: Seamlessly switch between storage providers without code changes
* **Fault Tolerance**: Graceful fallback between providers if primary provider fails
* **Progressive Enhancement**: Start simple with zero configuration (FAL), scale up to advanced providers (S3) as needed
* **Developer Experience**: Clear interfaces, comprehensive type hints, and runtime checks
* **Extensibility**: Well-defined provider protocol for adding new storage backends
## Quick Start
### Installation
Basic installation with FAL provider:
```bash
pip install twat-fs
```
Install with all providers and development tools:
```bash
pip install twat-fs[all,dev]
```
### Basic Usage
```python
from twat_fs import upload_file
# Simple upload (uses FAL by default)
url = upload_file("path/to/file.txt")
# Specify provider
url = upload_file("path/to/file.txt", provider="s3")
# Try specific providers in order
url = upload_file("path/to/file.txt", provider=["s3", "dropbox", "fal"])
```
### Command Line Interface
```bash
# Simple upload
python -m twat_fs upload_file path/to/file.txt
# Specify provider
python -m twat_fs upload_file path/to/file.txt --provider s3
# Check provider setup
python -m twat_fs setup provider s3
python -m twat_fs setup all
```
## Provider Configuration
### FAL (Default)
Zero configuration needed for basic usage. For production:
```bash
export FAL_KEY="your_key_here"
```
### Dropbox
```bash
export DROPBOX_ACCESS_TOKEN="your_token_here"
# Optional OAuth2 configuration
export DROPBOX_REFRESH_TOKEN="refresh_token"
export DROPBOX_APP_KEY="app_key"
export DROPBOX_APP_SECRET="app_secret"
```
### AWS S3
```bash
# Required
export AWS_S3_BUCKET="your_bucket"
export AWS_DEFAULT_REGION="us-east-1"
# Authentication (choose one)
export AWS_ACCESS_KEY_ID="key_id"
export AWS_SECRET_ACCESS_KEY="secret_key"
# Or use AWS CLI: aws configure
# Or use IAM roles in AWS infrastructure
# Optional
export AWS_ENDPOINT_URL="custom_endpoint" # For S3-compatible services
export AWS_S3_PATH_STYLE="true" # For path-style endpoints
export AWS_ROLE_ARN="role_to_assume"
```
## Architecture
### Provider System
The package uses a provider-based architecture with three key components:
1. **Provider Registry**: Central registry of available providers
* Maintains provider preference order
* Handles lazy loading of provider modules
* Provides runtime protocol checking
2. **Provider Protocol**: Formal interface that all providers must implement
* Credentials management
* Client initialization
* File upload functionality
* Help and setup information
3. **Provider Client**: The actual implementation that handles uploads
* Provider-specific upload logic
* Error handling and retries
* Progress tracking (where supported)
### Type System
Strong typing throughout with runtime checks:
* Type hints for all public APIs
* Runtime protocol verification
* Custom types for provider-specific data
## Implementing a New Provider
To add a new storage provider, create a module in `twat_fs/upload_providers/` that implements the Provider protocol:
```python
from pathlib import Path
from typing import Any, TypedDict
from twat_fs.upload_providers import ProviderClient, Provider
# Provider-specific help messages
PROVIDER_HELP = {
"setup": """Setup instructions for users...""",
"deps": """Additional dependencies needed..."""
}
def get_credentials() -> dict[str, Any] | None:
"""
Get provider credentials from environment.
Return None if not configured.
"""
# Implement credential checking
...
def get_provider() -> ProviderClient | None:
"""
Initialize and return the provider client.
Only import provider-specific dependencies here.
"""
creds = get_credentials()
if not creds:
return None
try:
# Initialize your provider client
client = YourProviderClient(creds)
return client
except Exception:
return None
def upload_file(local_path: str | Path, remote_path: str | Path | None = None) -> str:
"""
Upload a file and return its public URL.
This is a convenience wrapper around get_provider().
"""
client = get_provider()
if not client:
raise ValueError("Provider not configured")
return client.upload_file(local_path, remote_path)
# Your provider client implementation
class YourProviderClient:
def upload_file(
self,
local_path: str | Path,
remote_path: str | Path | None = None
) -> str:
"""Implement the actual upload logic."""
...
```
Then add your provider to `PROVIDERS_PREFERENCE` in `upload_providers/__init__.py`.
## Development
### Setup Environment
```bash
# Install development tools
pip install hatch
# Create and activate environment
hatch shell
# Install in development mode with all extras
pip install -e .[dev,all,test]
```
### Code Quality
```bash
# Format code
hatch run fmt
# Run type checks
hatch run typecheck
# Run linting
hatch run lint
# Run tests
hatch run test
hatch run test-cov # with coverage
# Quick development cycle
uv venv; source .venv/bin/activate; uv pip install --upgrade -e '.[dev,all,test]' ; fd -e py -x pyupgrade --py311-plus {}; hatch fmt --unsafe-fixes ; python -m pytest ;
```
### Publish
Make sure to have in your env:
```bash
export UV_PUBLISH_TOKEN="${PYPI_TOKEN}"
export HATCH_INDEX_AUTH="${UV_PUBLISH_TOKEN}"
export HATCH_INDEX_USER="__token__"
```
Build:
```bash
VER="v1.7.9" && echo "$VER" > VERSION.txt && git commit -am "$VER" && git tag "$VER"
```
Then either:
```
hatch build && hatch publish
```
Or:
```
uv build && uv publish
```
### Testing
The test suite includes:
* Unit tests for each provider
* Integration tests with real services
* Performance tests for large files
* Error condition tests
* Type checking tests
When adding a new provider:
1. Add unit tests in `tests/test_providers/`
2. Add integration tests in `tests/test_integration.py`
3. Add performance tests if relevant
4. Update provider discovery tests
## Error Handling & Troubleshooting
### Error Types
The package uses a structured error hierarchy:
```python
from twat_fs.errors import (
UploadError, # Base class for all upload errors
ProviderError, # Base class for provider-specific errors
ConfigurationError, # Missing or invalid configuration
AuthenticationError, # Authentication/credential issues
UploadFailedError, # Upload failed after max retries
ProviderNotFoundError, # Provider module not found/importable
)
```
### Common Issues
1. **Provider Not Available**
```python
try:
url = upload_file("file.txt", provider="s3")
except ProviderNotFoundError as e:
# Provider module not found
print(f"Provider not available: {e}")
print(e.setup_instructions) # Gets provider-specific setup guide
```
2. **Authentication Failures**
```python
try:
url = upload_file("file.txt")
except AuthenticationError as e:
# Credentials missing or invalid
print(f"Auth failed: {e}")
print(e.provider) # Which provider failed
print(e.credentials) # Which credentials are missing/invalid
```
3. **Upload Failures**
```python
try:
url = upload_file("file.txt")
except UploadFailedError as e:
print(f"Upload failed: {e}")
print(e.provider) # Which provider failed
print(e.attempts) # Number of attempts made
print(e.last_error) # Underlying error from provider
```
### Provider Status Checking
Use the setup commands to diagnose provider issues:
```bash
# Check specific provider
python -m twat_fs setup provider s3
# Check all providers
python -m twat_fs setup all
```
### Logging
The package uses `loguru` for structured logging:
```python
from loguru import logger
# Set log level
logger.level("DEBUG")
# Add file handler
logger.add("twat_fs.log", rotation="1 day")
# Log format includes provider info
logger.add(sys.stderr, format="{time} {level} [{extra[provider]}] {message}")
```
### Debugging Provider Issues
When implementing a new provider:
1. Enable debug logging:
```python
import logging
logging.getLogger("twat_fs").setLevel(logging.DEBUG)
```
2. Use the provider test helper:
```python
from twat_fs.testing import ProviderTestHelper
helper = ProviderTestHelper("your_provider")
helper.test_provider_implementation() # Checks protocol compliance
helper.test_provider_functionality() # Tests basic operations
```
3. Check provider initialization:
```python
from twat_fs.upload_providers import get_provider_module
provider = get_provider_module("your_provider")
print(provider.get_credentials()) # Check credential loading
print(provider.get_provider()) # Check client initialization
```
## License
MIT License
.
Raw data
{
"_id": null,
"home_page": null,
"name": "twat-fs",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "dropbox, fal, file-upload, s3, twat",
"author": null,
"author_email": "Adam Twardoch <adam+github@twardoch.com>",
"download_url": "https://files.pythonhosted.org/packages/f7/8b/dfbe5924ce0b987fbf3e905bb0e79b7c3390cd46d05377ba990c73c58a71/twat_fs-1.8.1.tar.gz",
"platform": null,
"description": "# twat-fs\n\nFile system utilities for twat, focusing on robust and extensible file upload capabilities with multiple provider support.\n\n## Rationale\n\n`twat-fs` provides a unified interface for uploading files to various storage providers while addressing common challenges:\n\n* **Provider Flexibility**: Seamlessly switch between storage providers without code changes\n* **Fault Tolerance**: Graceful fallback between providers if primary provider fails\n* **Progressive Enhancement**: Start simple with zero configuration (FAL), scale up to advanced providers (S3) as needed\n* **Developer Experience**: Clear interfaces, comprehensive type hints, and runtime checks\n* **Extensibility**: Well-defined provider protocol for adding new storage backends\n\n## Quick Start\n\n### Installation\n\nBasic installation with FAL provider:\n\n```bash\npip install twat-fs\n```\n\nInstall with all providers and development tools:\n\n```bash\npip install twat-fs[all,dev]\n```\n\n### Basic Usage\n\n```python\nfrom twat_fs import upload_file\n\n# Simple upload (uses FAL by default)\nurl = upload_file(\"path/to/file.txt\")\n\n# Specify provider\nurl = upload_file(\"path/to/file.txt\", provider=\"s3\")\n\n# Try specific providers in order\nurl = upload_file(\"path/to/file.txt\", provider=[\"s3\", \"dropbox\", \"fal\"])\n```\n\n### Command Line Interface\n\n```bash\n# Simple upload\npython -m twat_fs upload_file path/to/file.txt\n\n# Specify provider\npython -m twat_fs upload_file path/to/file.txt --provider s3\n\n# Check provider setup\npython -m twat_fs setup provider s3\npython -m twat_fs setup all\n```\n\n## Provider Configuration\n\n### FAL (Default)\n\nZero configuration needed for basic usage. For production:\n\n```bash\nexport FAL_KEY=\"your_key_here\"\n```\n\n### Dropbox\n\n```bash\nexport DROPBOX_ACCESS_TOKEN=\"your_token_here\"\n# Optional OAuth2 configuration\nexport DROPBOX_REFRESH_TOKEN=\"refresh_token\"\nexport DROPBOX_APP_KEY=\"app_key\"\nexport DROPBOX_APP_SECRET=\"app_secret\"\n```\n\n### AWS S3\n\n```bash\n# Required\nexport AWS_S3_BUCKET=\"your_bucket\"\nexport AWS_DEFAULT_REGION=\"us-east-1\"\n\n# Authentication (choose one)\nexport AWS_ACCESS_KEY_ID=\"key_id\"\nexport AWS_SECRET_ACCESS_KEY=\"secret_key\"\n# Or use AWS CLI: aws configure\n# Or use IAM roles in AWS infrastructure\n\n# Optional\nexport AWS_ENDPOINT_URL=\"custom_endpoint\" # For S3-compatible services\nexport AWS_S3_PATH_STYLE=\"true\" # For path-style endpoints\nexport AWS_ROLE_ARN=\"role_to_assume\"\n```\n\n## Architecture\n\n### Provider System\n\nThe package uses a provider-based architecture with three key components:\n\n1. **Provider Registry**: Central registry of available providers\n * Maintains provider preference order\n * Handles lazy loading of provider modules\n * Provides runtime protocol checking\n\n2. **Provider Protocol**: Formal interface that all providers must implement\n * Credentials management\n * Client initialization\n * File upload functionality\n * Help and setup information\n\n3. **Provider Client**: The actual implementation that handles uploads\n * Provider-specific upload logic\n * Error handling and retries\n * Progress tracking (where supported)\n\n### Type System\n\nStrong typing throughout with runtime checks:\n\n* Type hints for all public APIs\n* Runtime protocol verification\n* Custom types for provider-specific data\n\n## Implementing a New Provider\n\nTo add a new storage provider, create a module in `twat_fs/upload_providers/` that implements the Provider protocol:\n\n```python\nfrom pathlib import Path\nfrom typing import Any, TypedDict\nfrom twat_fs.upload_providers import ProviderClient, Provider\n\n# Provider-specific help messages\nPROVIDER_HELP = {\n \"setup\": \"\"\"Setup instructions for users...\"\"\",\n \"deps\": \"\"\"Additional dependencies needed...\"\"\"\n}\n\ndef get_credentials() -> dict[str, Any] | None:\n \"\"\"\n Get provider credentials from environment.\n Return None if not configured.\n \"\"\"\n # Implement credential checking\n ...\n\ndef get_provider() -> ProviderClient | None:\n \"\"\"\n Initialize and return the provider client.\n Only import provider-specific dependencies here.\n \"\"\"\n creds = get_credentials()\n if not creds:\n return None\n \n try:\n # Initialize your provider client\n client = YourProviderClient(creds)\n return client\n except Exception:\n return None\n\ndef upload_file(local_path: str | Path, remote_path: str | Path | None = None) -> str:\n \"\"\"\n Upload a file and return its public URL.\n This is a convenience wrapper around get_provider().\n \"\"\"\n client = get_provider()\n if not client:\n raise ValueError(\"Provider not configured\")\n return client.upload_file(local_path, remote_path)\n\n# Your provider client implementation\nclass YourProviderClient:\n def upload_file(\n self, \n local_path: str | Path, \n remote_path: str | Path | None = None\n ) -> str:\n \"\"\"Implement the actual upload logic.\"\"\"\n ...\n```\n\nThen add your provider to `PROVIDERS_PREFERENCE` in `upload_providers/__init__.py`.\n\n## Development\n\n### Setup Environment\n\n```bash\n# Install development tools\npip install hatch\n\n# Create and activate environment\nhatch shell\n\n# Install in development mode with all extras\npip install -e .[dev,all,test]\n```\n\n### Code Quality\n\n```bash\n# Format code\nhatch run fmt\n\n# Run type checks\nhatch run typecheck\n\n# Run linting\nhatch run lint\n\n# Run tests\nhatch run test\nhatch run test-cov # with coverage\n\n# Quick development cycle\nuv venv; source .venv/bin/activate; uv pip install --upgrade -e '.[dev,all,test]' ; fd -e py -x pyupgrade --py311-plus {}; hatch fmt --unsafe-fixes ; python -m pytest ;\n```\n\n### Publish\n\nMake sure to have in your env: \n\n```bash\nexport UV_PUBLISH_TOKEN=\"${PYPI_TOKEN}\"\nexport HATCH_INDEX_AUTH=\"${UV_PUBLISH_TOKEN}\"\nexport HATCH_INDEX_USER=\"__token__\"\n```\n\nBuild: \n\n```bash\nVER=\"v1.7.9\" && echo \"$VER\" > VERSION.txt && git commit -am \"$VER\" && git tag \"$VER\"\n```\n\nThen either:\n\n```\nhatch build && hatch publish\n```\n\nOr: \n\n```\nuv build && uv publish\n``` \n\n### Testing\n\nThe test suite includes:\n\n* Unit tests for each provider\n* Integration tests with real services\n* Performance tests for large files\n* Error condition tests\n* Type checking tests\n\nWhen adding a new provider:\n\n1. Add unit tests in `tests/test_providers/`\n2. Add integration tests in `tests/test_integration.py`\n3. Add performance tests if relevant\n4. Update provider discovery tests\n\n## Error Handling & Troubleshooting\n\n### Error Types\n\nThe package uses a structured error hierarchy:\n\n```python\nfrom twat_fs.errors import (\n UploadError, # Base class for all upload errors\n ProviderError, # Base class for provider-specific errors\n ConfigurationError, # Missing or invalid configuration\n AuthenticationError, # Authentication/credential issues\n UploadFailedError, # Upload failed after max retries\n ProviderNotFoundError, # Provider module not found/importable\n)\n```\n\n### Common Issues\n\n1. **Provider Not Available**\n\n```python\ntry:\n url = upload_file(\"file.txt\", provider=\"s3\")\nexcept ProviderNotFoundError as e:\n # Provider module not found\n print(f\"Provider not available: {e}\")\n print(e.setup_instructions) # Gets provider-specific setup guide\n```\n\n2. **Authentication Failures**\n\n```python\ntry:\n url = upload_file(\"file.txt\")\nexcept AuthenticationError as e:\n # Credentials missing or invalid\n print(f\"Auth failed: {e}\")\n print(e.provider) # Which provider failed\n print(e.credentials) # Which credentials are missing/invalid\n```\n\n3. **Upload Failures**\n\n```python\ntry:\n url = upload_file(\"file.txt\")\nexcept UploadFailedError as e:\n print(f\"Upload failed: {e}\")\n print(e.provider) # Which provider failed\n print(e.attempts) # Number of attempts made\n print(e.last_error) # Underlying error from provider\n```\n\n### Provider Status Checking\n\nUse the setup commands to diagnose provider issues:\n\n```bash\n# Check specific provider\npython -m twat_fs setup provider s3\n\n# Check all providers\npython -m twat_fs setup all\n```\n\n### Logging\n\nThe package uses `loguru` for structured logging:\n\n```python\nfrom loguru import logger\n\n# Set log level\nlogger.level(\"DEBUG\")\n\n# Add file handler\nlogger.add(\"twat_fs.log\", rotation=\"1 day\")\n\n# Log format includes provider info\nlogger.add(sys.stderr, format=\"{time} {level} [{extra[provider]}] {message}\")\n```\n\n### Debugging Provider Issues\n\nWhen implementing a new provider:\n\n1. Enable debug logging:\n\n```python\nimport logging\nlogging.getLogger(\"twat_fs\").setLevel(logging.DEBUG)\n```\n\n2. Use the provider test helper:\n\n```python\nfrom twat_fs.testing import ProviderTestHelper\n\nhelper = ProviderTestHelper(\"your_provider\")\nhelper.test_provider_implementation() # Checks protocol compliance\nhelper.test_provider_functionality() # Tests basic operations\n```\n\n3. Check provider initialization:\n\n```python\nfrom twat_fs.upload_providers import get_provider_module\n\nprovider = get_provider_module(\"your_provider\")\nprint(provider.get_credentials()) # Check credential loading\nprint(provider.get_provider()) # Check client initialization\n```\n\n## License\n\nMIT License\n \n.\n",
"bugtrack_url": null,
"license": null,
"summary": "File system utilities for twat with support for multiple upload providers",
"version": "1.8.1",
"project_urls": {
"Documentation": "https://github.com/twardoch/twat-fs#readme",
"Issues": "https://github.com/twardoch/twat-fs/issues",
"Source": "https://github.com/twardoch/twat-fs"
},
"split_keywords": [
"dropbox",
" fal",
" file-upload",
" s3",
" twat"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "91335ca34ef40c92b795cf55ce7b69430efb4a6fb7a1ed9e5d07f0c556b49822",
"md5": "2912b3587561f6f5a4cbcb560211e1bf",
"sha256": "9be6d46fa91e46e6b8af13e46e95b161052667463fb6fecf98fd5fd492cb6680"
},
"downloads": -1,
"filename": "twat_fs-1.8.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2912b3587561f6f5a4cbcb560211e1bf",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 22262,
"upload_time": "2025-02-15T05:50:16",
"upload_time_iso_8601": "2025-02-15T05:50:16.771781Z",
"url": "https://files.pythonhosted.org/packages/91/33/5ca34ef40c92b795cf55ce7b69430efb4a6fb7a1ed9e5d07f0c556b49822/twat_fs-1.8.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "f78bdfbe5924ce0b987fbf3e905bb0e79b7c3390cd46d05377ba990c73c58a71",
"md5": "6e5afb09fd85e37091dec96023fe131f",
"sha256": "e182da7c5de7f8c5ccd31866fd69e11408c8e5d802a0b0af7b4632a9ac525e30"
},
"downloads": -1,
"filename": "twat_fs-1.8.1.tar.gz",
"has_sig": false,
"md5_digest": "6e5afb09fd85e37091dec96023fe131f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 27098,
"upload_time": "2025-02-15T05:50:23",
"upload_time_iso_8601": "2025-02-15T05:50:23.464690Z",
"url": "https://files.pythonhosted.org/packages/f7/8b/dfbe5924ce0b987fbf3e905bb0e79b7c3390cd46d05377ba990c73c58a71/twat_fs-1.8.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-02-15 05:50:23",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "twardoch",
"github_project": "twat-fs#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "twat-fs"
}
