1
1
import requests
2
+ import jwt
3
+ import time
4
+ from datetime import datetime , timedelta
2
5
from dotenv import load_dotenv
6
+ import os
3
7
4
8
load_dotenv ()
5
9
6
10
7
11
class GitHubService :
8
- def __init__ (self , github_token ):
9
- self .github_token = github_token
10
- self .headers = {
11
- "Authorization" : f"token { self .github_token } " ,
12
- "Accept" : "application/vnd.github+json"
12
+ def __init__ (self ):
13
+ # Try app authentication first
14
+ self .client_id = os .getenv ("GITHUB_CLIENT_ID" )
15
+ self .private_key = os .getenv ("GITHUB_PRIVATE_KEY" )
16
+ self .installation_id = os .getenv ("GITHUB_INSTALLATION_ID" )
17
+
18
+ # Fallback to PAT if app credentials not found
19
+ self .github_token = os .getenv ("GITHUB_PAT" )
20
+
21
+ if not all ([self .client_id , self .private_key , self .installation_id ]) and not self .github_token :
22
+ raise ValueError (
23
+ "Either GitHub App credentials or PAT must be provided" )
24
+
25
+ self .access_token = None
26
+ self .token_expires_at = None
27
+
28
+ # autopep8: off
29
+ def _generate_jwt (self ):
30
+ now = int (time .time ())
31
+ payload = {
32
+ "iat" : now ,
33
+ "exp" : now + (10 * 60 ), # 10 minutes
34
+ "iss" : self .client_id
35
+ }
36
+ # Convert PEM string format to proper newlines
37
+ return jwt .encode (payload , self .private_key , algorithm = "RS256" ) # type: ignore
38
+ # autopep8: on
39
+
40
+ def _get_installation_token (self ):
41
+ if self .access_token and self .token_expires_at > datetime .now (): # type: ignore
42
+ return self .access_token
43
+
44
+ jwt_token = self ._generate_jwt ()
45
+ response = requests .post (
46
+ f"https://api.github.com/app/installations/{
47
+ self .installation_id } /access_tokens" ,
48
+ headers = {
49
+ "Authorization" : f"Bearer { jwt_token } " ,
50
+ "Accept" : "application/vnd.github+json"
51
+ }
52
+ )
53
+ data = response .json ()
54
+ self .access_token = data ["token" ]
55
+ self .token_expires_at = datetime .now () + timedelta (hours = 1 )
56
+ return self .access_token
57
+
58
+ def _get_headers (self ):
59
+ # Use PAT if app credentials not available
60
+ if not all ([self .client_id , self .private_key , self .installation_id ]):
61
+ return {
62
+ "Authorization" : f"token { self .github_token } " ,
63
+ "Accept" : "application/vnd.github+json"
64
+ }
65
+
66
+ # Otherwise use app authentication
67
+ token = self ._get_installation_token ()
68
+ return {
69
+ "Authorization" : f"Bearer { token } " ,
70
+ "Accept" : "application/vnd.github+json" ,
71
+ "X-GitHub-Api-Version" : "2022-11-28"
13
72
}
14
73
15
74
def get_default_branch (self , username , repo ):
16
75
"""Get the default branch of the repository."""
17
76
api_url = f"https://api.github.com/repos/{ username } /{ repo } "
18
- response = requests .get (api_url , headers = self .headers )
77
+ response = requests .get (api_url , headers = self ._get_headers () )
19
78
20
79
if response .status_code == 200 :
21
80
return response .json ().get ('default_branch' )
@@ -57,7 +116,7 @@ def should_include_file(path):
57
116
if branch :
58
117
api_url = f"https://api.github.com/repos/{
59
118
username } /{ repo } /git/trees/{ branch } ?recursive=1"
60
- response = requests .get (api_url , headers = self .headers )
119
+ response = requests .get (api_url , headers = self ._get_headers () )
61
120
62
121
if response .status_code == 200 :
63
122
data = response .json ()
@@ -71,7 +130,7 @@ def should_include_file(path):
71
130
for branch in ['main' , 'master' ]:
72
131
api_url = f"https://api.github.com/repos/{
73
132
username } /{ repo } /git/trees/{ branch } ?recursive=1"
74
- response = requests .get (api_url , headers = self .headers )
133
+ response = requests .get (api_url , headers = self ._get_headers () )
75
134
76
135
if response .status_code == 200 :
77
136
data = response .json ()
@@ -96,13 +155,7 @@ def get_github_readme(self, username, repo):
96
155
str: The contents of the README file.
97
156
"""
98
157
api_url = f"https://api.github.com/repos/{ username } /{ repo } /readme"
99
-
100
- headers = {
101
- "Authorization" : f"token { self .github_token } " ,
102
- "Accept" : "application/vnd.github+json"
103
- }
104
-
105
- response = requests .get (api_url , headers = headers )
158
+ response = requests .get (api_url , headers = self ._get_headers ())
106
159
107
160
if response .status_code == 404 :
108
161
raise ValueError ("Repository not found." )
0 commit comments