| from __future__ import annotations |
|
|
| import os |
| from dataclasses import dataclass |
| from datetime import datetime, timezone |
|
|
| import boto3 |
|
|
|
|
| def utc_now_iso() -> str: |
| return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") |
|
|
|
|
| @dataclass(frozen=True) |
| class S3Config: |
| endpoint: str |
| public_base_url: str | None |
| access_key: str |
| secret_key: str |
| bucket: str |
| region: str = "us-east-1" |
| secure: bool = False |
|
|
|
|
| def load_s3_config_from_env() -> S3Config: |
| endpoint = os.environ.get("RUNNER_S3_ENDPOINT", "").strip() |
| public_base_url = os.environ.get("RUNNER_S3_PUBLIC_BASE_URL", "").strip() or None |
| access_key = os.environ.get("RUNNER_S3_ACCESS_KEY", "").strip() |
| secret_key = os.environ.get("RUNNER_S3_SECRET_KEY", "").strip() |
| bucket = os.environ.get("RUNNER_S3_BUCKET", "").strip() |
| region = os.environ.get("RUNNER_S3_REGION", "us-east-1").strip() |
| secure_str = os.environ.get("RUNNER_S3_SECURE", "false").strip().lower() |
| secure = secure_str in ("1", "true", "yes") |
| if not endpoint or not access_key or not secret_key or not bucket: |
| raise RuntimeError("Missing required S3 env vars: RUNNER_S3_ENDPOINT/ACCESS_KEY/SECRET_KEY/BUCKET") |
| return S3Config( |
| endpoint=endpoint, |
| public_base_url=public_base_url, |
| access_key=access_key, |
| secret_key=secret_key, |
| bucket=bucket, |
| region=region, |
| secure=secure, |
| ) |
|
|
|
|
| def get_s3_client(cfg: S3Config): |
| return boto3.client( |
| "s3", |
| region_name=cfg.region, |
| aws_access_key_id=cfg.access_key, |
| aws_secret_access_key=cfg.secret_key, |
| endpoint_url=cfg.endpoint, |
| use_ssl=cfg.secure, |
| verify=cfg.secure, |
| ) |
|
|
|
|
| def put_bytes(*, cfg: S3Config, key: str, content: bytes, content_type: str) -> str: |
| s3 = get_s3_client(cfg) |
| s3.put_object(Bucket=cfg.bucket, Key=key, Body=content, ContentType=content_type) |
| key = key.lstrip("/") |
| |
| |
| if cfg.public_base_url: |
| return f"{cfg.public_base_url.rstrip('/')}/{key}" |
| |
| return f"{cfg.endpoint.rstrip('/')}/{cfg.bucket}/{key}" |
|
|
