made ruff happy, hopefully no issues with these updates.

portalocker/__main__.py

@@ -50,7 +50,7 @@ def main(argv: typing.Sequence[str] | None = None) -> None:
 
 def _read_file(
     path: pathlib.Path,
-    seen_files: typing.Set[pathlib.Path],
+    seen_files: set[pathlib.Path],
 ) -> typing.Iterator[str]:
     if path in seen_files:
         return
@@ -114,7 +114,7 @@ def combine(args: argparse.Namespace):
         _TEXT_TEMPLATE.format((base_path / 'LICENSE').read_text()),
     )
 
-    seen_files: typing.Set[pathlib.Path] = set()
+    seen_files: set[pathlib.Path] = set()
     for line in _read_file(src_path / '__init__.py', seen_files):
         output_file.write(line)
 

portalocker/portalocker.py

@@ -14,9 +14,7 @@ class HasFileno(typing.Protocol):
     def fileno(self) -> int: ...
 
 
-LOCKER: typing.Optional[
-    typing.Callable[[typing.Union[int, HasFileno], int], typing.Any]
-] = None
+LOCKER: typing.Callable[[int | HasFileno, int], typing.Any] | None = None
 
 if os.name == 'nt':  # pragma: no cover
     import msvcrt
@@ -28,7 +26,7 @@ def fileno(self) -> int: ...
 
     __overlapped = pywintypes.OVERLAPPED()
 
-    def lock(file_: typing.Union[typing.IO, int], flags: LockFlags):
+    def lock(file_: typing.IO | int, flags: LockFlags):
         # Windows locking does not support locking through `fh.fileno()` so
         # we cast it to make mypy and pyright happy
         file_ = typing.cast(typing.IO, file_)

portalocker/redis.py

@@ -64,29 +64,29 @@ class RedisLock(utils.LockBase):
 
     '''
 
-    redis_kwargs: typing.Dict[str, typing.Any]
-    thread: typing.Optional[PubSubWorkerThread]
+    redis_kwargs: dict[str, typing.Any]
+    thread: PubSubWorkerThread | None
     channel: str
     timeout: float
-    connection: typing.Optional[redis.client.Redis[str]]
-    pubsub: typing.Optional[redis.client.PubSub] = None
+    connection: redis.client.Redis[str] | None
+    pubsub: redis.client.PubSub | None = None
     close_connection: bool
 
-    DEFAULT_REDIS_KWARGS: typing.ClassVar[typing.Dict[str, typing.Any]] = dict(
+    DEFAULT_REDIS_KWARGS: typing.ClassVar[dict[str, typing.Any]] = dict(
         health_check_interval=10,
         decode_responses=True,
     )
 
     def __init__(
         self,
         channel: str,
-        connection: typing.Optional[redis.client.Redis[str]] = None,
-        timeout: typing.Optional[float] = None,
-        check_interval: typing.Optional[float] = None,
-        fail_when_locked: typing.Optional[bool] = False,
+        connection: redis.client.Redis[str] | None = None,
+        timeout: float | None = None,
+        check_interval: float | None = None,
+        fail_when_locked: bool | None = False,
         thread_sleep_time: float = DEFAULT_THREAD_SLEEP_TIME,
         unavailable_timeout: float = DEFAULT_UNAVAILABLE_TIMEOUT,
-        redis_kwargs: typing.Optional[typing.Dict[str, typing.Any]] = None,
+        redis_kwargs: dict[str, typing.Any] | None = None,
     ):
         # We don't want to close connections given as an argument
         self.close_connection = not connection
@@ -113,7 +113,7 @@ def get_connection(self) -> redis.client.Redis[str]:
 
         return self.connection
 
-    def channel_handler(self, message: typing.Dict[str, str]) -> None:
+    def channel_handler(self, message: dict[str, str]) -> None:
         if message.get('type') != 'message':  # pragma: no cover
             return
 
@@ -136,9 +136,9 @@ def client_name(self):
 
     def acquire(  # type: ignore[override]
         self,
-        timeout: typing.Optional[float] = None,
-        check_interval: typing.Optional[float] = None,
-        fail_when_locked: typing.Optional[bool] = None,
+        timeout: float | None = None,
+        check_interval: float | None = None,
+        fail_when_locked: bool | None = None,
     ) -> RedisLock:
         timeout = utils.coalesce(timeout, self.timeout, 0.0)
         check_interval = utils.coalesce(

portalocker/utils.py

@@ -94,15 +94,15 @@ def open_atomic(
     # Create the parent directory if it doesn't exist
     path.parent.mkdir(parents=True, exist_ok=True)
 
-    temp_fh = tempfile.NamedTemporaryFile(
+    with tempfile.NamedTemporaryFile(
         mode=binary and 'wb' or 'w',
         dir=str(path.parent),
         delete=False,
-    )
-    yield temp_fh
-    temp_fh.flush()
-    os.fsync(temp_fh.fileno())
-    temp_fh.close()
+    ) as temp_fh:
+        yield temp_fh
+        temp_fh.flush()
+        os.fsync(temp_fh.fileno())
+
     try:
         os.rename(temp_fh.name, path)
     finally:
@@ -120,9 +120,9 @@ class LockBase(abc.ABC):  # pragma: no cover
 
     def __init__(
         self,
-        timeout: typing.Optional[float] = None,
-        check_interval: typing.Optional[float] = None,
-        fail_when_locked: typing.Optional[bool] = None,
+        timeout: float | None = None,
+        check_interval: float | None = None,
+        fail_when_locked: bool | None = None,
     ):
         self.timeout = coalesce(timeout, DEFAULT_TIMEOUT)
         self.check_interval = coalesce(check_interval, DEFAULT_CHECK_INTERVAL)
@@ -134,15 +134,15 @@ def __init__(
     @abc.abstractmethod
     def acquire(
         self,
-        timeout: typing.Optional[float] = None,
-        check_interval: typing.Optional[float] = None,
-        fail_when_locked: typing.Optional[bool] = None,
+        timeout: float | None = None,
+        check_interval: float | None = None,
+        fail_when_locked: bool | None = None,
     ) -> typing.IO[typing.AnyStr]: ...
 
     def _timeout_generator(
         self,
-        timeout: typing.Optional[float],
-        check_interval: typing.Optional[float],
+        timeout: float | None,
+        check_interval: float | None,
     ) -> typing.Iterator[int]:
         f_timeout = coalesce(timeout, self.timeout, 0.0)
         f_check_interval = coalesce(check_interval, self.check_interval, 0.0)
@@ -167,10 +167,10 @@ def __enter__(self) -> typing.IO[typing.AnyStr]:
 
     def __exit__(
         self,
-        exc_type: typing.Optional[typing.Type[BaseException]],
-        exc_value: typing.Optional[BaseException],
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
         traceback: typing.Any,  # Should be typing.TracebackType
-    ) -> typing.Optional[bool]:
+    ) -> bool | None:
         self.release()
         return None
 
@@ -243,9 +243,9 @@ def __init__(
 
     def acquire(
         self,
-        timeout: typing.Optional[float] = None,
-        check_interval: typing.Optional[float] = None,
-        fail_when_locked: typing.Optional[bool] = None,
+        timeout: float | None = None,
+        check_interval: float | None = None,
+        fail_when_locked: bool | None = None,
     ) -> typing.IO[typing.AnyStr]:
         '''Acquire the locked filehandle'''
 
@@ -383,9 +383,9 @@ def __init__(
 
     def acquire(
         self,
-        timeout: typing.Optional[float] = None,
-        check_interval: typing.Optional[float] = None,
-        fail_when_locked: typing.Optional[bool] = None,
+        timeout: float | None = None,
+        check_interval: float | None = None,
+        fail_when_locked: bool | None = None,
     ) -> typing.IO[typing.AnyStr]:
         fh: typing.IO[typing.AnyStr]
         if self._acquire_count >= 1:
@@ -449,23 +449,23 @@ class BoundedSemaphore(LockBase):
     'bounded_semaphore.01.lock'
     """
 
-    lock: typing.Optional[Lock]
+    lock: Lock | None
 
     def __init__(
         self,
         maximum: int,
         name: str = 'bounded_semaphore',
         filename_pattern: str = '{name}.{number:02d}.lock',
         directory: str = tempfile.gettempdir(),
-        timeout: typing.Optional[float] = DEFAULT_TIMEOUT,
-        check_interval: typing.Optional[float] = DEFAULT_CHECK_INTERVAL,
-        fail_when_locked: typing.Optional[bool] = True,
+        timeout: float | None = DEFAULT_TIMEOUT,
+        check_interval: float | None = DEFAULT_CHECK_INTERVAL,
+        fail_when_locked: bool | None = True,
     ):
         self.maximum = maximum
         self.name = name
         self.filename_pattern = filename_pattern
         self.directory = directory
-        self.lock: typing.Optional[Lock] = None
+        self.lock: Lock | None = None
         super().__init__(
             timeout=timeout,
             check_interval=check_interval,
@@ -496,10 +496,10 @@ def get_filename(self, number: int) -> pathlib.Path:
 
     def acquire(  # type: ignore[override]
         self,
-        timeout: typing.Optional[float] = None,
-        check_interval: typing.Optional[float] = None,
-        fail_when_locked: typing.Optional[bool] = None,
-    ) -> typing.Optional[Lock]:
+        timeout: float | None = None,
+        check_interval: float | None = None,
+        fail_when_locked: bool | None = None,
+    ) -> Lock | None:
         assert not self.lock, 'Already locked'
 
         filenames = self.get_filenames()
@@ -567,12 +567,12 @@ class NamedBoundedSemaphore(BoundedSemaphore):
     def __init__(
         self,
         maximum: int,
-        name: typing.Optional[str] = None,
+        name: str | None = None,
         filename_pattern: str = '{name}.{number:02d}.lock',
         directory: str = tempfile.gettempdir(),
-        timeout: typing.Optional[float] = DEFAULT_TIMEOUT,
-        check_interval: typing.Optional[float] = DEFAULT_CHECK_INTERVAL,
-        fail_when_locked: typing.Optional[bool] = True,
+        timeout: float | None = DEFAULT_TIMEOUT,
+        check_interval: float | None = DEFAULT_CHECK_INTERVAL,
+        fail_when_locked: bool | None = True,
     ):
         if name is None:
             name = 'bounded_semaphore.%d' % random.randint(0, 1000000)

portalocker_tests/tests.py

@@ -287,7 +287,7 @@ def exclusive_lock(filename, **kwargs):
 
 @dataclasses.dataclass(order=True)
 class LockResult:
-    exception_class: typing.Union[typing.Type[Exception], None] = None
+    exception_class: typing.Union[type[Exception], None] = None
     exception_message: typing.Union[str, None] = None
     exception_repr: typing.Union[str, None] = None
 

ruff.toml

@@ -1,7 +1,7 @@
 # We keep the ruff configuration separate so it can easily be shared across
 # all projects
 
-target-version = 'py38'
+target-version = 'py39'
 
 src = ['portalocker']
 exclude = ['docs']