lambda-tweet/lambda_function.py at master · influentialcodellc/lambda-tweet · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
importbase64
importlogging
importos
importtweet
importboto3
importtwitter


defdecrypt_env_var(env_var: str) ->str:
"""
 example return value from `decrypt`
 {
 'KeyId': 'string',
 'Plaintext': b'bytes'
 }
 """
returnboto3.client('kms').decrypt(CiphertextBlob=base64.b64decode(env_var))['Plaintext'].decode('utf-8')


URL_BASE=os.environ['URL_BASE']
DECRYPTED_TOKEN=decrypt_env_var(os.environ['TOKEN'])
DECRYPTED_TOKEN_SECRET=decrypt_env_var(os.environ['TOKEN_SECRET'])
DECRYPTED_CONSUMER_KEY=decrypt_env_var(os.environ['CONSUMER_KEY'])
DECRYPTED_CONSUMER_SECRET=decrypt_env_var(os.environ['CONSUMER_SECRET'])


deflambda_handler(event, context):
logger=logging.getLogger('tweet')
logger.setLevel(logging.INFO)
keys= [record['s3']['object']['key'] forrecordinevent.get('Records', [])]
ifnotkeysorlen(keys) >1:
logger.error('Only one new post at a time is expected.')
return

t=twitter.Twitter(auth=twitter.OAuth(token=DECRYPTED_TOKEN,
token_secret=DECRYPTED_TOKEN_SECRET,
consumer_key=DECRYPTED_CONSUMER_KEY,
consumer_secret=DECRYPTED_CONSUMER_SECRET))
tweeter=tweet.Tweeter(t)
tweeter.tweet(f"{URL_BASE}/{keys[0]}")