()
Pull webhook events off the queue and insert them into Postgres to calculate
    usage statistics.
Pull webhook events off the queue and insert them into Postgres to calculate
    usage statistics.
Pull webhook events off the queue and insert them into Postgres to calculate usage statistics .
def ingest_events() -> None:
    """
    Pull webhook events off the queue and insert them into Postgres to calculate
    usage statistics.
    """
    r = redis.Redis.from_url(os.environ["REDIS_URL"])
    while True:
        time.sleep(0)
        # we don't want to lose events when we terminate the process, so we
        # handle SIGINT and SIGTERM gracefully. We use a short timeout of Redis
        # BLPOP so we don't have to wait too long.
        with GracefulTermination():
            logger.info("block for event")
            res = r.blpop("kodiak:webhook_event", timeout=5)
            if res is None:
                # if res is None we likely hit the timeout.
                continue
            _, event_compressed = res

            logger.info("process event")
            dctx = zstd.ZstdDecompressor()
            decompressed = dctx.decompress(event_compressed)
            event = json.loads(decompressed)

            event_name = event["event_name"]
            if event_name in INTERESTING_EVENTS:
                payload = event["payload"]
                GitHubEvent.objects.create(event_name=event_name, payload=payload)
def ingest_events ( ) -> None : r = redis . Redis . from_url ( os . environ [ "REDIS_URL" ] ) while True : time . sleep ( 0 ) # we don't want to lose events when we terminate the process, so we # handle SIGINT and SIGTERM gracefully. We use a short timeout of Redis # BLPOP so we don't have to wait too long. with GracefulTermination ( ) : logger . info ( "block for event" ) res = r . blpop ( "kodiak:webhook_event" , timeout = 5 ) if res is None : # if res is None we likely hit the timeout. continue _ , event_compressed = res logger . info ( "process event" ) dctx = zstd . ZstdDecompressor ( ) decompressed = dctx . decompress ( event_compressed ) event = json . loads ( decompressed ) event_name = event [ "event_name" ] if event_name in INTERESTING_EVENTS : payload = event [ "payload" ] GitHubEvent . objects . create ( event_name = event_name , payload = payload )