Created
August 20, 2025 15:26
-
-
Save lexod/554955ccd954728f0cdfae6cbe209057 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import json | |
| import os | |
| import uuid | |
| import base64 | |
| from datetime import datetime, timezone | |
| import boto3 | |
| from botocore.exceptions import BotoCoreError, ClientError | |
| def _env(name: str, default: str | None = None) -> str | None: | |
| val = os.getenv(name) | |
| return val if val not in (None, "") else default | |
| def _now_iso() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def _build_object_key(prefix: str | None, key: str | None) -> str: | |
| if key: | |
| return key | |
| # default key: <prefix>/dummy-<uuid>.txt | |
| prefix = (prefix or "reports").strip("/") | |
| return f"{prefix}/dummy-{uuid.uuid4().hex}.txt" | |
| def _to_bytes(content: str | bytes) -> bytes: | |
| if isinstance(content, bytes): | |
| return content | |
| return content.encode("utf-8") | |
| def lambda_handler(event, context): | |
| """ | |
| AWS Lambda handler to upload a dummy file (or provided content) to an S3 bucket. | |
| Configuration: | |
| - Environment variables (fallbacks used if not provided): | |
| - BUCKET_NAME (required if not provided in event) | |
| - OBJECT_PREFIX (optional, defaults to 'uploads') | |
| - OBJECT_KEY (optional; if not provided a key will be generated under prefix) | |
| - AWS_REGION (optional; boto3 can infer from environment/role) | |
| Event payload can override defaults: | |
| { | |
| "bucket": "my-bucket", # optional; overrides BUCKET_NAME | |
| "key": "path/file.txt", # optional; overrides OBJECT_KEY | |
| "prefix": "my/prefix", # optional; overrides OBJECT_PREFIX | |
| "content": "Hello", # optional; if absent a dummy content is used | |
| "content_base64": "...", # optional; base64-encoded content (takes precedence) | |
| "metadata": {"foo": "bar"} # optional; S3 object metadata | |
| } | |
| Returns standard Lambda proxy response with JSON body. | |
| """ | |
| s3 = boto3.client("s3") | |
| # Extract configuration from env and event | |
| env_bucket = _env("BUCKET_NAME") | |
| env_prefix = _env("OBJECT_PREFIX", "uploads") | |
| env_key = _env("OBJECT_KEY") | |
| bucket = (event or {}).get("bucket") or env_bucket | |
| prefix = (event or {}).get("prefix") or env_prefix | |
| key = (event or {}).get("key") or _build_object_key(prefix, env_key) | |
| if not bucket: | |
| return { | |
| "statusCode": 400, | |
| "body": json.dumps({ | |
| "ok": False, | |
| "error": "Bucket name not provided. Set BUCKET_NAME env var or pass event.bucket", | |
| }), | |
| } | |
| # Choose content | |
| body_bytes: bytes | |
| if event and isinstance(event, dict) and event.get("content_base64"): | |
| try: | |
| body_bytes = base64.b64decode(event["content_base64"], validate=True) | |
| except Exception as e: | |
| return { | |
| "statusCode": 400, | |
| "body": json.dumps({ | |
| "ok": False, | |
| "error": f"Invalid content_base64: {str(e)}", | |
| }), | |
| } | |
| elif event and isinstance(event, dict) and event.get("content") is not None: | |
| body_bytes = _to_bytes(event["content"]) | |
| else: | |
| # Default dummy content | |
| body_bytes = _to_bytes( | |
| "Dummy file uploaded by Lambda at " + _now_iso() + "\n" | |
| ) | |
| metadata = {} | |
| if event and isinstance(event, dict): | |
| metadata = event.get("metadata") or {} | |
| if not isinstance(metadata, dict): | |
| return { | |
| "statusCode": 400, | |
| "body": json.dumps({ | |
| "ok": False, | |
| "error": "metadata must be an object/map", | |
| }), | |
| } | |
| try: | |
| s3.put_object( | |
| Bucket=bucket, | |
| Key=key, | |
| Body=body_bytes, | |
| Metadata=metadata, | |
| ) | |
| return { | |
| "statusCode": 200, | |
| "body": json.dumps({ | |
| "ok": True, | |
| "message": "File uploaded successfully", | |
| "bucket": bucket, | |
| "key": key, | |
| "size_bytes": len(body_bytes), | |
| }), | |
| } | |
| except (ClientError, BotoCoreError) as e: | |
| return { | |
| "statusCode": 500, | |
| "body": json.dumps({ | |
| "ok": False, | |
| "error": str(e), | |
| "bucket": bucket, | |
| "key": key, | |
| }), | |
| } | |
| if __name__ == "__main__": | |
| # Simple local test runner for manual verification | |
| # Example: BUCKET_NAME=my-bucket python lambda_function.py | |
| test_event = { | |
| # "bucket": "override-bucket", # uncomment to override | |
| # "prefix": "test/prefix", # uncomment to override | |
| # "key": "test/prefix/local-run.txt", | |
| # "content": "Hello from local run!", | |
| # "metadata": {"source": "local"}, | |
| } | |
| resp = lambda_handler(test_event, context=None) | |
| print(json.dumps(resp, indent=2)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment