Skip to content

Instantly share code, notes, and snippets.

@DylanBaker
Last active April 2, 2019 19:29
Show Gist options
  • Save DylanBaker/08a7841482b45c684004775307f9d61e to your computer and use it in GitHub Desktop.
Save DylanBaker/08a7841482b45c684004775307f9d61e to your computer and use it in GitHub Desktop.
Script to clean up old git branches in Looker. Will delete all non-personal branches where the last commit is before the WEEKS_AGO_CUTOFF set.
#!/usr/bin/env python
# coding: utf-8
import requests
from datetime import datetime, timedelta
## Configs - enter CLIENT_ID and CLIENT_SECRET
BASE_URL = ''
CLIENT_ID = ''
CLIENT_SECRET = ''
WEEKS_AGO_CUTOFF = 10
PROJECT_NAME = ''
class LookerConnection:
def __init__(self, client_id, client_secret, url):
if url[-1] == '/':
self.url = url[:-1] + ':19999/api/3.0/'
else:
self.url = url + ':19999/api/3.0/'
self.headers = self.connect(client_id, client_secret)
def connect(self, client_id, client_secret):
"""Gets Access Token from Looker, setting token on LookerConnection"""
login = requests.post(
url=compose_url(self.url, 'login'),
data={'client_id': client_id, 'client_secret': client_secret})
try:
access_token = login.json()['access_token']
headers = {'Authorization': 'token {}'.format(access_token)}
except KeyError:
headers = None
return headers
def _get(self, endpoint, endpointid=None, subendpoint=None, subendpointid=None):
r = requests.get(
url=compose_url(self.url, endpoint, endpointid=endpointid, subendpoint=subendpoint, subendpointid=subendpointid),
headers=self.headers)
return r
def _delete(self, endpoint, endpointid=None, subendpoint=None, subendpointid=None):
r = requests.delete(
url=compose_url(self.url, endpoint, endpointid=endpointid, subendpoint=subendpoint, subendpointid=subendpointid),
headers=self.headers)
return r
def _patch(self, endpoint, endpointid=None, subendpoint=None, subendpointid=None, payload=None):
r = requests.patch(
url=compose_url(self.url, endpoint, endpointid=endpointid, subendpoint=subendpoint, subendpointid=subendpointid),
headers=self.headers,
json=payload)
return r
def compose_url(urlbase, endpoint, endpointid=None, subendpoint=None, subendpointid=None):
"""
Composes a URL from a base, endpoint and possible an endpoint ID and a sub-endpoint.
Requires at least URL base and endpoint
"""
if not urlbase or not endpoint:
raise Exception('compose_url requires urlbase and endpoint.')
url = "{}{}".format(urlbase,endpoint)
if endpointid:
url += '/' + str(endpointid)
if subendpoint:
url += '/' + subendpoint
if subendpointid:
url += '/' + str(subendpointid)
return url
conn = LookerConnection(CLIENT_ID, CLIENT_SECRET, BASE_URL)
update_session = conn._patch('session',payload={'workspace_id': 'dev'})
all_projects = conn._get('projects')
for project in all_projects.json():
if project['name'] == PROJECT_NAME:
id = project['id']
branches = conn._get('projects', endpointid=id, subendpoint='git_branches')
date_cutoff = datetime.today() - timedelta(weeks=WEEKS_AGO_CUTOFF)
branch_count = 0
for branch in branches.json():
commit_date = datetime.utcfromtimestamp(branch['commit_at'])
branch_name = branch['name']
if commit_date < date_cutoff and not branch_name.startswith('dev') and branch_name != 'master':
print("Deleting branch '{}', last commit on {}".format(branch_name, commit_date.strftime('%Y-%m-%d')))
delete_branch = conn._delete('projects',id,'git_branch',branch_name)
branch_count += 1
print(branch_count)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment