Terraform is magic, i may be a little late to the game on this one and i’m sure it has it’s fair share of haters (i’ve seen some have a love hate relationship with it, maybe i’m still in my honeymoon period).

But from my point of view as a Data Scientist/ML Engineer playing around with various services in a multi-cloud environment (mainly GCP and AWS) its one of the things that keeps me sane in a sea of changing features and web ui’s that the cloud providers love to throw at us.

When it comes to cloud projects i’m working with, if its not under source control and managed by terraform then i don’t want to know about it.

So as I’ve been on my voyage of discovery with terraform i decided to put it to use in a little side project I’ve been wanting to do for a while.

(Note: All code is available in my reddit-links Github repo.)

The Set Up

The idea is to have a cloud function that runs each day, looks at top posts on r/MachineLearning, pulls out all links, and some metadata, and then saves those links somewhere useful.

My original goal (and something i might get to at some stage) was to pull all YouTube links shared on r/MachineLearning that have a decent score into an automatic playlist on YouTube or maybe automatically add them to my own watch later playlist. I have the data now so this might be an easy enough next project.

So given the goal above here are the various moving parts I’ve picked to use:

• Terraform: To manage all the GCP stuff used.
• GCP Cloud Function: This will be the code that pulls from r/MachineLearning, wrangles the text and html data from reddit, pulls out the links and then either inserts or updates Airtable accordingly.
• GCP PubSub Topic: This will be the trigger for the cloud function.
• GCP Cloud Scheduler: This will run a sort of cron job in GCP each day to push a message to the pubsub topic which will in turn trigger the cloud function.
• Airtable: This is where the data will be stored and published from. I had considered a GCP bucket but Airtable is much easier to share and a bit more user friendly for anyone who might want to use the links pulled from reddit.

Terraform

Once you work with and set up a Terraform project once then its pretty straight forward. I have used Serverless for some cloud functions before but i like the way terraform gives you everything GCP or AWS or Azure etc have to offer at your fingertips once you invest that little bit of learning up front.

Here is a list of the .tf files i’m using and what they all do. There are of course many ways to set things up in Terraform but this seemed like a straightforward enough way and works for me for smaller projects like this.

• backend.tf – this is optional and used to have a remote backend for the state of your project as opposed to somewhere on your laptop.
• conf.tf – this is a file i’m using to define any sensitive variables in terraform that i don’t want to go into source control (make sure you add to .gitignore) but need to make available to terraform to do it’s stuff. I have included an conf_example.tf file in the repo to show how this looks.
• variables.tf – used to define other variables used by terraform.
• provider.tf – conventional file to define the cloud providers you want to be able to use.
• gcp-cloud-functions.tf – this is where we will define all the things we need related to our Cloud Functions.
• gcp-cloud-scheduler.tf – used to define the Cloud Scheduler cron jobs we need.
• gcp-pubsub-topics.tf – used to define the PubSub topics that will trigger the cloud function.

GCP Function – redditlinks

The function used to pull from reddit is below. It might be a little verbose and could probably be refactored a little but hey – it works. Mainly using the PRAW library to pull from reddit, some usual data wrangling libraries, and then the airtable-python-wrapper library to insert/update records in Airtable.

	# -*- coding: utf-8 -*-
	import json
	import base64
	import logging
	import os
	from datetime import datetime
	from urllib.parse import urlparse
	import praw
	from bs4 import BeautifulSoup
	import pandas as pd
	from airtable import Airtable
	from urlextract import URLExtract
	def update_airtable(df_links, airtable_base_key, airtable_table_name):
	"""Update or insert new records into airtable.
	"""
	logging.info('… send to airtable …')
	# connect to airtable
	airtable = Airtable(airtable_base_key, airtable_table_name, api_key=os.environ['AIRTABLE_KEY'])
	inserted_count = 0
	updated_count = 0
	for i, row in df_links.iterrows():
	airtable_records = airtable.search('link', row['link'])
	if len(airtable_records) == 0:
	logging.info('… new record ({})…'.format(row['link']))
	record_fields = {
	'link': row['link'],
	'domain': row['domain'],
	'first_seen': row['created_utc'],
	'last_seen': row['created_utc'],
	'first_added': row['last_seen'],
	'last_added': row['last_seen'],
	'times_seen': 1,
	'score_avg': row['score'],
	'score_min': row['score'],
	'score_max': row['score'],
	'score_sum': row['score'],
	'title_list': row['title'],
	'last_title': row['title'],
	'permalink_list': row['permalink'],
	'last_permalink': row['permalink'],
	}
	airtable.insert(record_fields)
	logging.info('… inserted …')
	inserted_count += 1
	elif len(airtable_records) > 0:
	for record in airtable_records:
	logging.info('… update record ({})…'.format(row['link']))
	record_id = record['id']
	old = airtable.get(record_id)['fields']
	new_score_avg = (old['score_sum'] + row['score']) / (old['times_seen'] + 1)
	new_score_min = old['score_min'] if old['score_min'] <= row['score'] else row['score']
	new_score_max = old['score_max'] if old['score_max'] >= row['score'] else row['score']
	new_score_sum = old['score_sum'] + row['score']
	new_times_seen = old['times_seen'] + 1
	if row['title'] not in old['title_list']:
	new_title_list = '{}|{}'.format(old['title_list'], row['title'])
	else:
	new_title_list = old['title_list']
	if row['permalink'] not in old['permalink_list']:
	new_permalink_list = '{}|{}'.format(old['permalink_list'], row['permalink'])
	else:
	new_permalink_list = old['permalink_list']
	record_fields = {
	'link': old['link'],
	'domain': old.get('domain', 'N/A'),
	'first_seen': old['first_seen'],
	'last_seen': row['created_utc'],
	'first_added': old['first_added'],
	'last_added': row['last_seen'],
	'times_seen': new_times_seen,
	'score_avg': new_score_avg,
	'score_min': new_score_min,
	'score_max': new_score_max,
	'score_sum': new_score_sum,
	'title_list': new_title_list,
	'last_title': row['title'],
	'permalink_list': new_permalink_list,
	'last_permalink': row['permalink'],
	}
	airtable.update(record_id, record_fields)
	logging.info('… updated …')
	updated_count += 1
	logging.info('… inserted_count = {} …'.format(inserted_count))
	logging.info('… updated_count = {} …'.format(updated_count))
	return inserted_count, updated_count
	def redditlinks(event, context):
	"""
	Given the event recieved from pubsub pull from specified subreddit into specified airtable.
	Triggered from a message on a Cloud Pub/Sub topic.
	Args:
	event (dict): Event payload.
	context (google.cloud.functions.Context): Metadata for the event.
	"""
	logging.info('… begin …')
	# get data from event into a dict
	event_data_json = json.loads(base64.b64decode(event['data']).decode('utf-8'))
	logging.info('… event_data_json …')
	logging.info(event_data_json)
	subreddit = event_data_json['subreddit']
	submissions_time_filter = event_data_json['time_filter']
	comments_max = int(event_data_json['comments_max'])
	airtable_base_key = event_data_json['airtable_base']
	airtable_table_name = event_data_json['airtable_table']
	# create a url extractor
	extractor = URLExtract()
	# connect to reddit
	r = praw.Reddit(
	user_agent='reddit-links by /u/andrewm4894',
	client_id=os.getenv("REDDIT_CLIENT_ID"),
	client_secret=os.getenv("REDDIT_CLIENT_SECRET")
	)
	# create list to collect links into
	links = []
	# get reddit submissions
	submissions = r.subreddit(subreddit).top(submissions_time_filter)
	# process each submission
	for submission in submissions:
	submission_date = pd.to_datetime(submission.created_utc, unit='s').strftime('%Y-%m-%d')
	last_seen = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
	link = submission.url
	links.append([
	link, submission_date, submission.created_utc, last_seen, 'submission',
	submission.title, submission.permalink, submission.score, submission.upvote_ratio
	])
	# try also get urls from text of the submission
	submission_urls = extractor.find_urls(submission.selftext)
	if len(submission_urls) > 0:
	for submission_url in submission_urls:
	links.append([
	submission_url, submission_date, submission.created_utc, last_seen, 'submission', submission.title,
	submission.permalink, submission.score, submission.upvote_ratio
	])
	# pull links from comments related to each submission using bs4
	for comment in submission.comments.list()[0:comments_max]:
	if 'href' in comment.body_html:
	comment_date = pd.to_datetime(comment.created_utc, unit='s').strftime('%Y-%m-%d')
	soup = BeautifulSoup(comment.body_html, 'html.parser')
	# add a row for each link found by bs4
	for a in soup.find_all('a'):
	link = a.get('href')
	links.append([
	link, comment_date, comment.created_utc, last_seen, 'comment',
	submission.title, comment.permalink, comment.score, 0
	])
	# create a df for all links
	df_links = pd.DataFrame(
	links,
	columns=[
	'link', 'created_date', 'created_utc', 'last_seen', 'type',
	'title', 'permalink', 'score', 'upvote_ratio'
	]
	)
	df_links['permalink'] = 'https://www.reddit.com&#39; + df_links['permalink']
	df_links['created_utc'] = pd.to_datetime(df_links['created_utc'], unit='s').astype('str')
	# try get domain for each link
	domains = []
	for link in df_links.link:
	try:
	domain = urlparse(link).netloc
	except:
	domain = 'N/A'
	pass
	domains.append(domain)
	df_links['domain'] = domains
	df_links['domain'] = df_links['domain'].str.lower().str.replace('www.', '')
	logging.info(f' … df_links.shape = {df_links.shape} …')
	# update airtable with results
	inserted_count, updated_count = update_airtable(df_links, airtable_base_key, airtable_table_name)
	# result message
	result_message = f' … df_links.shape = {df_links.shape} …'
	result_message += f'… inserted_count = {inserted_count} …'
	result_message += f'… updated_count = {updated_count} …'
	logging.info(result_message)
	# build response
	response = {
	"statusCode": 200,
	"event": event,
	"context": context,
	"message": event_data_json,
	"result_message": result_message
	}
	logging.info(response)
	return response

view raw reddit-links.py hosted with ❤ by GitHub

End Results – Airtable

The end results of all this being a cloud function that runs once a day to update or insert records into this Airtable. Then from that i have made some views for links from specific domains.

(Note: If you wanted to do this but for a different subreddit then you should need to make minimal changes once you have your terraform variables all set up – primarily changing or adding another cloud scheduler job to here with the relevant params, as well as making sure all keys etc you need are available in the right places).

r/MachineLearning – YouTube Links

r/MachineLearning – Arxiv Links