ObjectDB S3 SDK examples.

from __future__ import annotations

import boto3
from botocore import UNSIGNED
from botocore.config import Config
from botocore.exceptions import ClientError

endpoint = "http://127.0.0.1:18080"
bucket = "docs-058-sdk"
key = "raw/photo.txt"
body = b"objectdb-sdk-body"

s3 = boto3.client(
    "s3",
    endpoint_url=endpoint,
    region_name="us-east-1",
    config=Config(signature_version=UNSIGNED, s3={"addressing_style": "path"}),
)

s3.create_bucket(Bucket=bucket)
s3.put_object(
    Bucket=bucket,
    Key=key,
    Body=body,
    ContentType="text/plain",
    Metadata={"source": "docs-058", "workflow": "sdk-example"},
    Tagging="kind=raw&stage=example",
)

payload = s3.get_object(Bucket=bucket, Key=key)["Body"].read()
assert payload == body

listed = s3.list_objects_v2(Bucket=bucket, Prefix="raw/")
assert key in [entry["Key"] for entry in listed.get("Contents", [])]

tags = s3.get_object_tagging(Bucket=bucket, Key=key)["TagSet"]
assert {"Key": "stage", "Value": "example"} in tags

try:
    s3.get_bucket_versioning(Bucket=bucket)
except ClientError as error:
    assert error.response["Error"]["Code"] == "NotImplemented"
else:
    raise AssertionError("ObjectDB s3-core-v1 should not report versioning support")

s3.delete_object(Bucket=bucket, Key=key)
s3.delete_bucket(Bucket=bucket)