-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgithub_auth.py
More file actions
81 lines (68 loc) · 2.42 KB
/
Copy pathgithub_auth.py
File metadata and controls
81 lines (68 loc) · 2.42 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import os
import requests
GITHUB_AUTHORIZE_URL = "https://github.com/login/oauth/authorize"
GITHUB_TOKEN_URL = "https://github.com/login/oauth/access_token"
GITHUB_API_URL = "https://api.github.com/user"
def oauth_configured() -> bool:
return bool(os.getenv("GITHUB_CLIENT_ID") and os.getenv("GITHUB_CLIENT_SECRET"))
def get_redirect_uri() -> str:
"""
Resolve the OAuth callback URL.
Set APP_URL in .env (local) or Streamlit Cloud secrets (production).
Defaults to localhost for local development.
"""
url = os.getenv("APP_URL", "http://localhost:8501").rstrip("/")
return url
def get_auth_url() -> str:
client_id = os.getenv("GITHUB_CLIENT_ID")
redirect_uri = get_redirect_uri()
return (
f"{GITHUB_AUTHORIZE_URL}"
f"?client_id={client_id}"
f"&redirect_uri={redirect_uri}"
f"&scope=repo"
)
def exchange_code(code: str) -> str:
"""Exchange OAuth code for access token. Returns the token string."""
resp = requests.post(
GITHUB_TOKEN_URL,
headers={"Accept": "application/json"},
data={
"client_id": os.getenv("GITHUB_CLIENT_ID"),
"client_secret": os.getenv("GITHUB_CLIENT_SECRET"),
"code": code,
},
timeout=10,
)
resp.raise_for_status()
data = resp.json()
if "access_token" not in data:
raise ValueError(f"GitHub OAuth failed: {data.get('error_description', data)}")
return data["access_token"]
def get_github_user(token: str) -> dict:
"""Return GitHub user info dict (login, avatar_url, name)."""
resp = requests.get(
GITHUB_API_URL,
headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"},
timeout=10,
)
resp.raise_for_status()
return resp.json()
def list_user_repos(token: str) -> list[dict]:
"""Return list of repos the authenticated user has access to."""
repos = []
page = 1
while len(repos) < 100:
resp = requests.get(
"https://api.github.com/user/repos",
headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"},
params={"per_page": 50, "page": page, "sort": "pushed", "affiliation": "owner,collaborator"},
timeout=10,
)
resp.raise_for_status()
batch = resp.json()
if not batch:
break
repos.extend(batch)
page += 1
return repos