(
    event: Dict[str, Any],
    *,
    request: Request,
    x_github_event: str = Header(None),
    x_hub_signature: str = Header(None),
)
Verify and accept GitHub Webhook payloads.
Verify and accept GitHub Webhook payloads.
Verify and accept GitHub Webhook payloads .
async def webhook_event(
    event: Dict[str, Any],
    *,
    request: Request,
    x_github_event: str = Header(None),
    x_hub_signature: str = Header(None),
) -> None:
    """
    Verify and accept GitHub Webhook payloads.
    """
    # FastAPI allows x_github_event to be nullable and we cannot type it as
    # Optional in the function definition
    # https://github.com/tiangolo/fastapi/issues/179
    github_event = cast(Optional[str], x_github_event)
    github_signature = cast(Optional[str], x_hub_signature)
    expected_sha = hmac.new(
        key=conf.SECRET_KEY.encode(), msg=(await request.body()), digestmod=hashlib.sha1
    ).hexdigest()
    if github_event is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Missing required header: X-Github-Event",
        )
    if github_signature is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Missing required signature: X-Hub-Signature",
        )
    sha = github_signature.replace("sha1=", "")
    if not hmac.compare_digest(sha, expected_sha):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid signature: X-Hub-Signature",
        )
    installation_id: int | None = event.get("installation", {}).get("id")
    assert installation_id is not None

    ingest_queue = get_ingest_queue(installation_id)
    redis = await get_redis()
    await redis.rpush(
        ingest_queue,
        [RawWebhookEvent(event_name=github_event, payload=event).json()],
    )

    await redis.ltrim(ingest_queue, 0, conf.INGEST_QUEUE_LENGTH)
    await redis.sadd(INGEST_QUEUE_NAMES, [ingest_queue])
    await redis.publish(
        QUEUE_PUBSUB_INGEST,
        PubsubIngestQueueSchema(installation_id=installation_id).json(),
    )
async def webhook_event ( event : Dict [ str , Any ] , * , request : Request , x_github_event : str = Header ( None ) , x_hub_signature : str = Header ( None ) , ) -> None : # FastAPI allows x_github_event to be nullable and we cannot type it as # Optional in the function definition # https://github.com/tiangolo/fastapi/issues/179 github_event = cast ( Optional [ str ] , x_github_event ) github_signature = cast ( Optional [ str ] , x_hub_signature ) expected_sha = hmac . new ( key = conf . SECRET_KEY . encode ( ) , msg = ( await request . body ( ) ) , digestmod = hashlib . sha1 ) . hexdigest ( ) if github_event is None : raise HTTPException ( status_code = status . HTTP_400_BAD_REQUEST , detail = "Missing required header: X-Github-Event" , ) if github_signature is None : raise HTTPException ( status_code = status . HTTP_400_BAD_REQUEST , detail = "Missing required signature: X-Hub-Signature" , ) sha = github_signature . replace ( "sha1=" , "" ) if not hmac . compare_digest ( sha , expected_sha ) : raise HTTPException ( status_code = status . HTTP_400_BAD_REQUEST , detail = "Invalid signature: X-Hub-Signature" , ) installation_id : int | None = event . get ( "installation" , { } ) . get ( "id" ) assert installation_id is not None ingest_queue = get_ingest_queue ( installation_id ) redis = await get_redis ( ) await redis . rpush ( ingest_queue , [ RawWebhookEvent ( event_name = github_event , payload = event ) . json ( ) ] , ) await redis . ltrim ( ingest_queue , 0 , conf . INGEST_QUEUE_LENGTH ) await redis . sadd ( INGEST_QUEUE_NAMES , [ ingest_queue ] ) await redis . publish ( QUEUE_PUBSUB_INGEST , PubsubIngestQueueSchema ( installation_id = installation_id ) . json ( ) , )