Skip to content

Commit

Permalink
chore(compat): python 3.8 fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
Goldziher committed Jul 30, 2023
1 parent 592c893 commit a38aad7
Show file tree
Hide file tree
Showing 6 changed files with 32 additions and 25 deletions.
5 changes: 3 additions & 2 deletions litestar/_signature/model.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# ruff: noqa: UP006
# ruff: noqa: UP006, UP007
from __future__ import annotations

import re
Expand All @@ -14,6 +14,7 @@
Optional,
Sequence,
Set,
Type,
TypedDict,
Union,
cast,
Expand Down Expand Up @@ -95,7 +96,7 @@ def _deserializer(target_type: Any, value: Any, default_deserializer: Callable[[
class SignatureModel(Struct):
"""Model that represents a function signature that uses a msgspec specific type or types."""

_data_dto: ClassVar[type[AbstractDTO] | None]
_data_dto: ClassVar[Optional[Type[AbstractDTO]]]
_dependency_name_set: ClassVar[Set[str]]
# NOTE: we have to use Set and Dict here because python 3.8 goes haywire if we use 'set' and 'dict'
_fields: ClassVar[Dict[str, FieldDefinition]]
Expand Down
6 changes: 5 additions & 1 deletion litestar/handlers/websocket_handlers/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,11 @@ def __init__(
self._pass_socket = "socket" in parsed_signature.parameters

async def __call__(
self, *args: Any, socket: WebSocket, connection_lifespan_dependencies: dict[str, Any], **kwargs: Any
self,
*args: Any,
socket: WebSocket,
connection_lifespan_dependencies: dict[str, Any],
**kwargs: Any,
) -> None:
lifespan_mananger = self._listener._connection_lifespan or self._listener.default_connection_lifespan

Check notice on line 102 in litestar/handlers/websocket_handlers/_utils.py

View workflow job for this annotation

GitHub Actions / Qodana Community for Python

Accessing a protected member of a class or a module

Access to a protected member _connection_lifespan of a class
handle_send = self._listener.resolve_send_handler() if self._can_send_data else None
Expand Down
6 changes: 4 additions & 2 deletions litestar/handlers/websocket_handlers/listener.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@
Any,
AsyncGenerator,
Callable,
Dict,
Mapping,
Optional,
cast,
overload,
)
Expand Down Expand Up @@ -276,8 +278,8 @@ def signature_model(self) -> type[SignatureModel]:
async def default_connection_lifespan(
self,
socket: WebSocket,
on_accept_dependencies: dict[str, Any] | None = None,
on_disconnect_dependencies: dict[str, Any] | None = None,
on_accept_dependencies: Optional[Dict[str, Any]] = None, # noqa: UP006, UP007
on_disconnect_dependencies: Optional[Dict[str, Any]] = None, # noqa: UP006, UP007
) -> AsyncGenerator[None, None]:
"""Handle the connection lifespan of a :class:`WebSocket <.connection.WebSocket>`.
Expand Down
22 changes: 11 additions & 11 deletions poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 2 additions & 2 deletions tests/unit/test_contrib/test_prometheus.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import time
from http.client import HTTPException
from pathlib import Path
from typing import Any
from typing import Any, Dict

import pytest
from _pytest.monkeypatch import MonkeyPatch
Expand Down Expand Up @@ -182,7 +182,7 @@ def test_prometheus_with_websocket() -> None:
config = create_config()

@websocket_listener("/test")
def test(data: str) -> dict:
def test(data: str) -> Dict[str, str]:
return {"hello": data}

with create_test_client([test, PrometheusController], middleware=[config.middleware]) as client:
Expand Down
14 changes: 7 additions & 7 deletions tests/unit/test_dto/test_factory/test_integration.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
# ruff: noqa: UP007
# ruff: noqa: UP007, UP006
from __future__ import annotations

from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Dict, Generic, Optional, Sequence, TypeVar, cast
from typing import TYPE_CHECKING, Dict, Generic, List, Optional, Sequence, TypeVar, cast
from unittest.mock import MagicMock
from uuid import UUID

Expand Down Expand Up @@ -164,7 +164,7 @@ class NestingBar:

def test_dto_data_injection_with_nested_model() -> None:
@post(dto=DataclassDTO[Annotated[NestingBar, DTOConfig(exclude={"foo.baz"})]], return_dto=None)
def handler(data: DTOData[NestingBar]) -> dict[str, Any]:
def handler(data: DTOData[NestingBar]) -> Dict[str, Any]:
assert isinstance(data, DTOData)
return cast("dict[str, Any]", data.as_builtins())

Expand Down Expand Up @@ -296,7 +296,7 @@ class User:
dto = DataclassDTO[Annotated[User, DTOConfig(partial=True)]]

@post(dto=dto, return_dto=None, signature_namespace={"User": User, "dict": Dict})
def handler(data: DTOData[User] = Body(media_type=RequestEncodingType.URL_ENCODED)) -> dict[str, Any]:
def handler(data: DTOData[User] = Body(media_type=RequestEncodingType.URL_ENCODED)) -> Dict[str, Any]:
return data.as_builtins() # type:ignore[no-any-return]

with create_test_client(route_handlers=[handler]) as client:
Expand Down Expand Up @@ -461,7 +461,7 @@ class Wrapped(Generic[T, V]):

def test_dto_generic_dataclass_wrapped_list_response() -> None:
@get(dto=DataclassDTO[Annotated[PaginatedUser, DTOConfig(exclude={"age"})]])
def handler() -> Wrapped[list[PaginatedUser], int]:
def handler() -> Wrapped[List[PaginatedUser], int]:
return Wrapped(
data=[PaginatedUser(name="John", age=42), PaginatedUser(name="Jane", age=43)],
other=2,
Expand Down Expand Up @@ -490,7 +490,7 @@ def handler() -> Wrapped[PaginatedUser, int]:
@dataclass
class WrappedWithDict(Generic[K, V, T]):
data: T
other: dict[K, V]
other: Dict[K, V]


def test_dto_generic_dataclass_wrapped_scalar_response_with_additional_mapping_data() -> None:
Expand Down Expand Up @@ -518,7 +518,7 @@ def handler() -> Response[PaginatedUser]:

def test_dto_response_wrapped_collection_return_type() -> None:
@get(dto=DataclassDTO[Annotated[PaginatedUser, DTOConfig(exclude={"age"})]])
def handler() -> Response[list[PaginatedUser]]:
def handler() -> Response[List[PaginatedUser]]:
return Response(content=[PaginatedUser(name="John", age=42), PaginatedUser(name="Jane", age=43)])

with create_test_client(handler) as client:
Expand Down

0 comments on commit a38aad7

Please sign in to comment.