def METHOD_NAME(context):
    if not bucket:
        raise Exception(
            "S3 bucket not specified at environment variable `DAGSTER_TOY_SENSOR_S3_BUCKET`."
        )
    new_s3_keys = get_s3_keys(bucket, since_key=context.last_run_key)
    if not new_s3_keys:
        yield SkipReason(f"No s3 updates found for bucket {bucket}.")
        return
    for s3_key in new_s3_keys:
        yield RunRequest(
            run_key=s3_key,
            run_config={
                "ops": {"read_s3_key": {"config": {"bucket": bucket, "s3_key": s3_key}}}
            },
        )