pincer.core.heartbeat

Complexity

Total Complexity

7

Size/Duplication

Total Lines	119
Duplicated Lines	0 %

Importance

Changes

0

5 Functions

Rating	Name	Duplication	Size	Complexity
A	handle_hello()	0	30	3
A	handle_heartbeat()	0	10	1
A	get_heartbeat()	0	10	1
A	__send_heartbeat()	0	8	1
A	update_sequence()	0	10	1

# -*- coding: utf-8 -*-

Missing module docstring

# MIT License
#
# Copyright (c) 2021 Pincer
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files
# (the "Software"), to deal in the Software without restriction,
# including without limitation the rights to use, copy, modify, merge,
# publish, distribute, sublicense, and/or sell copies of the Software,
# and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

from __future__ import annotations

import logging
from asyncio import sleep
from typing import Optional

from websockets.legacy.client import WebSocketClientProtocol

from pincer import __package__

This seems to re-define the built-in __package__.

from pincer.core.dispatch import GatewayDispatch
from pincer.exceptions import HeartbeatError

heartbeat: float = 0
sequence: Optional[int] = None

log = logging.getLogger(__package__)


def get_heartbeat() -> float:
    """
    Get the current heartbeat.

    :return:
        The current heartbeat of the client.
        Default is 0 (client has not initialized the heartbeat yet.)

    """
    return heartbeat


async def __send_heartbeat(socket: WebSocketClientProtocol):
    """
    Sends a heartbeat to the API gateway.
    """
    global sequence

Constant name "sequence" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

Usage of the global statement should be avoided.

    log.debug(f"Sending heartbeat (seq: {sequence})")

Use lazy % formatting in logging functions

    await socket.send(str(GatewayDispatch(1, sequence)))


async def handle_hello(socket: WebSocketClientProtocol,
                       payload: GatewayDispatch):
    """
    Handshake between the discord API and the client.
    Retrieve the heartbeat for maintaining a connection.
    """
    global heartbeat

Constant name "heartbeat" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

Usage of the global statement should be avoided.

    log.debug("Handling initial discord hello websocket message.")
    heartbeat = payload.data.get("heartbeat_interval")

    if not heartbeat:
        log.error(

Use lazy % formatting in logging functions

            "No `heartbeat_interval` is present. Has the API changed? "
            f"(payload: {payload})"
        )

        raise HeartbeatError(
            "Discord hello is missing `heartbeat_interval` in payload."
            "Because of this the client can not maintain a connection. "
            "Check logging for more information."
        )

    heartbeat /= 1000
    log.debug(f"Maintaining a connection with heartbeat: {heartbeat}")

Use lazy % formatting in logging functions

    if sequence:
        await socket.send(str(GatewayDispatch(6, sequence, seq=sequence)))
    else:
        await __send_heartbeat(socket)


async def handle_heartbeat(socket: WebSocketClientProtocol, _):
    """
    Handles a heartbeat, which means that it rests and then sends a new
    heartbeat.
    """
    global heartbeat

Usage of the global statement should be avoided.

Constant name "heartbeat" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

    logging.debug(f"Resting heart for {heartbeat}s")

Use lazy % formatting in logging functions

    await sleep(heartbeat)
    await __send_heartbeat(socket)


def update_sequence(seq: int):
    """
    Update the heartbeat sequence.

    :param seq:
        The new heartbeat sequence to be updated with.
    """
    global sequence

Usage of the global statement should be avoided.

Constant name "sequence" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

    log.debug("Updating heartbeat sequence...")
    sequence = seq