# pylint: skip-file
# flake8: noqa
+
def run():
# throws nice syntax error on old Python versions:
- 0_0 # Python >= 3.6 required
+ 0_0 # Python >= 3.7 required
from knot_resolver_manager import main
+
main.main()
return {}
def run(self, args: CommandArgs) -> None:
-
with open(self.input_file, "r") as f:
data = f.read()
return {}
def run(self, args: CommandArgs) -> None:
-
if self.input_file:
with open(self.input_file, "r") as f:
data = f.read()
import argparse
-from abc import ABC, abstractmethod
+from abc import ABC, abstractmethod # pylint: disable=[no-name-in-module]
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Type, TypeVar
from urllib.parse import quote
raise NotImplementedError()
@abstractmethod
- def __init__(self, namespace: argparse.Namespace) -> None:
+ def __init__(self, namespace: argparse.Namespace) -> None: # pylint: disable=[unused-argument]
super().__init__()
@abstractmethod
# version 3.8 and higher, call directly
if sys.version_info.major >= 3 and sys.version_info.minor >= 8:
# pylint: disable=unexpected-keyword-arg
- return asyncio.create_task(coro, name=name) # type: ignore[attr-defined]
+ return asyncio.create_task(coro, name=name) # type: ignore[attr-defined,arg-type,call-arg]
# version 3.7 and higher, call directly without the name argument
if sys.version_info.major >= 3 and sys.version_info.minor >= 8:
- return asyncio.create_task(coro) # type: ignore[attr-defined]
+ return asyncio.create_task(coro) # type: ignore[attr-defined,arg-type]
# earlier versions, use older function
else:
# version 3.7 and higher, call directly
# disabled due to incompatibilities
if sys.version_info.major >= 3 and sys.version_info.minor >= 7:
- return asyncio.run(coro, debug=debug) # type: ignore[attr-defined]
+ return asyncio.run(coro, debug=debug) # type: ignore[attr-defined,arg-type]
# earlier versions, use backported version of the function
if events._get_running_loop() is not None: # pylint: disable=protected-access
def __init__(
self, source_value: Any, parents: Tuple["UncheckedPath", ...] = tuple(), object_path: str = "/"
) -> None:
-
super().__init__(source_value, object_path=object_path)
self._object_path: str = object_path
self._parents: Tuple[UncheckedPath, ...] = parents
if isinstance(source_value, str):
-
# we do not load global validation context if the path is absolute
# this prevents errors when constructing defaults in the schema
if source_value.startswith("/"):
class SizeUnit(UnitBase):
- _units = {"B": 1, "K": 1024, "M": 1024 ** 2, "G": 1024 ** 3}
+ _units = {"B": 1, "K": 1024, "M": 1024**2, "G": 1024**3}
def bytes(self) -> int:
return self._value
import logging
-from os import kill
+from os import kill # pylint: disable=[no-name-in-module]
from pathlib import Path
from typing import Any, Dict, Iterable, NoReturn, Optional, Union, cast
from xmlrpc.client import Fault, ServerProxy
return Result.ok(None)
async def _reload_config(self) -> None:
-
if self._config_path is None:
logger.warning("The manager was started with inlined configuration - can't reload")
else:
from prometheus_client import Histogram, exposition # type: ignore
from prometheus_client.bridge.graphite import GraphiteBridge # type: ignore
-from prometheus_client.core import ( # type: ignore
- REGISTRY,
- CounterMetricFamily,
- GaugeMetricFamily,
- HistogramMetricFamily,
- Metric,
-)
+from prometheus_client.core import GaugeMetricFamily # type: ignore
+from prometheus_client.core import REGISTRY, CounterMetricFamily, HistogramMetricFamily, Metric
from knot_resolver_manager import compat
from knot_resolver_manager.config_store import ConfigStore, only_on_real_changes
import enum
import inspect
-from abc import ABC, abstractmethod
+from abc import ABC, abstractmethod # pylint: disable=[no-name-in-module]
from typing import Any, Callable, Dict, Generic, List, Optional, Set, Tuple, Type, TypeVar, Union, cast
import yaml
"""
def __init__(self, constructor: Callable[..., T], *args: Any, **kwargs: Any) -> None:
+ # pylint: disable=[super-init-not-called]
self._func = constructor
self._args = args
self._kwargs = kwargs
_LAYER: Optional[Type["BaseSchema"]] = None
_MAPPER: ObjectMapper = ObjectMapper()
- def __init__(self, source: TSource = None, object_path: str = ""):
+ def __init__(self, source: TSource = None, object_path: str = ""): # pylint: disable=[super-init-not-called]
# save source data (and drop information about nullness)
source = source or {}
self.__source: Union[Dict[str, Any], BaseSchema] = source
-from abc import ABC, abstractmethod
+from abc import ABC, abstractmethod # pylint: disable=[no-name-in-module]
from typing import Any, Dict, Type
import copy
-from abc import ABC, abstractmethod
+from abc import ABC, abstractmethod # pylint: disable=[no-name-in-module]
from typing import Any, List, Optional, Tuple, Union
from typing_extensions import Literal
assert l = rl.original()
"""
-from abc import ABC, abstractmethod
+from abc import ABC, abstractmethod # pylint: disable=[no-name-in-module]
from typing import Any, Dict, List, TypeVar
def is_union(tp: Any) -> bool:
- """ Returns true even for optional types, because they are just a Union[T, NoneType] """
+ """Returns true even for optional types, because they are just a Union[T, NoneType]"""
return getattr(tp, "__origin__", None) == Union # type: ignore
def get_generic_type_argument(tp: Any) -> Any:
- """ same as function get_generic_type_arguments, but expects just one type argument"""
+ """same as function get_generic_type_arguments, but expects just one type argument"""
args = get_generic_type_arguments(tp)
assert len(args) == 1
super().__init__()
def open_(self: UnixHTTPHandler, req: Any) -> Any:
- return self.do_open(UnixHTTPConnection, req)
+ return self.do_open(UnixHTTPConnection, req) # type: ignore[arg-type]
setattr(UnixHTTPHandler, "http+unix_open", open_)
setattr(UnixHTTPHandler, "http+unix_request", AbstractHTTPHandler.do_request_)
projects / thirdparty / knot-resolver.git / commitdiff
