Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
image tweet with S3 in Lambda
import os
import sys
import uuid
import boto3
import tweepy
from urllib.parse import unquote_plus
# Twitter authentication variables
consumer_key = os.environ.get('TWITTER_CONSUMER_KEY', 'ap-northeast-1')
consumer_secret = os.environ.get('TWITTER_CONSUMER_SECRET', 'ap-northeast-1')
access_token = os.environ.get('TWITTER_ACCESS_TOKEN', 'ap-northeast-1')
access_secret = os.environ.get('TWITTER_ACCESS_SECRET', 'ap-northeast-1')
def tweet_with_image(path):
try:
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_secret)
api = tweepy.API(auth)
if os.path.exists(path):
api.update_with_media(filename=path)
else:
print('empty tweet')
except Exception as e:
print(e)
raise e
def lambda_handler(event, context):
for record in event['Records']:
bucket = record['s3']['bucket']['name'] # retrieve bucket name from record
key = unquote_plus(record['s3']['object']['key']
splitkey = key.split('/') # split folder name and key name
download_path = '/tmp/{}-{}'.format(uuid.uuid4(), splitkey[1])
s3_client.download_file(bucket, key, download_path)
# tweet a post with downloaded image
tweet_with_image(download_path)
# call lambda_hander
if __name__ == "__main__":
lambda_handler(json.loads(sys.argv[1]), {})
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment