migrate sphinxcontrib.h5p.utils
This commit is contained in:
parent
cc4b567181
commit
5bf4a7eee4
8 changed files with 742 additions and 0 deletions
78
src/byteb4rb1e/utils/saas/bitbucket.py
Normal file
78
src/byteb4rb1e/utils/saas/bitbucket.py
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Bitbucket Cloud REST API v2.0 wrapper.
|
||||
|
||||
Thin layer over http.py for Bitbucket-specific operations:
|
||||
|
||||
- Bearer token authentication
|
||||
- Repository existence checks
|
||||
- Repository creation within a workspace/project
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from byteb4rb1e.utils.http import client as http_client
|
||||
|
||||
|
||||
BITBUCKET_API = "https://api.bitbucket.org/2.0"
|
||||
|
||||
|
||||
def http_headers(token: str) -> Dict[str, str]:
|
||||
"""Construct Bitbucket API headers with Bearer token auth."""
|
||||
return {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
|
||||
def repository_exists(
|
||||
workspace: str,
|
||||
repo_slug: str,
|
||||
token: str,
|
||||
) -> bool:
|
||||
"""Check whether a repository exists in the workspace."""
|
||||
url = f"{BITBUCKET_API}/repositories/{workspace}/{repo_slug}"
|
||||
resp = http_client.get(url, headers=http_headers(token))
|
||||
return resp.status_code == 200
|
||||
|
||||
|
||||
def create_repository(
|
||||
workspace: str,
|
||||
repo_slug: str,
|
||||
token: str,
|
||||
project: Optional[str] = None,
|
||||
description: str = "",
|
||||
is_private: bool = True,
|
||||
) -> http_client.HttpResponse:
|
||||
"""Create a new repository in the workspace.
|
||||
|
||||
When *project* is given the repository is assigned to that
|
||||
Bitbucket project (by key). This is required for workspaces
|
||||
that scope access keys at the project level.
|
||||
|
||||
Returns the API response. Caller should check status_code == 200
|
||||
for success.
|
||||
"""
|
||||
url = f"{BITBUCKET_API}/repositories/{workspace}/{repo_slug}"
|
||||
body: Dict[str, Any] = {
|
||||
"scm": "git",
|
||||
"is_private": is_private,
|
||||
"description": description,
|
||||
"fork_policy": "no_forks",
|
||||
}
|
||||
if project:
|
||||
body["project"] = {"key": project}
|
||||
return http_client.put(
|
||||
url,
|
||||
data=json.dumps(body).encode("utf-8"),
|
||||
headers=http_headers(token),
|
||||
)
|
||||
|
||||
|
||||
def clone_url(
|
||||
workspace: str,
|
||||
repo_slug: str,
|
||||
) -> str:
|
||||
"""Return the SSH clone URL for a Bitbucket repository."""
|
||||
return f"git@bitbucket.org:{workspace}/{repo_slug}.git"
|
||||
65
src/byteb4rb1e/utils/saas/github.py
Normal file
65
src/byteb4rb1e/utils/saas/github.py
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
#!/usr/bin/env python3
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from byteb4rb1e.utils.http import client as http_client
|
||||
|
||||
|
||||
GITHUB_API = "https://api.github.com"
|
||||
|
||||
|
||||
def http_headers(token: Optional[str]) -> Dict[str, str]:
|
||||
headers = {
|
||||
"Accept": "application/vnd.github+json",
|
||||
"User-Agent": "sphinx-h5p-worker1"
|
||||
}
|
||||
if token:
|
||||
# Use standard PAT header; token not logged anywhere.
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
return headers
|
||||
|
||||
|
||||
def blob_sha(path: Path) -> str:
|
||||
"""Calculate Git blob SHA-1 for a file, matching GitHub API 'sha'."""
|
||||
data = path.read_bytes()
|
||||
header = f"blob {len(data)}\0".encode("utf-8")
|
||||
store = header + data
|
||||
return hashlib.sha1(store).hexdigest()
|
||||
|
||||
|
||||
def list_org_repos(org: str, token: Optional[str]) -> List[Dict[str, Any]]:
|
||||
repos: List[Dict[str, Any]] = []
|
||||
page = 1
|
||||
per_page = 100
|
||||
while True:
|
||||
url = f"{GITHUB_API}/orgs/{org}/repos"
|
||||
resp = http_client.get(
|
||||
url,
|
||||
params={"page": page, "per_page": per_page, "type": "public"},
|
||||
headers=http_headers(token),
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
raise RuntimeError(f"Failed to list repos for org {org}: {resp.status_code} {resp.text}")
|
||||
batch = resp.json()
|
||||
if not batch:
|
||||
break
|
||||
repos.extend(batch)
|
||||
page += 1
|
||||
return repos
|
||||
|
||||
|
||||
def fetch_file(
|
||||
org: str,
|
||||
repo: str,
|
||||
path: str,
|
||||
token: str
|
||||
) -> http_client.HttpResponse:
|
||||
"""
|
||||
"""
|
||||
url = f"{GITHUB_API}/repos/{org}/{repo}/{path}"
|
||||
|
||||
return http_client.get(
|
||||
url,
|
||||
headers=http_headers(token),
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue