AWS S3 (Object Storage)

@app.run
def run() -> None:
    bucket = Bucket("todo-bucket")

    api = Api("todo-api")
    api.route("GET", "/write", handler="functions/write.get", links=[bucket])
    api.route("GET", "/read", handler="functions/read.get", links=[bucket])

import boto3
from stlv_resources import Resources


def get(event, context):
    s3_client = boto3.client("s3")
    bucket_name = Resources.todo_bucket.bucket_name
    s3_client.put_object(Bucket=bucket_name, Key="hello.txt", Body="Hello, World!")
    return {"statusCode": 200, "body": "Hello, World!"}

@app.run
def run() -> None:
    bucket = Bucket("todo-bucket", access="public")

@app.run
def run() -> None:
    bucket = Bucket("uploads")

    # Notify on all object created events
    bucket.notify_function(
        "on-upload",
        events=["s3:ObjectCreated:*"],
        function="functions/process_upload.handler",
    )

@app.run
def run() -> None:
    bucket = Bucket("media")

    # Only trigger for images in the uploads folder
    bucket.notify_function(
        "process-images",
        events=["s3:ObjectCreated:*"],
        filter_prefix="uploads/",
        filter_suffix=".jpg",
        function="functions/process_image.handler",
        memory=512,
        timeout=60,
    )

bucket.notify_function(
    "on-upload",
    events=["s3:ObjectCreated:*"],
    function="functions/process_upload.handler",
    links=[table, results_queue],
)

@app.run
def run() -> None:
    bucket = Bucket("orders")
    processing_queue = Queue("order-processing")

    # Send notifications to queue
    bucket.notify_queue(
        "order-created",
        events=["s3:ObjectCreated:Put"],
        queue=processing_queue,
    )

    # Subscribe a function to process from the queue
    processing_queue.subscribe("processor", "functions/process_order.handler")

@app.run
def run() -> None:
    bucket = Bucket("orders")

    # Send to an external queue (you manage the queue policy)
    bucket.notify_queue(
        "order-created",
        events=["s3:ObjectCreated:Put"],
        queue="arn:aws:sqs:us-east-1:123456789012:my-external-queue",
    )

@app.run
def run() -> None:
    bucket = Bucket("uploads")
    notifications = Topic("upload-notifications")

    # Publish upload events to the topic
    bucket.notify_topic(
        "on-upload",
        events=["s3:ObjectCreated:*"],
        topic=notifications,
    )

    # Subscribe multiple handlers to the topic
    notifications.subscribe("processor", "functions/process_upload.handler")
    notifications.subscribe("logger", "functions/log_upload.handler")

@app.run
def run() -> None:
    bucket = Bucket("uploads")

    # Send to an external topic (you manage the topic policy)
    bucket.notify_topic(
        "on-upload",
        events=["s3:ObjectCreated:*"],
        topic="arn:aws:sns:us-east-1:123456789012:my-external-topic",
    )

@app.run
def run() -> None:
    bucket = Bucket("media")
    processing_queue = Queue("processing")
    alerts = Topic("alerts")

    # Process uploaded images
    bucket.notify_function(
        "process-images",
        events=["s3:ObjectCreated:*"],
        filter_suffix=".jpg",
        function="functions/process_image.handler",
        memory=512,
    )

    # Queue videos for async processing
    bucket.notify_queue(
        "queue-videos",
        events=["s3:ObjectCreated:*"],
        filter_suffix=".mp4",
        queue=processing_queue,
    )

    # Alert on all deletions in the archive folder
    bucket.notify_topic(
        "deletion-alert",
        events=["s3:ObjectRemoved:*"],
        filter_prefix="archive/",
        topic=alerts,
    )

@app.run
def run() -> None:
    config = mkdocs.config.load_config("mkdocs.yml")
    mkdocs.commands.build.build(config)

    _ = S3StaticWebsite(
        "s3-static-mkdocs",
        custom_domain="s3-2." + CUSTOM_DOMAIN_NAME,
        directory="site"
    )

@app.run
def run() -> None:
    config = mkdocs.config.load_config("mkdocs.yml")
    mkdocs.commands.build.build(config)

    website = S3StaticWebsite(
        "s3-static-mkdocs",
        custom_domain="s3-2." + CUSTOM_DOMAIN_NAME,
    )

    # Upload files to the bucket
    s3_bucket = website.bucket
    boto3_client = boto3.client("s3")
    boto3_client.put_object(
        Bucket=s3_bucket.bucket_name, Key="index.html", Body="<h1>Hello, World!</h1>"
    )

bucket = Bucket(
    "my-bucket",
    customize={
        "bucket": {
            "force_destroy": True,
            "tags": {"Environment": "dev"},
        }
    }
)

from stelvio.aws.s3 import S3StaticWebsite

website = S3StaticWebsite(
    "my-website",
    directory="./dist",
    custom_domain="www.example.com",
    customize={
        "bucket": {
            "bucket": {"tags": {"Type": "static-assets"}}
        },
        "cloudfront_distribution": {
            "distribution": {"price_class": "PriceClass_100"}
        }
    }
)