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 process(cls, server: ni_abc.ServerHost,
request: web.Request, client: aiohttp.ClientSession) -> "Host":
"""Process the pull request."""
event = sansio.Event.from_http(request.headers,
await request.read(),
secret=server.contrib_secret())
if event.event == "ping":
# A ping event; nothing to do.
# https://developer.github.com/webhooks/#ping-event
raise ni_abc.ResponseExit(status=http.HTTPStatus.OK)
elif event.event != "pull_request":
# Only happens if GitHub is misconfigured to send the wrong events.
raise TypeError(f"don't know how to handle a {event.event!r} event")
elif event.data['action'] not in cls._useful_actions:
raise ni_abc.ResponseExit(status=http.HTTPStatus.NO_CONTENT)
elif event.data['action'] in {PullRequestEvent.opened.value, PullRequestEvent.synchronize.value}:
if event.data['action'] == PullRequestEvent.opened.value:
# GitHub is eventually consistent, so add a delay to wait for
# the API to digest the new pull request.
await asyncio.sleep(1)
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}, \
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)
async def dispatch(request):
github = request.app.plugins["github"]
payload = await request.read()
try:
event = Event.from_http(request.headers, payload, secret=github.verify)
await github.router.dispatch(event, app=request.app)
except ValidationFailure:
LOG.debug(
"Github webhook failed verification: %s, %s", request.headers, payload
)
return Response(status=401)
except Exception as e:
LOG.exception(e)
return Response(status=500)
else:
return Response(status=200)
async def github_callback_handler(request):
event = gh_sansio.Event.from_http(request.headers, await request.read())
event.app = request.app
await gh_router.dispatch(event)
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)