github.py

import os
import requests
from dotenv import load_dotenv
from models import RepoStatus
load_dotenv()


class GithubConnection:
    def __init__(self) -> None:
        self.GH_TOKEN = os.getenv("GH_TOKEN")
        self.GH_USER = os.getenv("GH_USER")
        self.REPO_URL = "https://api.github.com/user/repos"
        self.WORKFLOW_URL_PREFIX = f"https://api.github.com/repos/{self.GH_USER}"
        self.headers = {"Authorization": f"Bearer {self.GH_TOKEN}"}

    def get_list_repos(self) -> list[dict]:
        response = requests.get(self.REPO_URL, headers=self.headers)

        
        if response.status_code == 200:
            data = response.json()
            return [item for item in data if item.get("owner").get("login").lower()==self.GH_USER]
        else:
            raise Exception(f"Failed to get list of repos: {response.status_code}")

    def get_repo_info(self, repo_name: str) -> dict:
        url = f"{self.WORKFLOW_URL_PREFIX}/{repo_name}/actions/runs"
        response = requests.get(url, headers=self.headers)
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"Failed to get repo info: {response.status_code}\n{repo_name}")

    def worker(self) -> list[RepoStatus]:
        workflow_info = []
        repos = self.get_list_repos()
        if not repos:
            raise Exception("Failed to get list of repos")
        for repo in repos:
            repo_info = self.get_repo_info(repo["name"])

            if repo_info.get("workflow_runs"):
                if repo_info["workflow_runs"][0].get("conclusion") is None:
                    status = "in_progress"
                else:
                    status = repo_info["workflow_runs"][0]["conclusion"]
            else:
                status = "unknown"
            workflow_info.append(RepoStatus(
                repo_name=repo["name"], 
                status=status, 
                workflow_url=repo["html_url"]
            ))
        
        return workflow_info