Module `trase.tools.r2.connection`

Access the Cloudflare R2 bucket (trase-r2) via Secrets Manager-vaulted tokens.

Cloudflare has no OIDC/IAM-role federation for R2, so access goes through long-lived R2 API tokens — but nobody holds one directly. They're stored in AWS Secrets Manager and fetched at call time, same pattern as trase.tools.ducklake.connection:

ro — shared read-only token, rotated ~yearly. Cached on disk for _CACHE_TTL_SECONDS under ${XDG_RUNTIME_DIR:-/tmp}/trase-r2/ (0700 dir, 0600 file, written atomically, ownership + permissions validated on read) so repeated calls in one session don't all hit Secrets Manager.
rw — shared "developer write" token, rotated ~yearly. Never cached — fetched fresh on every call, so a copy never outlives the process that needed it.

Each token is stored with two representations of the same grant (same bucket, same permission tier, rotated/revoked together) — but they are not interchangeable, because of a Cloudflare API limitation:

S3 Access Key ID / Secret Access Key — the S3-compatible API. Use this for everything that actually reads/writes objects: pandas/polars (via s3fs), DuckDB, boto3, aws s3 cp. This is the only one of the two that works with a bucket-scoped "Object Read & Write" token like ours.
Cloudflare API token (cfut_…) — Cloudflare's own REST API, what the wrangler CLI and the official cloudflare Python package speak. Object writes 403 on this API for bucket-scoped tokens — Cloudflare only accepts Admin-tier (account-wide) tokens there, confirmed the hard way in docs.yml's upload step (see https://github.com/cloudflare/workers-sdk/issues/9235). Since widening our tokens to Admin just to use wrangler would give up the whole point of scoping them to one bucket, treat :func:api_token()/:func:wrangler_env()/ :func:cloudflare_client() as usable only if you separately create an Admin-tier token for bucket-level administration (CORS, lifecycle, custom domains) — not for routine object get/put, which stays on the S3 API below.

Usage::

from trase.tools.r2 import connection as r2

# pandas / polars, via s3fs
df = pd.read_csv(r2.uri("dbt_elementary/elementary_report_parsed.html"),
                  storage_options=r2.storage_options("ro"))
df = pl.read_parquet(r2.uri("engagement/.../file.parquet"),
                      storage_options=r2.storage_options("ro"))

# duckdb
con = duckdb.connect()
r2.attach_duckdb_secret(con, role="ro")
con.sql(f"SELECT * FROM read_parquet('{r2.uri('some/file.parquet', scheme='r2')}')")

# boto3 — object get/put/list, presigned URLs, etc.
r2.client("ro").list_objects_v2(Bucket=r2.bucket(), Prefix="engagement/")

Functions

def api_token(role: Role = 'ro') ‑> str

The Cloudflare API token (cfut_…) for role.

Same secret, same bucket restriction as the S3 Access Key/Secret pair from :func:client()/:func:storage_options() — but not usable for object get/put via this route: Cloudflare's REST API (what wrangler and the official cloudflare Python package speak) 403s on object writes for anything less than an Admin-tier token, and ours are deliberately bucket-scoped "Object Read & Write" (https://github.com/cloudflare/workers-sdk/issues/9235). Only useful if you separately hold/create an Admin-tier token — see :func:cloudflare_client().

def attach_duckdb_secret(con, role: Role = 'ro', name: str = 'trase_r2') ‑> str

Create a TYPE r2 DuckDB secret named name on con. Returns name.

Lets callers then address objects as r2://<bucket>/<key> directly — DuckDB derives the endpoint from ACCOUNT_ID itself.

def bucket() ‑> str

def clear_ro_cache() ‑> bool

Delete the RO credential cache file (and dir if empty). Returns True if removed.

def client(role: Role = 'ro')

A boto3 S3 client configured against the R2 endpoint with the resolved role's credentials.

The way to actually read/write objects — pandas/polars (:func:storage_options()) and this function both go through the S3-compatible API, which is the only one our bucket-scoped tokens work against for object operations (see :func:api_token()).

def cloudflare_client(role: Role = 'ro')

The official cloudflare Python SDK client, authenticated for role.

Only useful for R2 bucket-level administration (CORS, lifecycle rules, custom domains, bucket create/delete) — Cloudflare's REST API has no object-level get/put/list at all (those stay on :func:client() / :func:storage_options(), S3-compatible), and even bucket admin needs an Admin-tier token: our tokens are bucket-scoped "Object Read & Write" and will 403 here (https://github.com/cloudflare/workers-sdk/issues/9235).

def endpoint_url() ‑> str

def ro_cache_path() ‑> pathlib.Path

Path of the on-disk RO credential cache file (may not exist).

def storage_options(role: Role = 'ro') ‑> dict

storage_options() for pandas/polars (via s3fs) reads/writes against R2.

def uri(key: str, scheme: "Literal['s3', 'r2']" = 's3') ‑> str

Build an s3:// (boto3/pandas/polars) or r2:// (native DuckDB) URI for key.

def wrangler_env(role: Role = 'ro') ‑> dict[str, str]

Env vars for shelling out to the wrangler CLI.

Merge into a subprocess call, e.g.::

subprocess.run(
    ["npx", "wrangler", "r2", "bucket", "cors", "set", bucket(), "--rules", path],
    env={**os.environ, **wrangler_env("rw")}, check=True,
)

Not for object get/put — wrangler r2 object … needs an Admin-tier token and 403s with ours (bucket-scoped "Object Read & Write"); use :func:storage_options()/:func:attach_duckdb_secret()/ :func:client() (S3-compatible API) for that instead. This is only useful for wrangler r2 bucket() … bucket-admin subcommands, and only if the resolved token happens to be Admin-tier.