pocketutils.misc.query_utils

Complexity

Total Complexity

12

Size/Duplication

Total Lines	90
Duplicated Lines	0 %

Importance

Changes

0

1 Function

Rating	Name	Duplication	Size	Complexity
A	download_urllib()	0	3	2

import logging

Missing module docstring

import random
import time
from collections.abc import ByteString, Callable, Mapping
from dataclasses import dataclass
from datetime import timedelta
from typing import Any
from urllib import request

logger = logging.getLogger("pocketutils")


def download_urllib(req: request.Request) -> bytes:

Missing function or method docstring

    with request.urlopen(req) as q:

Variable name "q" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        return q.read()


@dataclass(frozen=True, repr=True, order=True)

Missing class docstring

class TimeTaken:
    query: timedelta
    wait: timedelta


class QueryExecutor:
    """
    A synchronous GET/POST query executor that limits the rate of requests.
    """

    def __init__(
        self,

Wrong hanging indentation before block (add 4 spaces).

        sec_delay_min: float = 0.25,

Wrong hanging indentation before block (add 4 spaces).

        sec_delay_max: float = 0.25,

Wrong hanging indentation before block (add 4 spaces).

        encoding: str | None = "utf-8",

No space allowed around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

        querier: Callable[[request.Request], ByteString] | None = None,

No space allowed around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

Value 'Callable' is unsubscriptable

    ):
        self._min = sec_delay_min
        self._max = sec_delay_max
        self._rand = random.Random()  # nosec
        self._encoding = encoding
        self._next_at = 0
        self._querier = download_urllib if querier is None else querier
        self._time_taken = None

    @property
    def last_time_taken(self) -> TimeTaken:

Missing function or method docstring

        return self._time_taken

    def __call__(

Too many arguments (6/5)

        self,

Wrong hanging indentation before block (add 4 spaces).

        url: str,

Wrong hanging indentation before block (add 4 spaces).

        method: str = "get",

Wrong hanging indentation before block (add 4 spaces).

        encoding: str | None = "-1",

No space allowed around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

        headers: Mapping[str, str] | None = None,

No space allowed around keyword argument assignment

Wrong hanging indentation before block (add 4 spaces).

Value 'Mapping' is unsubscriptable

        errors: str = "ignore",

Wrong hanging indentation before block (add 4 spaces).

    ) -> str:
        headers = {} if headers is None else headers
        encoding = self._encoding if encoding == "-1" else encoding
        now = time.monotonic()
        wait_secs = self._next_at - now
        if now < self._next_at:
            time.sleep(wait_secs)
        now = time.monotonic()
        req = request.Request(url=url, method=method, headers=headers)
        content = self._querier(req)
        if encoding is None:
            data = content.decode(errors=errors)
        else:
            data = content.decode(encoding=encoding, errors=errors)
        now_ = time.monotonic()
        self._time_taken = TimeTaken(timedelta(seconds=wait_secs), timedelta(seconds=now_ - now))
        self._next_at = now_ + self._rand.uniform(self._min, self._max)
        return data


class QueryMixin:

Missing class docstring

    @property
    def executor(self) -> QueryExecutor:

Missing function or method docstring

        raise NotImplementedError()

    def _query(self, url: str, *, sink: Callable[[str], Any] = logger.debug) -> str:
        data = self.executor(url)
        tt = self.executor.last_time_taken

Variable name "tt" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        wt, qt = tt.wait.total_seconds(), tt.query.total_seconds()

Variable name "wt" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Variable name "qt" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        bts = int(len(data) * 8 / 1024)
        sink(f"Queried {bts} kb from {url} in {qt:.1} s with {wt:.1} s of wait")

Use lazy % formatting in logging functions

        return data


__all__ = ["QueryExecutor", "TimeTaken", "QueryMixin"]