Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def backport_task_asyncio(
commit_hash, branch, *, issue_number, created_by, merged_by
):
"""Backport a commit into a branch."""
oauth_token = os.environ.get("GH_AUTH")
async with aiohttp.ClientSession() as session:
gh = gh_aiohttp.GitHubAPI(
session, "python/cpython", oauth_token=oauth_token, cache=cache
)
if not util.is_cpython_repo():
# cd to cpython if we're not already in it
if "cpython" in os.listdir("."):
os.chdir("./cpython")
else:
print(f"pwd: {os.getcwd()}, listdir: {os.listdir('.')}")
await util.comment_on_pr(
gh,
issue_number,
f"""{util.get_participants(created_by, merged_by)}, I can't backport for now. Please try again later or
backport using [cherry_picker](https://pypi.org/project/cherry-picker/) on command line.
```
async def main(request):
try:
body = await request.read()
secret = os.environ.get("GH_SECRET")
event = sansio.Event.from_http(request.headers, body, secret=secret)
print("GH delivery ID", event.delivery_id, file=sys.stderr)
if event.event == "ping":
return web.Response(status=200)
oauth_token = os.environ.get("GH_AUTH")
async with aiohttp.ClientSession() as session:
gh = gh_aiohttp.GitHubAPI(
session,
os.environ.get("GH_USERNAME"),
oauth_token=oauth_token,
cache=cache,
)
# Give GitHub some time to reach internal consistency.
await asyncio.sleep(1)
await router.dispatch(event, gh)
try:
print("GH requests remaining:", gh.rate_limit.remaining)
except AttributeError:
pass
return web.Response(status=200)
except Exception as exc:
traceback.print_exc(file=sys.stderr)
return web.Response(status=500)
async def on_startup(app):
token_file = os.environ.get('GITHUB_TOKEN_PATH',
'/secrets/scorecard-github-access-token.txt')
with open(token_file, 'r') as f:
token = f.read().strip()
session = aiohttp.ClientSession(
raise_for_status=True,
timeout=aiohttp.ClientTimeout(total=60))
gh_client = gidgethub.aiohttp.GitHubAPI(session, 'scorecard', oauth_token=token)
app['gh_client'] = gh_client
await update_data(gh_client)
asyncio.ensure_future(poll(gh_client))
async def main(request):
try:
body = await request.read()
secret = os.environ.get("GH_SECRET")
event = sansio.Event.from_http(request.headers, body, secret=secret)
print("GH delivery ID", event.delivery_id, file=sys.stderr)
if event.event == "ping":
return web.Response(status=200)
oauth_token = os.environ.get("GH_AUTH")
async with aiohttp.ClientSession() as session:
gh = gh_aiohttp.GitHubAPI(
session, "python/cpython", oauth_token=oauth_token, cache=cache
)
# Give GitHub some time to reach internal consistency.
await asyncio.sleep(1)
await router.dispatch(event, gh)
try:
print(
f"""\
GH requests remaining: {gh.rate_limit.remaining}/{gh.rate_limit.limit}, \
reset time: {gh.rate_limit.reset_datetime:%b-%d-%Y %H:%M:%S %Z}, \
oauth token length {len(oauth_token)}, \
last 4 digits {oauth_token[-4:]}, \
GH delivery ID {event.delivery_id} \
"""
)
except AttributeError:
async def add_pin_to_pinboard(self, channel, data):
if config.github_oauth_token == "":
# Don't add to gist pinboard if we don't have an oauth token
return
async with aiohttp.ClientSession() as session:
gh = gidgethub.aiohttp.GitHubAPI(
session, "RoboCop-NG", oauth_token=config.github_oauth_token
)
(id, content) = await self.get_pinboard(gh, channel)
content += "- " + data + "\n"
await gh.patch(
f"/gists/{id}", data={"files": {"pinboard.md": {"content": content}}}
)
async def get_installation_token(self, installation: str, name: str = None) -> str:
"""Returns OAUTH token for installation referenced in **event**"""
if name is None:
name = installation
now = int(time.time())
expires, token = self._tokens.get(installation, (0, ''))
if not expires or expires < now + 60:
api = gidgethub.aiohttp.GitHubAPI(self._session, self.name, cache=self._cache)
try:
res = await api.post(
self.INSTALLATION_TOKEN,
{'installation_id': installation},
data=b"",
accept="application/vnd.github.machine-man-preview+json",
jwt=self.get_app_jwt()
)
except gidgethub.BadRequest:
logger.exception("Failed to get installation token for %s", name)
raise
expires = self.parse_isotime(res['expires_at'])
token = res['token']
self._tokens[installation] = (expires, token)
msg = "Created new"
async def main(request):
try:
body = await request.read()
secret = os.environ.get("GH_SECRET")
event = sansio.Event.from_http(request.headers, body, secret=secret)
print('GH delivery ID', event.delivery_id, file=sys.stderr)
if event.event == "ping":
return web.Response(status=200)
oauth_token = os.environ.get("GH_AUTH")
async with aiohttp.ClientSession() as session:
gh = gh_aiohttp.GitHubAPI(session, "python/bedevere",
oauth_token=oauth_token,
cache=cache)
# Give GitHub some time to reach internal consistency.
await asyncio.sleep(1)
await router.dispatch(event, gh, session=session)
try:
print('GH requests remaining:', gh.rate_limit.remaining)
except AttributeError:
pass
return web.Response(status=200)
except Exception as exc:
traceback.print_exc(file=sys.stderr)
return web.Response(status=500)
async def on_startup(app):
session = aiohttp.ClientSession(
raise_for_status=True,
timeout=aiohttp.ClientTimeout(total=60))
app['client_session'] = session
app['github_client'] = gh_aiohttp.GitHubAPI(session, 'ci', oauth_token=oauth_token)
app['batch_client'] = await BatchClient('ci', session=session)
with open('/ci-user-secret/sql-config.json', 'r') as f:
config = json.loads(f.read().strip())
app['dbpool'] = await aiomysql.create_pool(host=config['host'],
port=config['port'],
db=config['db'],
user=config['user'],
password=config['password'],
charset='utf8',
cursorclass=aiomysql.cursors.DictCursor,
autocommit=True)
asyncio.ensure_future(update_loop(app))