mirrors/pferd
1
0
Fork 0
mirror of https://github.com/Garmelon/PFERD.git synced 2023-12-21 10:23:01 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fix mypy errors

Browse Source
This commit is contained in:
Joscha 2021-04-29 15:47:52 +02:00
parent d2103d7c44
commit 502654d853
2 changed files with 6 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
PFERD/crawler.py
View File

@ -49,7 +49,7 @@ class Crawler(ABC):
self._conductor.print(text)
def exclusive_output(self):
def exclusive_output(self) -> AsyncContextManager[None]:
"""
Acquire exclusive rights™ to the terminal output. While this context
manager is held, output such as printing and progress bars from other

10
PFERD/utils.py
View File

@ -1,6 +1,6 @@
import functools
import contextvars
import asyncio
import contextvars
import functools
import getpass
from typing import Any, Callable, Optional, TypeVar
@ -13,14 +13,14 @@ async def to_thread(func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
loop = asyncio.get_event_loop()
ctx = contextvars.copy_context()
func_call = functools.partial(ctx.run, func, *args, **kwargs)
return await loop.run_in_executor(None, func_call)
return await loop.run_in_executor(None, func_call) # type: ignore
async def ainput(prompt: Optional[str] = None) -> str:
async def ainput(prompt: str) -> str:
return await to_thread(lambda: input(prompt))
async def agetpass(prompt: Optional[str] = None) -> str:
async def agetpass(prompt: str) -> str:
return await to_thread(lambda: getpass.getpass(prompt))