Skip to content

Instantly share code, notes, and snippets.

@sidwarkd sidwarkd/
Last active Jun 4, 2018

What would you like to do?
Lambda Functions for the NHL Score Twitter Bot

Need to locally install all dependencies with pip install [package] -t ./ so they can be packaged before uploading to AWS. The packages need to be created in their own folders and the filenames changed to


  • The nhl package is required for get_nhl_game_brief
import json
import pytz
from nhl import Api
from datetime import date, datetime, timedelta
def get_todays_games(api):
today =
today = today - timedelta(hours=12)
return api.get_games_for_date(today)
def get_upcoming_games_and_date(api):
date =
date = date - timedelta(hours=12)
games = api.get_games_for_date(date)
while (len(games) == 0):
date = date + timedelta(days=1)
games = api.get_games_for_date(date)
return date, games
# Construct a tweet based on what's going on right now around the NHL
def lambda_handler(event, context):
api = Api()
games = get_todays_games(api)
if len(games) == 0:
# There are no games today so find the first day there are games
# in the future and show those
date, games = get_upcoming_games_and_date(api)
pre_msg = 'Next games on {}'.format(
lines = [game.short_summary for game in games]
lines.insert(0, pre_msg)
return lines
return [game.short_summary for game in games]
if __name__ == '__main__':
print(lambda_handler(None, None))
import decimal
import json
import re
import os
import urllib
import boto3
import twitter
from boto3.dynamodb.conditions import Key, Attr
CONSUMER_KEY = os.environ['consumer_key']
CONSUMER_SECRET = os.environ['consumer_secret']
ACCESS_KEY = os.environ['access_token_key']
ACCESS_SECRET = os.environ['access_token_secret']
SEARCH_REGEX = os.environ['regex']
class AWSLambdaException(Exception): pass
def get_last_mention_id(table):
response = table.query(KeyConditionExpression=Key('name').eq('last_mention_id'))
items = response['Items']
last_mention_id = None
if len(items) == 1:
last_mention_id = items[0]['val']
return last_mention_id
def update_last_mention_id(table, last_id, new_id):
if last_id is None:
# Create a new entry
# Update the existing entry
UpdateExpression="set val = :v",
':v': new_id
def get_list_requesting_scores(api, table):
reply_list = []
# Get the last mention ID for query. If one doesn't exist get all recent mentions
last_id = get_last_mention_id(table)
mentions = api.GetMentions(20, last_id)
if len(mentions) > 0:
new_id = mentions[0].id_str
for mention in mentions:
if, mention.text):
update_last_mention_id(table, last_id, new_id)
return reply_list
def lambda_handler(event, context):
api = twitter.Api(consumer_key=CONSUMER_KEY,
dynamodb = boto3.resource('dynamodb', region_name='us-west-2')
state_table = dynamodb.Table('app_state')
reply_list = get_list_requesting_scores(api, state_table)
if len(reply_list) > 0:
# Get the tweet text from get_nhl_game_brief lambda
client = boto3.client('lambda', region_name='us-west-2')
resp = client.invoke(FunctionName='get_nhl_game_brief')
if resp['StatusCode'] != 200:
raise AWSLambdaException("Something went wrong getting the tweet text")
lines_json = resp['Payload'].read()
lines = json.loads(lines_json)
tweet_body = '\n'.join(lines)
for user,tweet_id in reply_list:
print('Replying to {} with tweet ID {}'.format(user, tweet_id))
api.PostUpdate(tweet_body, None, None, None, tweet_id, True)
return reply_list
return None
if __name__ == '__main__':
print(lambda_handler(None, None))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.