clean_gitlab_branches.py

'''
Clean up branches older than a specified time based on last commit.
Token rights required: api
'''
import sys
import os
import requests
from datetime import datetime

# Get personal access token from env. variables
# Personal access token needs full repo rights.
GITLAB_PAT_KEY = 'PERSONAL_ACCESS_TOKEN'
GITLAB_PAT = os.getenv(GITLAB_PAT_KEY)
if GITLAB_PAT is None:
    sys.exit(f"{GITLAB_PAT_KEY} environment variable doesn't exist.")

# Modified requests item
r = requests.Session()
r.headers.update({
    'PRIVATE-TOKEN': f'{GITLAB_PAT}'
})

# Configuration values
API_URL = "https://gitlab.com/api/v4/projects/"

def main():
    '''Delete all branches older than a specified day count'''

    # need to get the project ID from settings
    projects = {
        'PROJECT_NAME': YOUR_PROJECT_ID
    }

    for project in projects:
        project_id = projects[project]
        branches = get_branches(project_id)
        for branch in branches:
            print (branch)
            delete_branch(project_id, branch)

def get_branches(project_id):
    '''
    Gets all branches for a repo with a commit older than specified timeframe
    https://docs.gitlab.com/ee/api/branches.html
    '''
    result = []
    page = 1
    per_page = 100
    
    while True:
        try:
            response = r.get(
                f'{API_URL}{project_id}/repository/branches',
                params={'per_page': per_page, 'page': page})
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            sys.exit(f'Error when getting branches ({response.status_code}): {response.text}')

        response_objects = response.json()
        if not response_objects:  # No more items to fetch
            break

        branches = (ro['name'] for ro in response_objects if is_outdated(ro['commit']['created_at'][:10]))
        result.extend(branches)
        page += 1

    return result

def delete_branch(project_id, branch_name):
    try:
        response = r.delete(
            f'{API_URL}{project_id}/repository/branches/{branch_name}')
        response.raise_for_status()
    except requests.exceptions.HTTPError:
        sys.exit(f'Error when deleting branch {branch_name} ({response.status_code}): {response.text}')

def is_outdated(date_string):
    today = datetime.now()
    date_format = "%Y-%m-%d"
    provided_date = datetime.strptime(date_string, date_format)

    difference = today - provided_date

    return difference.days > 90

if __name__ == "__main__":
    main()