~sgeisenh/streambot

3797850e049f6e71530805d1bd64f3893e750a39 — Samuel Eisenhandler 1 year, 5 months ago 1d694ae main
Types and lints
2 files changed, 20 insertions(+), 9 deletions(-)

M streambot.py
A tox.ini
M streambot.py => streambot.py +18 -9
@@ 2,7 2,7 @@ import logging
import os
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, Final, List, Optional, Tuple
from typing import Any, cast, Dict, Final, List, Optional, Protocol, Tuple

import requests



@@ 45,10 45,15 @@ def get_oauth_token() -> str:
USERS_URL: Final[str] = "https://api.twitch.tv/helix/users"


def get_channel_id(channel_name) -> str:
def get_channel_id(channel_name: str) -> str:
    payload = {"login": channel_name}
    response = requests.get(USERS_URL, params=payload, headers=create_headers()).json()
    return response["data"][0]["id"]
    return cast(str, response["data"][0]["id"])


class BotHandler(Protocol):
    def send_reply(self, message: Dict[str, Any], reply: str) -> None:
        ...


class StreamBotHandler:


@@ 69,10 74,10 @@ class StreamBotHandler:
            @streambot add sgeisenh
            """

    def handle_message(self, message, bot_handler) -> None:
    def handle_message(self, message: Dict[str, Any], bot_handler: BotHandler) -> None:
        logger.info("received message: %s", message)

        def handle_error(error):
        def handle_error(error: str) -> None:
            bot_handler.send_reply(message, f"@streambot Error: {error}")

        content = message["content"]


@@ 97,7 102,9 @@ class StreamBotHandler:
            )


def handle_add(message, channel_id, channel_name, bot_handler):
def handle_add(
    message: Dict[str, Any], channel_id: str, channel_name: str, bot_handler: BotHandler
) -> None:
    data = {
        "type": "stream.online",
        "version": "1",


@@ 141,7 148,9 @@ def handle_add(message, channel_id, channel_name, bot_handler):
    )


def handle_remove(message, channel_id, channel_name, bot_handler):
def handle_remove(
    message: Dict[str, Any], channel_id: str, channel_name: str, bot_handler: BotHandler
) -> None:
    subscription_id = get_subscription_id(channel_id)
    if subscription_id is None:
        delete_channel(channel_id)


@@ 164,7 173,7 @@ def handle_remove(message, channel_id, channel_name, bot_handler):
    bot_handler.send_reply(message, f"no longer subscribed to '{channel_name}'")


def handle_list(message, bot_handler):
def handle_list(message: Dict[str, Any], bot_handler: BotHandler) -> None:
    channel_list = "\n".join(f"- {channel}" for channel in list_channels())
    bot_handler.send_reply(
        message, f"List of current channel subscriptions:\n{channel_list}"


@@ 202,7 211,7 @@ ON CONFLICT(channel_id) DO UPDATE SET
"""


def insert_channel(channel_id: str, channel_name: str, subscription_id: str):
def insert_channel(channel_id: str, channel_name: str, subscription_id: str) -> None:
    con = get_db()
    con.execute(
        INSERT_CHANNEL,

A tox.ini => tox.ini +2 -0
@@ 0,0 1,2 @@
[flake8]
max-line-length = 100