Skip to content

Commit

Permalink
Simplify with type aliases
Browse files Browse the repository at this point in the history
  • Loading branch information
@A5rocks
A5rocks committed Feb 16, 2025
1 parent ccfafe4 commit 09fe32f
Showing 1 changed file with 40 additions and 117 deletions.
157 changes: 40 additions & 117 deletions src/trio/_subprocess.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1085,7 +1085,7 @@ async def my_deliver_cancel(process):
# overloads. But might still be a problem for other static analyzers / docstring
# readers (?)

class UnixProcessArgs(GeneralProcessArgs, total=False):
class UnixProcessArgs3_9(GeneralProcessArgs, total=False):
preexec_fn: Callable[[], object] | None
restore_signals: bool
start_new_session: bool
Expand All @@ -1097,7 +1097,7 @@ class UnixProcessArgs(GeneralProcessArgs, total=False):
user: str | int | None
umask: int

class UnixProcessArgs3_10(UnixProcessArgs, total=False):
class UnixProcessArgs3_10(UnixProcessArgs3_9, total=False):
pipesize: int

class UnixProcessArgs3_11(UnixProcessArgs3_10, total=False):
Expand All @@ -1111,127 +1111,50 @@ class UnixRunProcessMixin(TypedDict, total=False):
deliver_cancel: Callable[[Process], Awaitable[None]] | None

if sys.version_info >= (3, 11):

@overload # type: ignore[no-overload-impl]
async def open_process(
command: StrOrBytesPath,
*,
stdin: int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixProcessArgs3_11],
) -> trio.Process: ...

@overload
async def open_process(
command: Sequence[StrOrBytesPath],
*,
stdin: int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixProcessArgs3_11],
) -> trio.Process: ...

class UnixRunProcessArgs(UnixProcessArgs3_11, UnixRunProcessMixin):
pass

@overload # type: ignore[no-overload-impl]
async def run_process(
command: StrOrBytesPath,
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...

@overload
async def run_process(
command: Sequence[StrOrBytesPath],
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...

UnixProcessArgs = UnixProcessArgs3_11
elif sys.version_info >= (3, 10):
UnixProcessArgs = UnixProcessArgs3_10
else:
UnixProcessArgs = UnixProcessArgs3_9

@overload # type: ignore[no-overload-impl]
async def open_process(
command: StrOrBytesPath,
*,
stdin: int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixProcessArgs3_10],
) -> trio.Process: ...

@overload
async def open_process(
command: Sequence[StrOrBytesPath],
*,
stdin: int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixProcessArgs3_10],
) -> trio.Process: ...

class UnixRunProcessArgs(UnixProcessArgs3_10, UnixRunProcessMixin):
pass
@overload # type: ignore[no-overload-impl]
async def open_process(
command: StrOrBytesPath,
*,
stdin: int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixProcessArgs],
) -> trio.Process: ...

@overload # type: ignore[no-overload-impl]
async def run_process(
command: StrOrBytesPath,
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...

@overload
async def run_process(
command: Sequence[StrOrBytesPath],
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...
@overload
async def open_process(
command: Sequence[StrOrBytesPath],
*,
stdin: int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixProcessArgs],
) -> trio.Process: ...

else:
class UnixRunProcessArgs(UnixProcessArgs, UnixRunProcessMixin):

Check failure on line 1138 in src/trio/_subprocess.py

View workflow job for this annotation

GitHub Actions / Ubuntu (3.13, check formatting)

Mypy-Linux+Mac 

src/trio/_subprocess.py:(1138:9 - 1138:9): All bases of a new TypedDict must be TypedDict types  [misc]
pass

@overload # type: ignore[no-overload-impl]
async def open_process(
command: StrOrBytesPath,
*,
stdin: int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixProcessArgs],
) -> trio.Process: ...

@overload
async def open_process(
command: Sequence[StrOrBytesPath],
*,
stdin: int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixProcessArgs],
) -> trio.Process: ...

class UnixRunProcessArgs(UnixProcessArgs, UnixRunProcessMixin):
pass
@overload # type: ignore[no-overload-impl]
async def run_process(
command: StrOrBytesPath,
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...

@overload # type: ignore[no-overload-impl]
async def run_process(
command: StrOrBytesPath,
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: Literal[True],
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...

@overload
async def run_process(
command: Sequence[StrOrBytesPath],
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...
@overload
async def run_process(
command: Sequence[StrOrBytesPath],
*,
stdin: bytes | bytearray | memoryview | int | HasFileno | None = None,
shell: bool = False,
**kwargs: Unpack[UnixRunProcessArgs],
) -> subprocess.CompletedProcess[bytes]: ...

else:
# At runtime, use the actual implementations.
Expand Down

0 comments on commit 09fe32f

Please sign in to comment.