Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/usr/bin/env python3.6
'''
AWS Lambda handler to send a weekly email digest for GitHub repository releases.
For the full write up on how to use this script, see:
https://www.tonylykke.com/posts/subscribing-to-project-releases-with-lambda/
'''
from base64 import b64decode
from datetime import datetime, timedelta
from email.mime.text import MIMEText
import os
import smtplib
import boto3
import requests
def _get_decrypted(key):
'''
Helper to decrypt the stored credentials from AWS KMS.
Arguments:
key (str): name of environment variable to fetch encrypted value from.
Returns:
decrypted (str): decrypted os.environ[key] value.
'''
return boto3.client('kms').decrypt(
CiphertextBlob=b64decode(os.environ[key]))['Plaintext'].decode('UTF8')
def repos_with_releases(since=None):
'''
Generator that yields projects with ordered releases, for any projects
with releases more recently than `since`.
Arguments:
since (datetime.datetime): Only yield releases more recent than this.
If not provided, defaults to now - 7d
(rounded down to 00:00:00).
Yields:
release (dict): Project with CREATED_AT DESC ordered releases.
Returns:
N/A
'''
graphql_query = '''
query {{
viewer {{
starredRepositories(first:100{}) {{
pageInfo {{
endCursor
startCursor
}}
edges {{
node {{
id
nameWithOwner
releases(first:5, orderBy: {{field: CREATED_AT, direction: DESC}}) {{
edges {{
node {{
name
tag {{
name
}}
description
url
createdAt
}}
}}
}}
}}
}}
}}
}}
}}
'''
if not since:
since = datetime.now() - timedelta(days=7)
# Zero out everything after the day, this effectively rounds the
# datetime down to midnight.
for prop in ['hour', 'minute', 'second', 'microsecond']:
since = since - timedelta(**{
"{}s".format(prop): getattr(since, prop),
})
end_cursor_filter = ""
while True:
resp = requests.post("https://api.github.com/graphql", headers={
"Authorization": "token {}".format(_get_decrypted("GITHUB_TOKEN")),
"Content-Type": "application/json",
"Accept": "application/json",
}, json={
"query": graphql_query.format(end_cursor_filter),
"variables": {},
})
resp.raise_for_status()
data = resp.json()
repos = data["data"]["viewer"]["starredRepositories"]
end_cursor = repos["pageInfo"]["endCursor"]
if not end_cursor:
break
end_cursor_filter = ", after: \"{}\"".format(end_cursor)
for edge in repos["edges"]:
node = edge["node"]
repo_name = node["nameWithOwner"]
recent_releases = [release["node"]
for release in node["releases"]["edges"]
if datetime.strptime(
release["node"]["createdAt"],
"%Y-%m-%dT%H:%M:%SZ") > since]
yield {
"name": repo_name,
"releases": recent_releases,
}
def _build_email(releases, no_releases):
'''
Build a basic email with releases (or lack thereof) for the given projects.
Arguments:
releases (list(dict)): Projects with list of releases, sorted by
project name.
no_releases (list(dict)): Projects with no releaes, sorted by
project name.
Returns:
subject, body (tuple(str, str)): Email subject and body.
'''
title = "Project Releases for the Week Ending {}".format(
datetime.now().strftime("%Y-%m-%d"))
body = "<h1>{}</h1>".format(title)
for project in releases:
body += '''<h2>
<a href="https://github.com/{project}" title="{project}">
{project}
</a>
</h2>'''.format(project=project["name"])
for release in project["releases"]:
body += '''<p>
<ul>
<li>
<a href="https://github.com/{project}/releases/tag/{tag}"
title="{tag}">{tag}</a> {created_at}<br />
{description}
</li>
</ul>
</p>'''.format(
project=project["name"],
tag=release["tag"]["name"],
created_at=release["createdAt"],
description=release["description"].replace("\r\n", "<br />"))
if no_releases:
body += "<h1>No Releases</h1>"
for project in no_releases:
body += '''<ul>
<li><a href="https://github.com/{project}" title="{project}">{project}</a>
</ul>'''.format(project=project["name"])
return title, body.replace("\n", "")
def _send_email(subject, body):
'''
Send email using credentials from the environment.
Arguments:
subject (str): Subject of the email.
body (str): Body of the email.
Returns:
N/A
'''
from_addr = _get_decrypted("FROM_EMAIL")
to_addr = _get_decrypted("TO_EMAIL")
email_pass = _get_decrypted("EMAIL_PASSWORD")
conn = smtplib.SMTP(host="smtp.mailgun.org", port=587)
conn.starttls()
conn.login(from_addr, email_pass)
msg = MIMEText(body, "html")
msg["From"] = from_addr
msg["To"] = to_addr
msg["Subject"] = subject
conn.sendmail(msg["From"], [msg["To"]], msg.as_string())
conn.quit()
def digest_handler(event, context):
'''
Lambda entrypoint. Calls necessary functions to build and send the digest.
Arguments:
event (dict, list, str, int, float, None): Lambda event data.
context (LambdaContext): Lambda runtime information and other context.
Documentation on this type can be found here:
https://docs.aws.amazon.com/lambda/latest/dg/python-context-object.html
Returns:
N/A (return value is unused by Lambda when using an asynchronous
invocation method, such as periodic execution a la cron)
'''
main()
def main():
''' Main func. '''
no_releases = []
releases = []
for repo in repos_with_releases(
since=(datetime.now() - timedelta(days=7)).replace(
hour=0, minute=0, second=0, microsecond=0)):
if repo["releases"]:
releases.append(repo)
else:
no_releases.append(repo)
no_releases.sort(key=lambda x: x["name"])
releases.sort(key=lambda x: x["name"])
title, body = _build_email(releases, no_releases)
_send_email(title, body)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment