BioArchLinux/nvchecker
2
0
Fork 0
mirror of https://github.com/lilydjwg/nvchecker.git synced 2025-03-10 06:14:02 +00:00
Code Issues Projects Releases Packages Wiki Activity

ii

Browse source
This commit is contained in:
envolution 2024-11-19 22:51:56 -05:00
parent 8c1448193b
commit ec556f6957
4 changed files with 546 additions and 200 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

55
Structuregraphql.py Normal file
View file

@ -0,0 +1,55 @@
# Extract important variables from the GitHub GraphQL JSON response
data = j["data"]
# Rate limit information
rate_limit = {
"max_requests": data["rateLimit"]["limit"], # Maximum allowed requests
"remaining_requests": data["rateLimit"]["remaining"], # Remaining requests in the current window
"reset_time": data["rateLimit"]["resetAt"], # Time when the rate limit resets
}
# Repository information
repository = data["repository"]
# Default branch commit history
default_branch_commit = {
"total_commits": repository["defaultBranchRef"]["target"]["history"]["totalCount"], # Total number of commits
"latest_commit_oid": repository["defaultBranchRef"]["target"]["history"]["edges"][0]["node"]["oid"], # Latest commit hash (OID)
"latest_commit_date": repository["defaultBranchRef"]["target"]["history"]["edges"][0]["node"]["committedDate"], # Latest commit date
}
# Tags information
tags = [
edge["node"]["name"] for edge in repository["refs"]["edges"]
] # List of tag names (if available)
# Releases information
releases = [
{
"name": release["node"]["name"], # Release name
"url": release["node"]["url"], # Release URL
"tag": release["node"]["tagName"], # Tag associated with the release
"is_prerelease": release["node"]["isPrerelease"], # Whether this is a pre-release
"is_latest": release["node"]["isLatest"], # Whether this is the latest release
"created_at": release["node"]["createdAt"], # Release creation date
}
for release in repository["releases"]["edges"]
]
# Pagination info for releases
releases_pagination = {
"has_next_page": repository["releases"]["pageInfo"]["hasNextPage"], # Whether there are more releases
"end_cursor": repository["releases"]["pageInfo"]["endCursor"], # Cursor for the next page of releases
}
# Organized result as a dictionary
result = {
"rate_limit": rate_limit,
"default_branch_commit": default_branch_commit,
"tags": tags,
"releases": releases,
"releases_pagination": releases_pagination,
}
# Example of accessing the organized data
print(result)

14
graphqlquery.txt
View file

@ -1,5 +1,10 @@
query() {
repository(owner: "GNOME", name: "gnome-shell") {
query {
rateLimit {
limit
remaining
resetAt
}
repository(owner: "drwetter", name: "testssl.sh") {
# Default branch commits
defaultBranchRef {
target {
@ -24,19 +29,22 @@ query() {
target {
... on Commit {
oid
url
}
}
}
}
}
# All releases (filter pre-releases in your application logic)
releases(first: 100) {
releases(first: 100, orderBy: { field: CREATED_AT, direction: DESC }) {
totalCount
edges {
node {
name
url
tagName
isPrerelease
isLatest
createdAt
}
}

274
nvchecker_source/github-test.py Normal file
View file

@ -0,0 +1,274 @@
import os # Added for environment variable access
import time
from urllib.parse import urlencode
from typing import List, Tuple, Union, Optional
import asyncio
import json # Added for JSON handling
import structlog
from nvchecker.api import (
VersionResult, Entry, AsyncCache, KeyManager,
HTTPError, session, RichResult, GetVersionError,
)
logger = structlog.get_logger(logger_name=__name__)
ALLOW_REQUEST = None
RATE_LIMITED_ERROR = False
GITHUB_GRAPHQL_URL = 'https://api.%s/graphql'
async def get_version(name, conf, **kwargs):
global RATE_LIMITED_ERROR, ALLOW_REQUEST
if RATE_LIMITED_ERROR:
raise RuntimeError('rate limited')
if ALLOW_REQUEST is None:
ALLOW_REQUEST = asyncio.Event()
ALLOW_REQUEST.set()
for _ in range(2): # retry once
try:
await ALLOW_REQUEST.wait()
return await get_version_real(name, conf, **kwargs)
except HTTPError as e:
if e.code in [403, 429]:
if n := check_ratelimit(e, name):
ALLOW_REQUEST.clear()
await asyncio.sleep(n+1)
ALLOW_REQUEST.set()
continue
RATE_LIMITED_ERROR = True
raise
QUERY_GITHUB = """
query {
rateLimit {
limit
remaining
resetAt
}
repository(owner: "$name", name: "$owner") {
# Default branch commits
defaultBranchRef {
target {
... on Commit {
history(first: 1) {
totalCount
edges {
node {
oid
committedDate
}
}
}
}
}
}
# All tags
refs(refPrefix: "refs/tags/", first: 1, orderBy: {
field: TAG_COMMIT_DATE,
direction: DESC})
{
edges {
node {
name
target {
... on Commit {
oid
url
}
}
}
}
}
# All releases (filter pre-releases in your application logic)
releases(first: 100, orderBy: { field: CREATED_AT, direction: DESC }) {
totalCount
edges {
node {
name
url
tagName
isPrerelease
isLatest
createdAt
}
}
pageInfo {
hasNextPage
endCursor
}
}
}
}
"""
async def get_latest_tag(key: Tuple[str, str, str, str]) -> RichResult:
host, repo, query, token = key
owner, reponame = repo.split('/')
headers = {
'Authorization': f'bearer {token}',
'Content-Type': 'application/json',
}
# Make GraphQL query
query_vars = QUERY_GITHUB.replace("$owner", owner).replace("$name", reponame)
async with session.post(
GITHUB_GRAPHQL_URL % host,
headers=headers,
json={'query': query_vars}
) as res:
j = await res.json()
if 'errors' in j:
raise GetVersionError(f"GitHub API error: {j['errors']}")
refs = j['data']['repository']['refs']['edges']
if not refs:
raise GetVersionError('no tag found')
version = refs[0]['node']['name']
revision = refs[0]['node']['target']['oid']
return RichResult(
version=version,
gitref=f"refs/tags/{version}",
revision=revision,
url=f'https://github.com/{repo}/releases/tag/{version}',
)
async def get_latest_release_with_prereleases(key: Tuple[str, str, str, str]) -> RichResult:
host, repo, token, use_release_name = key
owner, reponame = repo.split('/')
headers = {
'Authorization': f'bearer {token}',
'Content-Type': 'application/json',
}
# Make GraphQL query
query_vars = QUERY_GITHUB.replace("$owner", owner).replace("$name", reponame)
async with session.post(
GITHUB_GRAPHQL_URL % host,
headers=headers,
json={'query': query_vars}
) as res:
j = await res.json()
if 'errors' in j:
raise GetVersionError(f"GitHub API error: {j['errors']}")
releases = j['data']['repository']['releases']['edges']
if not releases:
raise GetVersionError('no release found')
latest_release = releases[0]['node']
tag_name = latest_release['tagName']
version = latest_release['name'] if use_release_name else tag_name
return RichResult(
version=version,
gitref=f"refs/tags/{tag_name}",
revision=latest_release['target']['oid'],
url=latest_release['url'],
)
async def get_version_real(
name: str, conf: Entry, *,
cache: AsyncCache, keymanager: KeyManager,
**kwargs,
) -> VersionResult:
repo = conf['github']
owner, reponame = repo.split('/')
host = conf.get('host', "github.com")
# Load token from config
token = conf.get('token')
# Load token from keyman
if token is None:
token = keymanager.get_key(host.lower(), 'github')
# Load token from environment
if token is None:
token = os.environ.get('GITHUB_TOKEN')
use_latest_tag = conf.get('use_latest_tag', False)
if use_latest_tag:
if not token:
raise GetVersionError('token not given but it is required')
query = conf.get('query', '')
return await cache.get((host, repo, query, token), get_latest_tag)
headers = {
'Authorization': f'bearer {token}',
'Content-Type': 'application/json',
}
# Make GraphQL query
query_vars = QUERY_GITHUB.replace("$owner", owner).replace("$name", reponame)
async with session.post(
GITHUB_GRAPHQL_URL % host,
headers=headers,
json={'query': query_vars}
) as res:
j = await res.json()
if 'errors' in j:
raise GetVersionError(f"GitHub API error: {j['errors']}")
use_max_tag = conf.ger('use_max_tag', False)
if use_max_tag:
refs = j['data']['repository']['refs']['edges']
tags: List[Union[str, RichResult]] = [
RichResult(
version=ref['node']['name'],
gitref=f"refs/tags/{ref['node']['name']}",
revision=ref['node']['target']['oid'],
url=f'https://github.com/{repo}/releases/tag/{ref["node"]["name"]}',
) for ref in refs
]
if not tags:
raise GetVersionError('No tag found in upstream repository.')
return tags
use_latest_release = conf.ger('use_latest_release', False)
if use_latest_release:
releases = j['data']['repository']['releases']['edges']
if not releases:
raise GetVersionError('No release found in upstream repository.')
latest_release = releases[0]['node']
use_release_name = conf.ger('use_release_name', False)
version = latest_release['name'] if use_release_name else latest_release['tagName']
return RichResult(
version=version,
gitref=f"refs/tags/{latest_release['tagName']}",
url=latest_release['url'],
)
else:
commit = j['data']['repository']['defaultBranchRef']['target']['history']['edges'][0]['node']
return RichResult(
version=commit['committedDate'].rstrip('Z').replace('-', '').replace(':', '').replace('T', '.'),
revision=commit['oid'],
url=f'https://github.com/{repo}/commit/{commit["oid"]}',
)
def check_ratelimit(exc: HTTPError, name: str) -> Optional[int]:
res = exc.response
if not res:
raise exc
if v := res.headers.get('retry-after'):
n = int(v)
logger.warning('retry-after', n=n)
return n
# default -1 is used to re-raise the exception
n = int(res.headers.get('X-RateLimit-Remaining', -1))
if n == 0:
reset = int(res.headers.get('X-RateLimit-Reset'))
logger.error(f'rate limited, resetting at {time.ctime(reset)}. '
'Or get an API token to increase the allowance if not yet',
name = name,
reset = reset)
return None
raise exc

403
nvchecker_source/github.py
View file

@ -1,47 +1,13 @@
# MIT licensed
# Copyright (c) 2013-2020, 2024 lilydjwg <lilydjwg@gmail.com>, et al.
#
import time
import os
from urllib.parse import urlencode, parse_qs, urlparse
from urllib.parse import urlencode
from typing import List, Tuple, Union, Optional
import asyncio
from nvchecker.api import KeyManager
import structlog
def get_github_token(conf: dict, host: str, keymanager: KeyManager) -> Optional[str]:
"""
Get GitHub token with the following priority:
1. Token from config
2. Token from keymanager
3. Token from GITHUB_TOKEN environment variable
Args:
conf: Configuration dictionary
host: GitHub host (e.g., "github.com")
keymanager: KeyManager instance for managing tokens
Returns:
str or None: GitHub token if found, None otherwise
"""
# Check config first
token = conf.get('token')
if token is not None:
return token
# Then check keymanager
try:
token = keymanager.get_key(host.lower(), 'github')
if token:
return token
except Exception:
pass
# Finally check environment variable
return os.environ.get('GITHUB_TOKEN')
from nvchecker.api import (
VersionResult, Entry, AsyncCache, KeyManager,
HTTPError, session, RichResult, GetVersionError,
@ -53,32 +19,10 @@ RATE_LIMITED_ERROR = False
GITHUB_URL = 'https://api.%s/repos/%s/commits'
GITHUB_LATEST_RELEASE = 'https://api.%s/repos/%s/releases/latest'
# https://developer.github.com/v3/git/refs/#get-all-references
GITHUB_MAX_TAG = 'https://api.%s/repos/%s/git/refs/tags'
GITHUB_GRAPHQL_URL = 'https://api.%s/graphql'
async def get_commit_count(url: str, headers: dict) -> int:
"""Get the total commit count using pagination."""
params = {'per_page': '1'}
response = await session.get(
url,
params=params,
headers=headers
)
commit_count = 1
if 'Link' in response.headers:
link_header = response.headers['Link']
for link in link_header.split(', '):
if 'rel="last"' in link:
url = link[link.find("<") + 1:link.find(">")]
query_params = parse_qs(urlparse(url).query)
if 'page' in query_params:
commit_count = int(query_params['page'][0])
break
return commit_count
async def get_version(name, conf, **kwargs):
global RATE_LIMITED_ERROR, ALLOW_REQUEST
@ -103,160 +47,225 @@ async def get_version(name, conf, **kwargs):
RATE_LIMITED_ERROR = True
raise
async def enhance_version_with_commit_info(
result: RichResult,
host: str,
repo: str,
headers: dict,
use_commit_info: bool
) -> RichResult:
"""Add commit count and SHA to version if use_commit_info is True."""
if not use_commit_info:
return result
url = GITHUB_URL % (host, repo)
commit_count = await get_commit_count(url, headers)
# Create new version string with commit info
enhanced_version = f"{result.version}.r{commit_count}.g{result.revision[:9]}"
return RichResult(
version=enhanced_version,
gitref=result.gitref,
revision=result.revision,
url=result.url
)
QUERY_LATEST_TAG = '''
{{
repository(name: "{name}", owner: "{owner}") {{
refs(refPrefix: "refs/tags/", first: 1,
query: "{query}",
orderBy: {{field: TAG_COMMIT_DATE, direction: DESC}}) {{
edges {{
node {{
name
target {{
oid
}}
}}
}}
}}
}}
}}
'''
QUERY_LATEST_RELEASE_WITH_PRERELEASES = '''
{{
repository(name: "{name}", owner: "{owner}") {{
releases(first: 1, orderBy: {{field: CREATED_AT, direction: DESC}}) {{
edges {{
node {{
name
url
tag {{
name
}}
tagCommit {{
oid
}}
}}
}}
}}
}}
}}
'''
async def get_latest_tag(key: Tuple[str, str, str, str]) -> RichResult:
host, repo, query, token = key
owner, reponame = repo.split('/')
headers = {
'Authorization': f'bearer {token}',
'Content-Type': 'application/json',
}
q = QUERY_LATEST_TAG.format(
owner = owner,
name = reponame,
query = query,
)
res = await session.post(
GITHUB_GRAPHQL_URL % host,
headers = headers,
json = {'query': q},
)
j = res.json()
refs = j['data']['repository']['refs']['edges']
if not refs:
raise GetVersionError('no tag found')
version = refs[0]['node']['name']
revision = refs[0]['node']['target']['oid']
return RichResult(
version = version,
gitref = f"refs/tags/{version}",
revision = revision,
url = f'https://github.com/{repo}/releases/tag/{version}',
)
async def get_latest_release_with_prereleases(key: Tuple[str, str, str, str]) -> RichResult:
host, repo, token, use_release_name = key
owner, reponame = repo.split('/')
headers = {
'Authorization': f'bearer {token}',
'Content-Type': 'application/json',
}
q = QUERY_LATEST_RELEASE_WITH_PRERELEASES.format(
owner = owner,
name = reponame,
)
res = await session.post(
GITHUB_GRAPHQL_URL % host,
headers = headers,
json = {'query': q},
)
j = res.json()
refs = j['data']['repository']['releases']['edges']
if not refs:
raise GetVersionError('no release found')
tag_name = refs[0]['node']['tag']['name']
if use_release_name:
version = refs[0]['node']['name']
else:
version = tag_name
return RichResult(
version = version,
gitref = f"refs/tags/{tag_name}",
revision = refs[0]['node']['tagCommit']['oid'],
url = refs[0]['node']['url'],
)
async def get_version_real(
name: str, conf: Entry, *,
cache: AsyncCache, keymanager: KeyManager,
**kwargs,
name: str, conf: Entry, *,
cache: AsyncCache, keymanager: KeyManager,
**kwargs,
) -> VersionResult:
repo = conf['github']
host = conf.get('host', "github.com")
use_commit_info = conf.get('use_commit_info', False)
repo = conf['github']
host = conf.get('host', "github.com")
# Load token from config, keymanager or env GITHUB_TOKEN
token = get_github_token(conf, host, keymanager)
# Load token from config
token = conf.get('token')
# Load token from keyman
if token is None:
token = keymanager.get_key(host.lower(), 'github')
headers = {
'Accept': 'application/vnd.github.quicksilver-preview+json',
}
if token:
if token.startswith('github_pat_'):
headers['Authorization'] = f'Bearer {token}'
else:
headers['Authorization'] = f'token {token}'
use_latest_tag = conf.get('use_latest_tag', False)
if use_latest_tag:
if not token:
raise GetVersionError('token not given but it is required')
use_latest_tag = conf.get('use_latest_tag', False)
use_latest_release = conf.get('use_latest_release', False)
include_prereleases = conf.get('include_prereleases', False)
use_max_tag = conf.get('use_max_tag', False)
use_release_name = conf.get('use_release_name', False)
query = conf.get('query', '')
return await cache.get((host, repo, query, token), get_latest_tag) # type: ignore
# Token requirement checks
if any([use_latest_tag, (use_latest_release and include_prereleases), use_max_tag]) and not token:
raise GetVersionError('token not given but it is required for this operation')
use_latest_release = conf.get('use_latest_release', False)
include_prereleases = conf.get('include_prereleases', False)
use_release_name = conf.get('use_release_name', False)
if use_latest_release and include_prereleases:
if not token:
raise GetVersionError('token not given but it is required')
try:
if use_latest_tag:
query = conf.get('query', '')
result = await cache.get((host, repo, query, token), get_latest_tag)
return await enhance_version_with_commit_info(result, host, repo, headers, use_commit_info)
return await cache.get(
(host, repo, token, use_release_name),
get_latest_release_with_prereleases) # type: ignore
if use_latest_release:
url = GITHUB_LATEST_RELEASE % (host, repo)
try:
data = await cache.get_json(url, headers=headers)
if 'tag_name' not in data:
raise GetVersionError('No release found in upstream repository.')
br = conf.get('branch')
path = conf.get('path')
use_max_tag = conf.get('use_max_tag', False)
if use_latest_release:
url = GITHUB_LATEST_RELEASE % (host, repo)
elif use_max_tag:
url = GITHUB_MAX_TAG % (host, repo)
else:
url = GITHUB_URL % (host, repo)
parameters = {}
if br:
parameters['sha'] = br
if path:
parameters['path'] = path
url += '?' + urlencode(parameters)
headers = {
'Accept': 'application/vnd.github.quicksilver-preview+json',
}
if token:
headers['Authorization'] = f'token {token}'
version = data['name'] if use_release_name else data['tag_name']
result = RichResult(
version=version,
gitref=f"refs/tags/{data['tag_name']}",
url=data['html_url'],
)
return await enhance_version_with_commit_info(result, host, repo, headers, use_commit_info)
except HTTPError as e:
if e.code == 404:
raise GetVersionError(f'No releases found for repository {repo}. The repository might not have any releases yet.')
raise
data = await cache.get_json(url, headers = headers)
if use_max_tag:
url = GITHUB_MAX_TAG % (host, repo)
try:
data = await cache.get_json(url, headers=headers)
tags: List[Union[str, RichResult]] = [
RichResult(
version=ref['ref'].split('/', 2)[-1],
gitref=ref['ref'],
revision=ref['object']['sha'],
url=f'https://github.com/{repo}/releases/tag/{ref["ref"].split("/", 2)[-1]}',
) for ref in data
]
if not tags:
raise GetVersionError('No tags found in upstream repository.')
if use_commit_info:
return [await enhance_version_with_commit_info(
tag, host, repo, headers, use_commit_info
) for tag in tags if isinstance(tag, RichResult)]
return tags
except HTTPError as e:
if e.code == 404:
raise GetVersionError(f'No tags found for repository {repo}. The repository might not have any tags yet.')
raise
if use_max_tag:
tags: List[Union[str, RichResult]] = [
RichResult(
version = ref['ref'].split('/', 2)[-1],
gitref = ref['ref'],
revision = ref['object']['sha'],
url = f'https://github.com/{repo}/releases/tag/{ref["ref"].split("/", 2)[-1]}',
) for ref in data
]
if not tags:
raise GetVersionError('No tag found in upstream repository.')
return tags
# Default: use commits
br = conf.get('branch')
path = conf.get('path')
url = GITHUB_URL % (host, repo)
parameters = {}
if br:
parameters['sha'] = br
if path:
parameters['path'] = path
if parameters:
url += '?' + urlencode(parameters)
if use_latest_release:
if 'tag_name' not in data:
raise GetVersionError('No release found in upstream repository.')
data = await cache.get_json(url, headers=headers)
result = RichResult(
version=data[0]['commit']['committer']['date'].rstrip('Z').replace('-', '').replace(':', '').replace('T', '.'),
revision=data[0]['sha'],
url=data[0]['html_url'],
)
return await enhance_version_with_commit_info(result, host, repo, headers, use_commit_info)
if use_release_name:
version = data['name']
else:
version = data['tag_name']
except HTTPError as e:
if e.code == 404:
raise GetVersionError(f'Repository {repo} not found or access denied.')
elif e.code in [403, 429]:
if n := check_ratelimit(e, name):
raise GetVersionError(f'Rate limited. Try again in {n} seconds or use an API token.')
raise GetVersionError('Rate limit exceeded. Please use an API token to increase the allowance.')
raise
return RichResult(
version = version,
gitref = f"refs/tags/{data['tag_name']}",
url = data['html_url'],
)
else:
return RichResult(
# YYYYMMDD.HHMMSS
version = data[0]['commit']['committer']['date'].rstrip('Z').replace('-', '').replace(':', '').replace('T', '.'),
revision = data[0]['sha'],
url = data[0]['html_url'],
)
def check_ratelimit(exc: HTTPError, name: str) -> Optional[int]:
res = exc.response
if not res:
raise exc
res = exc.response
if not res:
raise exc
if v := res.headers.get('retry-after'):
n = int(v)
logger.warning('retry-after', n=n)
return n
if v := res.headers.get('retry-after'):
n = int(v)
logger.warning('retry-after', n=n)
return n
# default -1 is used to re-raise the exception
n = int(res.headers.get('X-RateLimit-Remaining', -1))
if n == 0:
reset = int(res.headers.get('X-RateLimit-Reset'))
logger.error(f'rate limited, resetting at {time.ctime(reset)}. '
'Or get an API token to increase the allowance if not yet',
name=name,
reset=reset)
return None
# default -1 is used to re-raise the exception
n = int(res.headers.get('X-RateLimit-Remaining', -1))
if n == 0:
reset = int(res.headers.get('X-RateLimit-Reset'))
logger.error(f'rate limited, resetting at {time.ctime(reset)}. '
'Or get an API token to increase the allowance if not yet',
name = name,
reset = reset)
return None
raise exc
raise exc