Skip to content

Commit

Permalink
Clean-up existing typing
Browse files Browse the repository at this point in the history
Put unnecessary run-time imports in to TYPE_CHECKING if.
Only use a single T TypeVar.
Make `new_proxy` not require typing assert with the bounded TypeVar.
  • Loading branch information
igo95862 committed Nov 26, 2023
1 parent 838a373 commit effe5d8
Show file tree
Hide file tree
Showing 19 changed files with 206 additions and 195 deletions.
8 changes: 6 additions & 2 deletions src/sdbus/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,15 +22,19 @@
from argparse import ArgumentParser
from pathlib import Path
from sys import stdout
from typing import List, Optional
from typing import TYPE_CHECKING

from .interface_generator import (
DbusInterfaceIntrospection,
generate_py_file,
interfaces_from_file,
interfaces_from_str,
)

if TYPE_CHECKING:
from typing import List, Optional

from .interface_generator import DbusInterfaceIntrospection


def run_gen_from_connection(
connection_name: str,
Expand Down
9 changes: 6 additions & 3 deletions src/sdbus/autodoc.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,10 @@
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

from __future__ import annotations

from typing import Any, Dict
from typing import TYPE_CHECKING

from sphinx.application import Sphinx
from sphinx.ext.autodoc import AttributeDocumenter, MethodDocumenter

from .dbus_proxy_async_method import DbusMethodAsyncBinded
Expand All @@ -32,6 +30,11 @@
)
from .dbus_proxy_async_signal import DbusSignalAsync, DbusSignalBinded

if TYPE_CHECKING:
from typing import Any, Dict

from sphinx.application import Sphinx


class DbusMethodDocumenter(MethodDocumenter):

Expand Down
30 changes: 16 additions & 14 deletions src/sdbus/dbus_common_elements.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,24 +20,29 @@
from __future__ import annotations

from inspect import getfullargspec
from types import FunctionType
from typing import (
Any,
Callable,
Dict,
List,
Optional,
Sequence,
Tuple,
TypeVar,
)
from typing import TYPE_CHECKING

from .dbus_common_funcs import (
_is_property_flags_correct,
_method_name_converter,
)
from .sd_bus_internals import is_interface_name_valid, is_member_name_valid

if TYPE_CHECKING:
from types import FunctionType
from typing import (
Any,
Callable,
Dict,
List,
Optional,
Sequence,
Tuple,
TypeVar,
)

T = TypeVar('T')


class DbusSomethingCommon:
def __init__(self) -> None:
Expand Down Expand Up @@ -269,9 +274,6 @@ class DbusBindedSync:
...


T = TypeVar('T')


class DbusOverload:
def __init__(self, original: T):
self.original = original
Expand Down
8 changes: 6 additions & 2 deletions src/sdbus/dbus_common_funcs.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@

from asyncio import Future, get_running_loop
from contextvars import ContextVar
from typing import Any, Dict, Generator, Iterator, Literal, Tuple
from typing import TYPE_CHECKING
from warnings import warn

from .sd_bus_internals import (
Expand All @@ -33,10 +33,14 @@
NameAllowReplacementFlag,
NameQueueFlag,
NameReplaceExistingFlag,
SdBus,
sd_bus_open,
)

if TYPE_CHECKING:
from typing import Any, Dict, Generator, Iterator, Literal, Tuple

from .sd_bus_internals import SdBus

DEFAULT_BUS: ContextVar[SdBus] = ContextVar('DEFAULT_BUS')

PROPERTY_FLAGS_MASK = (
Expand Down
20 changes: 14 additions & 6 deletions src/sdbus/dbus_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,30 +19,38 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
from __future__ import annotations

from typing import Any, Dict, Tuple, cast
from typing import TYPE_CHECKING

from .sd_bus_internals import (
SdBusBaseError,
add_exception_mapping,
map_exception_to_dbus_error,
)

if TYPE_CHECKING:
from typing import Any, Dict, Tuple


class DbusErrorMeta(type):

def __new__(cls, name: str,
bases: Tuple[type, ...],
namespace: Dict[str, Any],
) -> DbusErrorMeta:
def __new__(
cls,
name: str,
bases: Tuple[type, ...],
namespace: Dict[str, Any],
) -> DbusErrorMeta:

dbus_error_name = namespace.get('dbus_error_name')

if dbus_error_name is None:
raise TypeError('D-Bus error name not passed')

new_cls = super().__new__(cls, name, bases, namespace)
assert issubclass(new_cls, Exception), (
f"New class {new_cls} is not an Exception but {bases}."
)

add_exception_mapping(cast(Exception, new_cls))
add_exception_mapping(new_cls)

return new_cls

Expand Down
38 changes: 14 additions & 24 deletions src/sdbus/dbus_proxy_async_interface_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,27 +19,14 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
from __future__ import annotations

from asyncio import Queue
from copy import deepcopy
from inspect import getmembers
from types import MethodType
from typing import (
Any,
Callable,
Dict,
List,
Optional,
Set,
Tuple,
Type,
TypeVar,
cast,
)
from typing import TYPE_CHECKING, Any, Callable, cast
from warnings import warn
from weakref import ref as weak_ref

from .dbus_common_elements import (
DbusBindedAsync,
DbusInterfaceMetaCommon,
DbusOverload,
DbusSomethingAsync,
Expand All @@ -52,9 +39,16 @@
DbusPropertyAsyncBinded,
)
from .dbus_proxy_async_signal import DbusSignalAsync, DbusSignalBinded
from .sd_bus_internals import SdBus, SdBusInterface
from .sd_bus_internals import SdBusInterface

if TYPE_CHECKING:
from asyncio import Queue
from typing import Dict, List, Optional, Set, Tuple, Type, TypeVar

from .dbus_common_elements import DbusBindedAsync
from .sd_bus_internals import SdBus

T_input = TypeVar('T_input')
Self = TypeVar('Self', bound="DbusInterfaceBaseAsync")


class DbusInterfaceMetaAsync(DbusInterfaceMetaCommon):
Expand Down Expand Up @@ -296,32 +290,28 @@ def _proxify(

@classmethod
def new_connect(
cls: Type[T_input],
cls: Type[Self],
service_name: str,
object_path: str,
bus: Optional[SdBus] = None,
) -> T_input:
) -> Self:
warn(
("new_connect is deprecated in favor of equivalent new_proxy."
"Will be removed in version 1.0.0"),
DeprecationWarning,
)
new_object = cls.__new__(cls)
assert isinstance(new_object, DbusInterfaceBaseAsync)
new_object._proxify(service_name, object_path, bus)
assert isinstance(new_object, cls)
return new_object

@classmethod
def new_proxy(
cls: Type[T_input],
cls: Type[Self],
service_name: str,
object_path: str,
bus: Optional[SdBus] = None,
) -> T_input:
) -> Self:

new_object = cls.__new__(cls)
assert isinstance(new_object, DbusInterfaceBaseAsync)
new_object._proxify(service_name, object_path, bus)
assert isinstance(new_object, cls)
return new_object
21 changes: 14 additions & 7 deletions src/sdbus/dbus_proxy_async_interfaces.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,25 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
from __future__ import annotations

from typing import Any, Dict, List, Literal, Optional, Tuple
from typing import TYPE_CHECKING

from .dbus_common_funcs import _parse_properties_vardict, get_default_bus
from .dbus_proxy_async_interface_base import DbusInterfaceBaseAsync
from .dbus_proxy_async_method import dbus_method_async
from .dbus_proxy_async_signal import dbus_signal_async
from .sd_bus_internals import SdBus, SdBusSlot

if TYPE_CHECKING:
from typing import Any, Dict, List, Literal, Optional, Tuple

from .sd_bus_internals import SdBus, SdBusSlot

DBUS_PROPERTIES_CHANGED_TYPING = (
Tuple[
str,
Dict[str, Tuple[str, Any]],
List[str],
]
)


class DbusPeerInterfaceAsync(
Expand Down Expand Up @@ -54,11 +66,6 @@ async def dbus_introspect(self) -> str:
raise NotImplementedError


DBUS_PROPERTIES_CHANGED_TYPING = Tuple[str,
Dict[str, Tuple[str, Any]],
List[str]]


class DbusPropertiesInterfaceAsync(
DbusInterfaceBaseAsync,
interface_name='org.freedesktop.DBus.Properties',
Expand Down
37 changes: 15 additions & 22 deletions src/sdbus/dbus_proxy_async_method.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,16 +22,7 @@
from contextvars import ContextVar, copy_context
from inspect import iscoroutinefunction
from types import FunctionType
from typing import (
TYPE_CHECKING,
Any,
Callable,
Optional,
Sequence,
Type,
TypeVar,
cast,
)
from typing import TYPE_CHECKING, cast
from weakref import ref as weak_ref

from .dbus_common_elements import (
Expand All @@ -41,21 +32,23 @@
DbusSomethingAsync,
)
from .dbus_exceptions import DbusFailedError
from .sd_bus_internals import DbusNoReplyFlag, SdBusMessage

CURRENT_MESSAGE: ContextVar[SdBusMessage] = ContextVar('CURRENT_MESSAGE')
from .sd_bus_internals import DbusNoReplyFlag

if TYPE_CHECKING:
from typing import Any, Callable, Optional, Sequence, Type, TypeVar

def get_current_message() -> SdBusMessage:
return CURRENT_MESSAGE.get()
from .dbus_proxy_async_interface_base import DbusInterfaceBaseAsync
from .sd_bus_internals import SdBusMessage

T = TypeVar('T')
else:
T = None

T_input = TypeVar('T_input')
T = TypeVar('T')
CURRENT_MESSAGE: ContextVar[SdBusMessage] = ContextVar('CURRENT_MESSAGE')


if TYPE_CHECKING:
from .dbus_proxy_async_interface_base import DbusInterfaceBaseAsync
def get_current_message() -> SdBusMessage:
return CURRENT_MESSAGE.get()


class DbusMethodAsync(DbusMethodCommon, DbusSomethingAsync):
Expand Down Expand Up @@ -212,14 +205,14 @@ def dbus_method_async(
result_args_names: Sequence[str] = (),
input_args_names: Sequence[str] = (),
method_name: Optional[str] = None,
) -> Callable[[T_input], T_input]:
) -> Callable[[T], T]:

assert not isinstance(input_signature, FunctionType), (
"Passed function to decorator directly. "
"Did you forget () round brackets?"
)

def dbus_method_decorator(original_method: T_input) -> T_input:
def dbus_method_decorator(original_method: T) -> T:
assert isinstance(original_method, FunctionType)
assert iscoroutinefunction(original_method), (
"Expected coroutine function. ",
Expand All @@ -235,7 +228,7 @@ def dbus_method_decorator(original_method: T_input) -> T_input:
flags=flags,
)

return cast(T_input, new_wrapper)
return cast(T, new_wrapper)

return dbus_method_decorator

Expand Down

0 comments on commit effe5d8

Please sign in to comment.