ceph-adapter 0.2.0

A python adapter for interacting with Ceph RADOS Gateway (S3) object storage

ceph-adapter

A small, friendly Python wrapper around boto3 for administering and using a Ceph RADOS Gateway (RGW) over its S3-compatible API.

Ceph RGW implements S3, but it has quirks that plain boto3 makes awkward — most notably tenants (buckets addressed as tenant:bucket, whose : boto3 rejects by default) and the fact that policies attach to buckets, not users. ceph-adapter smooths those over with two classes:

• CephAdapter — bucket lifecycle, object upload/download (file, path, or stream), presigned URLs, an fsspec handle, and policy management.
• CephPolicyBuilder — a fluent builder that produces valid S3 bucket policy documents (per-user grants, public read-only, custom statements).

The docstrings on CephAdapter and CephPolicyBuilder are the authoritative per-method reference (parameters, tenant behaviour, and which exceptions each method raises). This README is the tour; help(CephAdapter) is the manual.

Installation

pip install ceph-adapter

Or with Poetry:

poetry add ceph-adapter

Requires Python 3.10+.

Quick start

from ceph_adapter import CephAdapter

ceph = CephAdapter(
    url="https://rgw.example.com",
    access_key="ACCESS_KEY",
    secret_key="SECRET_KEY",   # str or pydantic SecretStr; stored as SecretStr
)

# List buckets visible to these credentials
print(ceph.list_buckets())

# Upload / download a single object
ceph.upload_file(bucket_name="reports", file_path="./q3.pdf", path_in_bucket="2025/")
ceph.download_file(bucket_name="reports", file_name="2025/q3.pdf", path="./q3.pdf")

# Presigned URL for temporary sharing (default: GET, 1 hour)
url = ceph.get_signed_url("reports", "2025/q3.pdf", expiration_time=600)

Credentials are always wrapped in a pydantic.SecretStr, so they won't leak into logs or reprs. Use ceph.get_credentials() when you need the raw values.

Tenants

Ceph namespaces buckets per tenant and addresses a tenanted bucket as "tenant:bucket". Every data and policy method therefore accepts an optional tenant_name:

# Address a bucket that lives under the "acme" tenant
ceph.list_bucket_files(bucket_name="reports", tenant_name="acme")
ceph.upload_file("reports", "./q3.pdf", tenant_name="acme")

• When tenant_name is given, the adapter joins it to the bucket name (acme:reports) before the request.
• When it's omitted, the bare bucket name is used and the request resolves against the tenant of the authenticated credentials.

Creating a bucket is the one exception: the S3 API cannot create a bucket inside an arbitrary tenant, so create_user_bucket always creates in the authenticated tenant, and there tenant_name only feeds the generated grant policy's principal ARN.

get_tenant() is a best-effort helper that infers the current tenant name from existing buckets; treat it as experimental.

Buckets and users

# Create a private bucket and grant a user full access in one call
ceph.create_user_bucket(bucket_name="reports", user_name="alice", tenant_name="acme")

ceph.bucket_exists("reports", tenant_name="acme")   # -> bool (HEAD-based)
ceph.delete_user_bucket("reports", tenant_name="acme")  # bucket must be empty

create_user_bucket creates the bucket, sets its ACL to private, and attaches a full-privileges (ListBucket/GetObject/PutObject/DeleteObject) policy whose principal is arn:aws:iam::acme:user/alice.

Objects

Method	Purpose
upload_file(bucket, file_path, path_in_bucket, tenant_name)	Upload one local file
upload_path(bucket, path, path_in_bucket, recursive)	Upload every file in a directory
upload_stream(bucket, stream, file_name, tenant_name)	Upload from a file-like/bytes stream
download_file(bucket, file_name, path, tenant_name)	Download to a local path
download_stream(bucket, file_name, tenant_name)	Return a streaming body
delete_file(bucket, file_name, tenant_name)	Delete an object
list_bucket_files(bucket, tenant_name, verbose)	List keys (or full metadata when verbose=True)
get_signed_url(bucket, object_key, method, expiration_time, tenant_name)	Presigned URL
get_fsspec(bucket)	An fsspec S3 filesystem bound to these credentials

Bucket policies

Ceph attaches policies to buckets, not users. CephPolicyBuilder builds the policy document; CephAdapter.grant_policy_to_bucket attaches it.

import json
from ceph_adapter import CephPolicyBuilder

policy = (
    CephPolicyBuilder("reports-access")
    .add_users_read_privileges("reports", tenant_users=[("acme", "reader")])
    .add_users_write_privileges("reports", tenant_users=[("acme", "writer")])
    .build()
)

ceph.grant_policy_to_bucket(
    bucket_name="reports",
    policy_string=json.dumps(policy),
    tenant_name="acme",
)

Builder highlights:

• add_users_read_privileges / _write_ / _delete_ / _full_privileges — convenience grants for one or more (tenant, user) principals. Pass tenant=None in the tuple for a user in the default (untenanted) namespace.
• add_entry(bucket, object_names, privileges, conditions, tenant_users, allow) — full control: choose actions, scope to specific object keys, add IAM conditions, or make a Deny statement. tenant_users="*" makes the statement public (principal *).
• Calls chain and accumulate statements; build() returns the policy dict, and write_policy_file(name, path) serializes it to disk.

Convenience shortcut for public content:

# Keeps the bucket ACL private, publishes objects via a public GetObject policy
ceph.set_bucket_public_readonly_access("assets", object_names=["*"], tenant_name="acme")

# Inspect what is currently attached
print(ceph.describe_bucket_policy("assets", tenant_name="acme"))

Error handling

• BucketError — raised by the bucket-lifecycle helpers (create_user_bucket, delete_user_bucket, bucket_exists) so you can catch bucket problems (already exists, missing, not empty, forbidden) without importing botocore. The underlying error is preserved as __cause__.
• botocore.exceptions.ClientError — surfaced directly by the object and policy methods (e.g. NoSuchBucketPolicy from describe_bucket_policy when a bucket has no policy). Each method's docstring notes its specific failure modes.

from ceph_adapter import BucketError

try:
    ceph.create_user_bucket("reports", "alice", "acme")
except BucketError as err:
    print("bucket op failed:", err)

Development

poetry install          # install the package + dev tooling
poetry run pytest       # run the test suite
poetry run ruff check . # lint
poetry run black .      # format

Testing scope

The unit tests cover the parts that are genuinely our logic: the CephPolicyBuilder policy construction and the CephAdapter boto3 customisations (tenant-name handler removal, path-style addressing, credential wrapping) — none of which touch the network. The Ceph-specific S3 semantics (tenant addressing, policy enforcement) can't be faithfully reproduced by a generic S3 mock like moto, so they are intentionally left to integration testing against a real RGW rather than tests that would only re-assert boto3's behaviour.

CI / release

• CI (.github/workflows/ci.yml) runs ruff, black --check, and pytest on every push and pull request across Python 3.10–3.12.
• Release (.github/workflows/release.yml) publishes to PyPI via trusted publishing when a X.Y.Z tag is pushed, after re-running lint/tests and verifying the tag matches the version in pyproject.toml.

To cut a release: bump version in pyproject.toml, commit, then git tag 0.2.0 && git push --tags.