Unify file descriptor definitions (#3584)

The _types module can house any common type defintions used throughout
the rest of typeshed to keep defintions in sync.

First candidate is file descriptors where anything with `fileno()`
method is accepted. There were several different implementations in
various files that can be unified.
This commit is contained in:
Daniel Farley
2020-01-08 17:25:36 -08:00
committed by Rebecca Chen
parent 1651348a08
commit 955e9c7da4
10 changed files with 84 additions and 86 deletions

View File

@@ -1,4 +1,3 @@
import selectors
from socket import socket, _Address, _RetAddress
import ssl
import sys
@@ -9,6 +8,7 @@ from asyncio.events import AbstractEventLoop, AbstractServer, Handle, TimerHandl
from asyncio.protocols import BaseProtocol
from asyncio.tasks import Task
from asyncio.transports import BaseTransport
from _types import FileDescriptorLike
if sys.version_info >= (3, 7):
from contextvars import Context
@@ -183,10 +183,10 @@ class BaseEventLoop(AbstractEventLoop, metaclass=ABCMeta):
async def subprocess_exec(self, protocol_factory: _ProtocolFactory, *args: Any, stdin: Any = ...,
stdout: Any = ..., stderr: Any = ...,
**kwargs: Any) -> _TransProtPair: ...
def add_reader(self, fd: selectors._FileObject, callback: Callable[..., Any], *args: Any) -> None: ...
def remove_reader(self, fd: selectors._FileObject) -> None: ...
def add_writer(self, fd: selectors._FileObject, callback: Callable[..., Any], *args: Any) -> None: ...
def remove_writer(self, fd: selectors._FileObject) -> None: ...
def add_reader(self, fd: FileDescriptorLike, callback: Callable[..., Any], *args: Any) -> None: ...
def remove_reader(self, fd: FileDescriptorLike) -> None: ...
def add_writer(self, fd: FileDescriptorLike, callback: Callable[..., Any], *args: Any) -> None: ...
def remove_writer(self, fd: FileDescriptorLike) -> None: ...
# Completion based I/O methods returning Futures prior to 3.7
if sys.version_info >= (3, 7):
async def sock_recv(self, sock: socket, nbytes: int) -> bytes: ...

View File

@@ -1,4 +1,3 @@
import selectors
from socket import socket, _Address, _RetAddress
import ssl
import sys
@@ -8,6 +7,7 @@ from asyncio.futures import Future
from asyncio.protocols import BaseProtocol
from asyncio.tasks import Task
from asyncio.transports import BaseTransport
from _types import FileDescriptorLike
_T = TypeVar('_T')
_Context = Dict[str, Any]
@@ -253,13 +253,13 @@ class AbstractEventLoop(metaclass=ABCMeta):
stdout: Any = ..., stderr: Any = ...,
**kwargs: Any) -> _TransProtPair: ...
@abstractmethod
def add_reader(self, fd: selectors._FileObject, callback: Callable[..., Any], *args: Any) -> None: ...
def add_reader(self, fd: FileDescriptorLike, callback: Callable[..., Any], *args: Any) -> None: ...
@abstractmethod
def remove_reader(self, fd: selectors._FileObject) -> None: ...
def remove_reader(self, fd: FileDescriptorLike) -> None: ...
@abstractmethod
def add_writer(self, fd: selectors._FileObject, callback: Callable[..., Any], *args: Any) -> None: ...
def add_writer(self, fd: FileDescriptorLike, callback: Callable[..., Any], *args: Any) -> None: ...
@abstractmethod
def remove_writer(self, fd: selectors._FileObject) -> None: ...
def remove_writer(self, fd: FileDescriptorLike) -> None: ...
# Completion based I/O methods returning Futures prior to 3.7
if sys.version_info >= (3, 7):
@abstractmethod

View File

@@ -1,17 +1,14 @@
import io
import sys
from typing import Union, Protocol
class _HasFileno(Protocol):
def fileno(self) -> int: ...
_File = Union[_HasFileno, int]
from _types import FileDescriptorLike
def cancel_dump_traceback_later() -> None: ...
def disable() -> None: ...
def dump_traceback(file: _File = ..., all_threads: bool = ...) -> None: ...
def dump_traceback_later(timeout: float, repeat: bool = ..., file: _File = ..., exit: bool = ...) -> None: ...
def enable(file: _File = ..., all_threads: bool = ...) -> None: ...
def dump_traceback(file: FileDescriptorLike = ..., all_threads: bool = ...) -> None: ...
def dump_traceback_later(timeout: float, repeat: bool = ..., file: FileDescriptorLike = ..., exit: bool = ...) -> None: ...
def enable(file: FileDescriptorLike = ..., all_threads: bool = ...) -> None: ...
def is_enabled() -> bool: ...
if sys.platform != "win32":
def register(signum: int, file: _File = ..., all_threads: bool = ..., chain: bool = ...) -> None: ...
def register(signum: int, file: FileDescriptorLike = ..., all_threads: bool = ..., chain: bool = ...) -> None: ...
def unregister(signum: int) -> None: ...

View File

@@ -1,6 +1,7 @@
# Stubs for fcntl
from io import IOBase
from typing import Any, IO, Union
from _types import FileDescriptorLike
FASYNC: int
FD_CLOEXEC: int
@@ -75,21 +76,19 @@ LOCK_SH: int
LOCK_UN: int
LOCK_WRITE: int
_AnyFile = Union[int, IO[Any], IOBase]
# TODO All these return either int or bytes depending on the value of
# cmd (not on the type of arg).
def fcntl(fd: _AnyFile,
def fcntl(fd: FileDescriptorLike,
cmd: int,
arg: Union[int, bytes] = ...) -> Any: ...
# TODO This function accepts any object supporting a buffer interface,
# as arg, is there a better way to express this than bytes?
def ioctl(fd: _AnyFile,
def ioctl(fd: FileDescriptorLike,
request: int,
arg: Union[int, bytes] = ...,
mutate_flag: bool = ...) -> Any: ...
def flock(fd: _AnyFile, operation: int) -> None: ...
def lockf(fd: _AnyFile,
def flock(fd: FileDescriptorLike, operation: int) -> None: ...
def lockf(fd: FileDescriptorLike,
cmd: int,
len: int = ...,
start: int = ...,

View File

@@ -3,82 +3,77 @@
from abc import ABCMeta, abstractmethod
from typing import Any, List, Mapping, NamedTuple, Optional, Protocol, Tuple, Union
from _types import FileDescriptor, FileDescriptorLike
class _HasFileno(Protocol):
def fileno(self) -> int: ...
# Type aliases added mainly to preserve some context
_FileObject = Union[int, _HasFileno]
_FileDescriptor = int
_EventMask = int
EVENT_READ: _EventMask
EVENT_WRITE: _EventMask
class SelectorKey(NamedTuple):
fileobj: _FileObject
fd: _FileDescriptor
fileobj: FileDescriptorLike
fd: FileDescriptor
events: _EventMask
data: Any
class BaseSelector(metaclass=ABCMeta):
@abstractmethod
def register(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def register(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
@abstractmethod
def unregister(self, fileobj: _FileObject) -> SelectorKey: ...
def unregister(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
def modify(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def modify(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
@abstractmethod
def select(self, timeout: Optional[float] = ...) -> List[Tuple[SelectorKey, _EventMask]]: ...
def close(self) -> None: ...
def get_key(self, fileobj: _FileObject) -> SelectorKey: ...
def get_key(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
@abstractmethod
def get_map(self) -> Mapping[_FileObject, SelectorKey]: ...
def get_map(self) -> Mapping[FileDescriptorLike, SelectorKey]: ...
def __enter__(self) -> BaseSelector: ...
def __exit__(self, *args: Any) -> None: ...
class SelectSelector(BaseSelector):
def register(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: _FileObject) -> SelectorKey: ...
def register(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
def select(self, timeout: Optional[float] = ...) -> List[Tuple[SelectorKey, _EventMask]]: ...
def get_map(self) -> Mapping[_FileObject, SelectorKey]: ...
def get_map(self) -> Mapping[FileDescriptorLike, SelectorKey]: ...
class PollSelector(BaseSelector):
def register(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: _FileObject) -> SelectorKey: ...
def register(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
def select(self, timeout: Optional[float] = ...) -> List[Tuple[SelectorKey, _EventMask]]: ...
def get_map(self) -> Mapping[_FileObject, SelectorKey]: ...
def get_map(self) -> Mapping[FileDescriptorLike, SelectorKey]: ...
class EpollSelector(BaseSelector):
def fileno(self) -> int: ...
def register(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: _FileObject) -> SelectorKey: ...
def register(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
def select(self, timeout: Optional[float] = ...) -> List[Tuple[SelectorKey, _EventMask]]: ...
def get_map(self) -> Mapping[_FileObject, SelectorKey]: ...
def get_map(self) -> Mapping[FileDescriptorLike, SelectorKey]: ...
class DevpollSelector(BaseSelector):
def fileno(self) -> int: ...
def register(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: _FileObject) -> SelectorKey: ...
def register(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
def select(self, timeout: Optional[float] = ...) -> List[Tuple[SelectorKey, _EventMask]]: ...
def get_map(self) -> Mapping[_FileObject, SelectorKey]: ...
def get_map(self) -> Mapping[FileDescriptorLike, SelectorKey]: ...
class KqueueSelector(BaseSelector):
def fileno(self) -> int: ...
def register(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: _FileObject) -> SelectorKey: ...
def register(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
def select(self, timeout: Optional[float] = ...) -> List[Tuple[SelectorKey, _EventMask]]: ...
def get_map(self) -> Mapping[_FileObject, SelectorKey]: ...
def get_map(self) -> Mapping[FileDescriptorLike, SelectorKey]: ...
class DefaultSelector(BaseSelector):
def register(self, fileobj: _FileObject, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: _FileObject) -> SelectorKey: ...
def register(self, fileobj: FileDescriptorLike, events: _EventMask, data: Any = ...) -> SelectorKey: ...
def unregister(self, fileobj: FileDescriptorLike) -> SelectorKey: ...
def select(self, timeout: Optional[float] = ...) -> List[Tuple[SelectorKey, _EventMask]]: ...
def get_map(self) -> Mapping[_FileObject, SelectorKey]: ...
def get_map(self) -> Mapping[FileDescriptorLike, SelectorKey]: ...