From typing import x (#12)

Author: janbjorge

src/pgcachewatch/models.py

@@ -1,15 +1,15 @@
 import datetime
-import typing
+from typing import Literal, NewType
 
 import pydantic
 
-OPERATIONS = typing.Literal[
+OPERATIONS = Literal[
     "insert",
     "update",
     "delete",
 ]
 
-PGChannel = typing.NewType(
+PGChannel = NewType(
     "PGChannel",
     str,
 )

src/pgcachewatch/strategies.py

@@ -1,11 +1,11 @@
 import collections
 import datetime
-import typing
+from typing import Callable, Protocol
 
 from . import listeners, models, utils
 
 
-class Strategy(typing.Protocol):
+class Strategy(Protocol):
     """
     A protocol defining the clear method for different strategies.
     """
@@ -26,7 +26,7 @@ def __init__(
         self,
         listener: listeners.EventQueueProtocol,
         settings: models.DeadlineSetting = models.DeadlineSetting(),
-        predicate: typing.Callable[[models.Event], bool] = bool,
+        predicate: Callable[[models.Event], bool] = bool,
     ) -> None:
         super().__init__()
         self._listener = listener

src/pgcachewatch/utils.py

@@ -1,18 +1,18 @@
 import asyncio
 import datetime
 import functools
-import typing
+from typing import Generator, Hashable
 
 import asyncpg
 
 from pgcachewatch import listeners, models
 
 
 def make_key(
-    args: tuple[typing.Hashable, ...],
+    args: tuple[Hashable, ...],
     kwds: dict,
     typed: bool = False,
-) -> typing.Hashable:
+) -> Hashable:
     """
     Create a cache key from the given function arguments and keyword arguments.
     """
@@ -36,7 +36,7 @@ async def emit_event(
 def pick_until_deadline(
     queue: listeners.EventQueueProtocol,
     settings: models.DeadlineSetting,
-) -> typing.Iterator[models.Event]:
+) -> Generator[models.Event, None, None]:
     """
     Yield events from the queue until the deadline is reached or queue is empty.
     """

tests/conftest.py

@@ -1,12 +1,12 @@
 import os
-import typing
+from typing import AsyncGenerator
 
 import asyncpg
 import pytest
 
 
 @pytest.fixture(scope="function")
-async def pgconn() -> typing.AsyncGenerator[asyncpg.Connection, None]:
+async def pgconn() -> AsyncGenerator[asyncpg.Connection, None]:
     conn = await asyncpg.connect()
     try:
         yield conn
@@ -15,7 +15,7 @@ async def pgconn() -> typing.AsyncGenerator[asyncpg.Connection, None]:
 
 
 @pytest.fixture(scope="function")
-async def pgpool() -> typing.AsyncGenerator[asyncpg.Pool, None]:
+async def pgpool() -> AsyncGenerator[asyncpg.Pool, None]:
     async with asyncpg.create_pool() as pool:
         yield pool
 

tests/test_listeners.py

@@ -1,14 +1,14 @@
 import asyncio
 import datetime
-import typing
+from typing import get_args
 
 import asyncpg
 import pytest
 from pgcachewatch import listeners, models, utils
 
 
 @pytest.mark.parametrize("N", (4, 8, 32))
-@pytest.mark.parametrize("operation", typing.get_args(models.OPERATIONS))
+@pytest.mark.parametrize("operation", get_args(models.OPERATIONS))
 async def test_eventqueue_and_pglistner(
     N: int,
     operation: models.OPERATIONS,

tests/test_utils.py

@@ -1,15 +1,15 @@
 import asyncio
 import datetime
 import time
-import typing
+from typing import get_args
 
 import asyncpg
 import pytest
 from pgcachewatch import listeners, models, utils
 
 
 @pytest.mark.parametrize("N", (1, 2, 8))
-@pytest.mark.parametrize("operation", typing.get_args(models.OPERATIONS))
+@pytest.mark.parametrize("operation", get_args(models.OPERATIONS))
 async def test_emit_event(
     N: int,
     operation: models.OPERATIONS,