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_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_bucket always creates in the authenticated tenant, and there tenant_name only feeds the generated grant policy's principal ARN.

get_tenant() returns the authenticated user's own tenant (or "" if untenanted), read from the Owner of the ListBuckets response — so it needs neither an existing bucket nor any objects.

Buckets and users

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

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

create_bucket creates the bucket, sets its ACL to private, and — when grant_user is given — attaches a full-privileges (ListBucket/GetObject/PutObject/DeleteObject) policy whose principal is arn:aws:iam::acme:user/alice. Omit grant_user to just create a private bucket.

describe_bucket("reports", tenant_name="acme") returns a small summary — {"url", "name", "resolved", "owner", "object_count", "bytes_used", "empty"}. It's handy for a quick sanity check (seeing url next to empty makes a wrong-endpoint mix-up obvious), and object_count / bytes_used come straight from Ceph's x-rgw-* HEAD headers — an O(1) way to see how much data a bucket holds, no object listing required:

info = ceph.describe_bucket("coverage-maps")
print(info["object_count"], info["bytes_used"])   # 260  1342013096

# Pass unit= to get the size in KB/MB/GB/TB (1000-based) or KiB/MiB/GiB/...
# (1024-based); bytes_used is always kept raw alongside.
ceph.describe_bucket("coverage-maps", unit="GB")["size"]    # 1.342013096
ceph.describe_bucket("coverage-maps", unit="GiB")["size"]   # 1.2498470917...

(object_count / bytes_used are None on gateways that don't send those headers, e.g. AWS or older RGW.) The same unit= argument works on stat_file, and CephAdapter.convert_bytes(n, "GB") is exposed as a standalone helper.

Quotas (max size)

get_quota(bucket, tenant_name=None, unit="B") reports the limits Ceph applies, read from the same HEAD headers (-1/unset → None, meaning "no limit"):

ceph.get_quota("coverage-maps", unit="GiB")
# {'bucket_max_bytes': None,          # no per-bucket size cap
#  'bucket_max_objects': None,
#  'user_max_bytes': 1073741824000,   # shared across the user's buckets
#  'user_max_objects': None,
#  'max_buckets': 1000,
#  'bucket_max': None,                # bucket_max_bytes in `unit`
#  'user_max': 1000.0,                # user_max_bytes in `unit` (GiB)
#  'unit': 'GiB'}

There are two independent quotas: a per-bucket cap and a per-user cap that the user's buckets share. The effective ceiling for one bucket is the per-bucket quota when set, otherwise it's bounded by the shared user quota. The raw *_max_bytes values are always kept; bucket_max / user_max are those converted to unit.

Quotas (max size)

get_quota(bucket, tenant_name=None) reports the limits Ceph applies, read from the same HEAD headers (-1/unset → None, meaning "no limit"):

ceph.get_quota("coverage-maps")
# {'bucket_max_bytes': None,          # no per-bucket size cap
#  'bucket_max_objects': None,
#  'user_max_bytes': 1073741824000,   # 1000 GiB, shared across the user's buckets
#  'user_max_objects': None,
#  'max_buckets': 1000}

There are two independent quotas: a per-bucket cap and a per-user cap that the user's buckets share. The effective ceiling for one bucket is the per-bucket quota when set, otherwise it's bounded by the shared user quota. Byte values are raw — format with convert_bytes (e.g. ceph.convert_bytes(q["user_max_bytes"], "GiB") → 1000.0).

Objects

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("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_bucket, delete_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_bucket("reports", grant_user="alice", tenant_name="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.