feat(storage): add context-owned S3 object storage

.github/workflows/example-gate.yml

@@ -7,6 +7,11 @@ on:
 jobs:
   examples:
     runs-on: ubuntu-latest
+    env:
+      TEAQL_S3_ENDPOINT: http://127.0.0.1:19010
+      TEAQL_S3_BUCKET: teaql-python-example-ci
+      TEAQL_S3_ACCESS_KEY: teaql-python-example-ci
+      TEAQL_S3_SECRET_KEY: teaql-python-example-ci-fixture-only
     steps:
       - uses: actions/checkout@v4
       - name: Install example verification tools
@@ -15,5 +20,28 @@ jobs:
         with:
           python-version: '3.10'
           cache: pip
-      - run: pip install -e ".[test,telemetry]"
+      - run: pip install -e ".[test,telemetry,object-storage-s3]"
+      - name: Start dedicated S3 example fixture
+        run: |
+          docker run -d --name teaql-python-example-s3 \
+            -p 127.0.0.1:19010:9000 \
+            -e RUSTFS_ACCESS_KEY="$TEAQL_S3_ACCESS_KEY" \
+            -e RUSTFS_SECRET_KEY="$TEAQL_S3_SECRET_KEY" \
+            -e RUSTFS_CONSOLE_ENABLE=false \
+            rustfs/rustfs@sha256:41fe89380f4120a337790c02af192c3fe7bb55c3edc2e6e9357b487b47c6ab21 /data
+          for attempt in {1..30}; do
+            if curl -sS -o /dev/null --max-time 2 "$TEAQL_S3_ENDPOINT/"; then
+              exit 0
+            fi
+            if [[ "$(docker inspect --format '{{.State.Running}}' teaql-python-example-s3)" != true ]]; then
+              docker logs teaql-python-example-s3
+              exit 1
+            fi
+            sleep 1
+          done
+          docker logs teaql-python-example-s3
+          exit 1
       - run: ./scripts/verify-examples.sh
+      - name: Stop dedicated S3 example fixture
+        if: always()
+        run: docker rm -f teaql-python-example-s3 || true

README.md

@@ -399,6 +399,7 @@ Request JSON is not allowed to replace those trusted capabilities.
 - [Mutation Policy example](examples/mutation-policy)
 - [Facet example](examples/facet-trace/README.md)
 - [Opaque reference example](examples/opaque-entity-reference)
+- [Binary object storage and temporary S3 URLs](examples/object-storage/README.md)
 
 ## Safe troubleshooting
 

examples/object-storage/README.md

@@ -0,0 +1,53 @@
+# Context owned object storage
+
+Tracking: [Python #52](https://github.com/teaql/teaql-python/issues/52),
+[shared contract #122](https://github.com/teaql/teaql-conformance/issues/122).
+
+Install this checkout with `python -m pip install -e '.[object-storage-s3]'`.
+The optional boto3 provider owns signing, credentials and the shared HTTP pool.
+SDK types stay outside the core contract. The application installs the provider
+at trusted startup and closes it at shutdown.
+
+```python
+from teaql.core.object_storage import ObjectKey, PutOptions, WriteMode
+from teaql.provider.s3 import S3ObjectStorage
+from teaql.runtime import UserContext
+
+provider = S3ObjectStorage.connect("attachments")  # SDK default credentials chain
+context = UserContext().with_object_storage(provider)
+storage = context.object_storage("store payment receipt attachment")
+key = ObjectKey("receipts/opaque-reference.pdf")
+storage.put(key, source, PutOptions(content_length, WriteMode.CREATE_ONLY))
+link = storage.get_read_url(key)
+with storage.get(key) as response:
+    block = response.stream.read(65536)
+```
+
+This binary-stream API is synchronous, matching boto3 and Python BinaryIO.
+Async services can run the complete operation with `asyncio.to_thread`; never
+close the input while that worker is still running. Uploads stage in a private
+temporary file with bounded buffers, exact length checks and a 5 GiB limit.
+The facade closes upload inputs. Downloads stream and require explicit close;
+`get_bytes(key, maximum_bytes)` enforces an explicit memory ceiling.
+
+Create-only uses atomic conditional PUT; replacement is explicit. Only confirmed
+object absence becomes false. Missing buckets, permissions and network failures
+remain typed errors. Keys and default five-minute URLs have different lifetimes;
+the URL is a reusable bearer link, can expire early with credentials, and does
+not check existence or infer business permission. Diagnostic representations
+and operation events exclude keys, body, metadata, URLs and SDK error messages.
+Object storage is independent of SQL transactions.
+
+Set the four `TEAQL_S3_ENDPOINT`, `TEAQL_S3_BUCKET`, `TEAQL_S3_ACCESS_KEY`,
+`TEAQL_S3_SECRET_KEY` variables to a dedicated test service, then run twice:
+
+```bash
+bash examples/object-storage/verify.sh
+bash examples/object-storage/verify.sh
+```
+
+The example uses the repository source via PYTHONPATH, provisions its test bucket
+if absent and deletes only its unique keys. The production provider never
+creates buckets. Use `PYTHON=/path/to/venv/bin/python` to select an environment
+with boto3 and pytest. A local fixture pass qualifies that fixture, not every
+S3-compatible service. Range/list/copy/multipart and signed upload are extensions.

examples/object-storage/main.py

@@ -0,0 +1,101 @@
+"""Local-source S3 lifecycle fixture. Test bucket creation is administration."""
+from concurrent.futures import ThreadPoolExecutor
+from datetime import timedelta
+from io import BytesIO
+from uuid import uuid4
+import os
+import time
+import urllib.request
+import urllib.error
+import boto3
+from botocore.config import Config
+from teaql.core.object_storage import *
+from teaql.runtime import UserContext
+from teaql.provider.s3 import S3ObjectStorage
+
+
+def expect(code, work):
+    try:
+        work()
+    except StorageError as error:
+        assert error.code == code, (error.code, code)
+        return
+    raise AssertionError(f"expected {code}")
+
+
+def main():
+    bucket = os.environ["TEAQL_S3_BUCKET"]
+    client = boto3.client("s3", endpoint_url=os.environ["TEAQL_S3_ENDPOINT"], region_name="us-east-1",
+        aws_access_key_id=os.environ["TEAQL_S3_ACCESS_KEY"], aws_secret_access_key=os.environ["TEAQL_S3_SECRET_KEY"],
+        config=Config(signature_version="s3v4", retries={"max_attempts": 1}, s3={"addressing_style": "path"}))
+    from botocore.exceptions import ClientError
+    try:
+        client.head_bucket(Bucket=bucket)
+    except ClientError as error:
+        if error.response["ResponseMetadata"]["HTTPStatusCode"] != 404:
+            raise RuntimeError("test bucket setup failed") from None
+        client.create_bucket(Bucket=bucket)
+    provider = S3ObjectStorage(client, bucket)
+    events = []
+    context = UserContext().with_trace_id("storage-example").with_object_storage(provider).with_object_storage_observer(events.append)
+    storage = context.object_storage("verify binary object storage")
+    prefix = f"python-example/{uuid4().hex}/"
+    key = ObjectKey(prefix + "收据 +% binary")
+    empty, race = ObjectKey(prefix + "empty"), ObjectKey(prefix + "race")
+    data = bytes(i % 251 for i in range(256 * 1024))
+    try:
+        assert not storage.exists(key)
+        expect(StorageErrorCode.NOT_FOUND, lambda: storage.get(key))
+        options = PutOptions(len(data), WriteMode.CREATE_ONLY, "application/octet-stream", {"example": "binary"})
+        storage.put_bytes(key, data, options)
+        metadata = storage.head(key)
+        assert metadata.size == len(data) and metadata.custom_metadata["example"] == "binary" and metadata.etag
+        assert storage.get_bytes(key, len(data)) == data
+        expect(StorageErrorCode.CONFLICT, lambda: storage.put_bytes(key, b"", PutOptions(0, WriteMode.CREATE_ONLY)))
+        expect(StorageErrorCode.TOO_LARGE, lambda: storage.get_bytes(key, 1))
+        url = storage.get_read_url(key, ReadUrlOptions(disposition="attachment", filename="receipt.bin"))
+        with urllib.request.urlopen(url.url, timeout=5) as response:
+            assert response.read() == data
+            assert response.headers["Content-Disposition"] == 'attachment; filename="receipt.bin"'
+        expires = storage.get_read_url(key, ReadUrlOptions(timedelta(seconds=1)))
+        time.sleep(2.2)
+        try:
+            urllib.request.urlopen(expires.url, timeout=5)
+        except urllib.error.HTTPError as error:
+            assert error.code == 403
+        else:
+            raise AssertionError("expired URL must be denied")
+        storage.put_bytes(empty, b"", PutOptions(0, WriteMode.CREATE_ONLY))
+        assert storage.get_bytes(empty, 0) == b""
+        expect(StorageErrorCode.INVALID_INPUT, lambda: storage.put_bytes(key, b"ab", PutOptions(1, WriteMode.REPLACE)))
+        assert storage.get_bytes(key, len(data)) == data
+        storage.put_bytes(key, b"new", PutOptions(3, WriteMode.REPLACE))
+        assert storage.get_bytes(key, 3) == b"new"
+        other = UserContext().with_trace_id("storage-other").with_object_storage(provider).with_object_storage_observer(events.append)
+        def create(api):
+            try:
+                api.put_bytes(race, b"race", PutOptions(4, WriteMode.CREATE_ONLY))
+                return "success"
+            except StorageError as error:
+                assert error.code == StorageErrorCode.CONFLICT
+                return "conflict"
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            results = list(executor.map(create, [storage, other.object_storage("verify atomic create")]))
+        assert sorted(results) == ["conflict", "success"]
+        assert {e.trace_id for e in events} == {"storage-example", "storage-other"}
+        missing = UserContext().with_object_storage(S3ObjectStorage(client, "missing-" + uuid4().hex))
+        expect(StorageErrorCode.UNAVAILABLE, lambda: missing.object_storage("verify missing bucket").exists(key))
+        for item in (key, empty, race):
+            storage.delete(item)
+            storage.delete(item)
+            assert not storage.exists(item)
+        assert prefix not in repr(events) and "X-Amz-Signature" not in repr(events)
+        print("PASS Python S3: binary, empty, metadata, URL download/expiry, atomic create, replace, bounded reads, missing bucket, repeated delete")
+    finally:
+        for item in (key, empty, race):
+            storage.delete(item)
+        provider.close()
+
+
+if __name__ == "__main__":
+    main()

examples/object-storage/verify.sh

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+set -euo pipefail
+repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+: "${TEAQL_S3_ENDPOINT:?dedicated local test endpoint required}"
+: "${TEAQL_S3_BUCKET:?dedicated test bucket required}"
+: "${TEAQL_S3_ACCESS_KEY:?test access key required}"
+: "${TEAQL_S3_SECRET_KEY:?test secret key required}"
+cd "$repo"
+export PYTHONPATH="$repo/src"
+export PYTHONDONTWRITEBYTECODE=1
+"${PYTHON:-python3}" -m pytest tests/runtime/test_object_storage.py -q
+"${PYTHON:-python3}" examples/object-storage/main.py

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+object-storage-s3 = ["boto3==1.43.113"]
 telemetry = [
     "opentelemetry-api>=1.44.0,<2",
     "opentelemetry-sdk>=1.44.0,<2",

scripts/verify-examples.sh

@@ -2,7 +2,7 @@
 set -euo pipefail
 
 repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-expected=(business-clock business-id conformance facet-trace mutation-policy opaque-entity-reference order-management query-policy school-management task_board trace-chain)
+expected=(business-clock business-id conformance facet-trace mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain)
 mapfile -t actual < <(find "$repo/examples" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | sort)
 if [[ "${actual[*]}" != "${expected[*]}" ]]; then
   echo "example inventory changed; update scripts/verify-examples.sh: ${actual[*]}" >&2
@@ -20,6 +20,7 @@ if rg -l 'include\s*=.*teaql\*' "$repo/examples" --glob pyproject.toml >/dev/nul
   exit 1
 fi
 
+bash "$repo/examples/object-storage/verify.sh"
 PYTHONPATH="$repo/examples/conformance:$repo/src" python -m app.main
 PYTHONPATH="$repo/src" python -m unittest discover -s "$repo/examples/conformance" -p 'test_sql_log_intent.py' -v
 PYTHONPATH="$repo/examples/school-management:$repo/src" python -m app.main