Start adding types

twitchdl/commands/download.py

@@ -9,7 +9,7 @@ import tempfile
 
 from os import path
 from pathlib import Path
-from typing import OrderedDict
+from typing import List, Optional, OrderedDict
 from urllib.parse import urlparse, urlencode
 
 from twitchdl import twitch, utils
@@ -137,7 +137,7 @@ def _clip_target_filename(clip, args):
         raise ConsoleError("Invalid key {} used in --output. Supported keys are: {}".format(e, supported))
 
 
-def _get_vod_paths(playlist, start, end):
+def _get_vod_paths(playlist, start: Optional[int], end: Optional[int]) -> List[str]:
     """Extract unique VOD paths for download from playlist."""
     files = []
     vod_start = 0
@@ -157,7 +157,7 @@ def _get_vod_paths(playlist, start, end):
     return files
 
 
-def _crete_temp_dir(base_uri):
+def _crete_temp_dir(base_uri: str) -> str:
     """Create a temp dir to store downloads if it doesn't exist."""
     path = urlparse(base_uri).path.lstrip("/")
     temp_dir = Path(tempfile.gettempdir(), "twitch-dl", path)
@@ -166,11 +166,11 @@ def _crete_temp_dir(base_uri):
 
 
 def download(args):
-    for video in args.videos:
+    for video_id in args.videos:
-        download_one(video, args)
+        download_one(video_id, args)
 
 
-def download_one(video, args):
+def download_one(video: str, args):
     video_id = utils.parse_video_identifier(video)
     if video_id:
         return _download_video(video_id, args)
@@ -227,7 +227,7 @@ def get_clip_authenticated_url(slug, quality):
     return "{}?{}".format(url, query)
 
 
-def _download_clip(slug, args):
+def _download_clip(slug: str, args) -> None:
     print_out("<dim>Looking up clip...</dim>")
     clip = twitch.get_clip(slug)
     game = clip["game"]["name"] if clip["game"] else "Unknown"
@@ -260,7 +260,7 @@ def _download_clip(slug, args):
     print_out("Downloaded: <blue>{}</blue>".format(target))
 
 
-def _download_video(video_id, args):
+def _download_video(video_id, args) -> None:
     if args.start and args.end and args.end <= args.start:
         raise ConsoleError("End time must be greater than start time")
 

twitchdl/console.py

@@ -5,7 +5,7 @@ import sys
 import re
 
 from argparse import ArgumentParser, ArgumentTypeError
-from collections import namedtuple
+from typing import NamedTuple, List, Tuple, Any, Dict
 
 from twitchdl.exceptions import ConsoleError
 from twitchdl.output import print_err
@@ -13,12 +13,19 @@ from twitchdl.twitch import GQLError
 from . import commands, __version__
 
 
-Command = namedtuple("Command", ["name", "description", "arguments"])
+Argument = Tuple[List[str], Dict[str, Any]]
+
+
+class Command(NamedTuple):
+    name: str
+    description: str
+    arguments: List[Argument]
+
 
 CLIENT_WEBSITE = 'https://github.com/ihabunek/twitch-dl'
 
 
-def time(value):
+def time(value: str) -> int:
     """Parse a time string (hh:mm or hh:mm:ss) to number of seconds."""
     parts = [int(p) for p in value.split(":")]
 
@@ -35,19 +42,19 @@ def time(value):
     return hours * 3600 + minutes * 60 + seconds
 
 
-def pos_integer(value):
+def pos_integer(value: str) -> int:
     try:
-        value = int(value)
+        parsed = int(value)
     except ValueError:
         raise ArgumentTypeError("must be an integer")
 
-    if value < 1:
+    if parsed < 1:
         raise ArgumentTypeError("must be positive")
 
-    return value
+    return parsed
 
 
-def rate(value):
+def rate(value: str) -> int:
     match = re.search(r"^([0-9]+)(k|m|)$", value, flags=re.IGNORECASE)
 
     if not match:

twitchdl/download.py

@@ -10,7 +10,7 @@ class DownloadFailed(Exception):
     pass
 
 
-def _download(url, path):
+def _download(url: str, path: str):
     tmp_path = path + ".tmp"
     size = 0
     with httpx.stream("GET", url, timeout=CONNECT_TIMEOUT) as response:
@@ -23,7 +23,7 @@ def _download(url, path):
     return size
 
 
-def download_file(url, path, retries=RETRY_COUNT):
+def download_file(url: str, path: str, retries: int = RETRY_COUNT):
     if os.path.exists(path):
         from_disk = True
         return (os.path.getsize(path), from_disk)

twitchdl/http.py

@@ -55,7 +55,7 @@ class TokenBucket:
 
 class EndlessTokenBucket:
     """Used when download speed is not limited."""
-    def advance(self, size):
+    def advance(self, size: int):
         pass
 
 

twitchdl/output.py

@@ -6,6 +6,7 @@ import re
 
 from itertools import islice
 from twitchdl import utils
+from typing import Any, Match
 
 
 START_CODES = {
@@ -29,26 +30,26 @@ END_PATTERN = "</(" + "|".join(START_CODES.keys()) + ")>"
 USE_ANSI_COLOR = "--no-color" not in sys.argv
 
 
-def start_code(match):
+def start_code(match: Match[str]) -> str:
     name = match.group(1)
     return START_CODES[name]
 
 
-def colorize(text):
+def colorize(text: str) -> str:
     text = re.sub(START_PATTERN, start_code, text)
     text = re.sub(END_PATTERN, END_CODE, text)
 
     return text
 
 
-def strip_tags(text):
+def strip_tags(text: str) -> str:
     text = re.sub(START_PATTERN, '', text)
     text = re.sub(END_PATTERN, '', text)
 
     return text
 
 
-def truncate(string, length):
+def truncate(string: str, length: int) -> str:
     if len(string) > length:
         return string[:length - 1] + "…"
 
@@ -60,7 +61,7 @@ def print_out(*args, **kwargs):
     print(*args, **kwargs)
 
 
-def print_json(data):
+def print_json(data: Any):
     print(json.dumps(data))
 
 

twitchdl/twitch.py

@@ -4,6 +4,7 @@ Twitch API access.
 
 import httpx
 
+from typing import Dict
 from twitchdl import CLIENT_ID
 from twitchdl.exceptions import ConsoleError
 
@@ -37,7 +38,7 @@ def gql_post(query):
     return response
 
 
-def gql_query(query, headers={}):
+def gql_query(query: str, headers: Dict[str, str] = {}):
     url = "https://gql.twitch.tv/gql"
     response = authenticated_post(url, json={"query": query}, headers=headers).json()