StashKat

stashkat provides a pluggable S3 storage backend for corekat. It implements the Fetcher protocol, allowing corekat's DownloadClient to handle s3:// URLs. It also provides a CLI for direct S3 operations.

Features

• S3 Fetcher for CoreKat: Enables corekat.client.filedl.DownloadClient to download from S3.
• Async-first S3 Client: A high-level, asynchronous client for interacting with S3-compatible object storage.
• Command-Line Interface: stashkat CLI for uploading, downloading, and copying files from S3.

Installation

uv pip install stashkat

Integration with CoreKat

When stashkat is installed in the same environment as corekat, its S3 fetcher is automatically registered upon initialization.

Programmatic Usage with DownloadClient

import asyncio
from corekat.config import config
from corekat.client.filedl import DownloadClient
from stashkat.init import init as stashkat_init

async def main():
    # Load config (which should include S3 settings)
    conf = config()
    # Initialize stashkat to register S3 fetcher
    stashkat_init(conf)

    downloader = DownloadClient()

    # Download from S3 using corekat's DownloadClient
    file_info = await downloader.download(
        source="s3://my-bucket/my-file.txt",
        dest_dir="/tmp"
    )
    print(f"Downloaded {file_info.path}")

if __name__ == "__main__":
    asyncio.run(main())

Direct S3Client Usage

import asyncio
from stashkat.config import config
from stashkat.s3 import S3Client
from stashkat.models import StoragePath

async def main():
    conf = config()
    client = S3Client(conf.s3)
    
    # Upload file
    storage_path = await client.upload_file("local/file.txt", dest="uploads/file.txt")
    print(f"Uploaded to: {storage_path.url}")
    
    # Download file
    remote = StoragePath.model_validate("s3://my-bucket/uploads/file.txt")
    await client.download_file(remote, dest="/tmp/downloaded.txt")
    
    # List files
    files = await client.list(prefix="uploads/")
    for file in files:
        print(f"Found: {file.url}")

if __name__ == "__main__":
    asyncio.run(main())

Errors & retries

Every backend translates its provider-specific failures into a small, portable exception hierarchy (in stashkat.exceptions), so you catch the same types regardless of whether S3, the filesystem, or a future backend is behind the StashStorage interface. All inherit corekat's AppException, so an error that reaches a servekat route maps to a sensible HTTP status automatically.

from stashkat import StashNotFoundError, StashTransientError, get_storage

storage = get_storage(conf, scheme="s3")
try:
    await storage.download_file(remote, dest="/tmp/out.txt")
except StashNotFoundError:
    ...  # 404 to the user — the object isn't there
except StashTransientError as e:
    assert e.retryable  # back off and retry

Branch on err.retryable to drive backoff/retry without importing botocore or knowing which backend produced the error. Credentials are never validated eagerly at client construction — an empty credential pair is a valid configuration (resolved from the environment / IAM role / instance-profile chain), and a broken pair surfaces lazily as a StashConfigError on first use.

retry_on_transient wraps that contract in a ready-made backoff loop — it retries only StashTransientError and re-raises everything else immediately:

from stashkat import retry_on_transient

stat = await retry_on_transient(lambda: storage.stat(remote), attempts=3)

Metadata, existence & presigned URLs

Beyond the read/write/copy basics, every backend exposes the operations apps otherwise drop down to raw boto3 for:

storage = get_storage(conf, scheme="s3")

# Cheap existence check + metadata (no download):
if await storage.exists(remote):
    info = await storage.stat(remote)      # -> ObjectStat
    print(info.size, info.content_type, info.etag, info.last_modified)

# Record a content type on upload (served back by stat / on GET):
await storage.upload_file(file_obj, dest="docs/report.pdf", content_type="application/pdf")

# Presigned URLs — let the browser talk to object storage directly (S3 only):
get_url = await storage.presigned_get_url(remote, expires_in=300)
put = await storage.presigned_put_url(dest, content_type="image/png", expires_in=300)

presigned_get_url / presigned_put_url are pure local SigV4 crypto (no network I/O). The filesystem backend has no presign concept, so it raises StashConfigError; exists / stat work on both backends (the filesystem backend guesses content_type from the extension and has no etag).

For synchronous, high-volume signing (e.g. a sync view-builder that signs many URLs per request), use S3Presigner — a plain sync class that reuses one botocore client instead of building an async client per call:

from stashkat import S3Presigner

presigner = S3Presigner(conf.s3)          # built from the same S3ConfigSchema
if presigner.available:                   # explicit credentials present?
    get_url = presigner.get_url("s3://bucket/key.jpg", expires_in=300)
    put_url = presigner.put_url("uploads/x.png", content_type="image/png")

CLI Usage

Configuration

The CLI requires S3 configuration to be present in a config.yaml or localconfig.yaml file, or via environment variables (prefixed with STASHKAT_).

Example config.yaml:

s3:
  bucket: "my-default-bucket"
  region: "us-east-1"
  access_key: "..." # Or use environment variables
  secret_key: "..." # Or use environment variables

Commands

Upload a file:

stashkat s3 upload /path/to/local/file.txt

Upload with a specific destination key:

stashkat s3 upload /path/to/local/file.txt --dest "archive/file.txt"

Download a file:

stashkat s3 download s3://my-bucket/archive/file.txt /path/to/save/location.txt

Copy a file:

stashkat s3 copy s3://my-bucket/source.txt s3://my-bucket/destination.txt