internet/community.general
3
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-03-26 21:33:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add basic typing for module_utils (#11222)

Browse Source 
* Add basic typing for module_utils.

* Apply some suggestions.

Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>

* Make pass again.

* Add more types as suggested.

* Normalize extra imports.

* Add more type hints.

* Improve typing.

* Add changelog fragment.

* Reduce changelog.

* Apply suggestions from code review.

Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>

* Fix typo.

* Cleanup.

* Improve types and make type checking happy.

* Let's see whether older Pythons barf on this.

* Revert "Let's see whether older Pythons barf on this."

This reverts commit 9973af3dbe.

* Add noqa.

---------

Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>
This commit is contained in:
Felix Fontein
2025-12-01 20:40:06 +01:00
committed by GitHub
parent fb2f34ba85
commit c7f6a28d89
56 changed files with 725 additions and 469 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
plugins/module_utils/_filelock.py
View File

@@ -11,9 +11,13 @@ import os
import stat
import time
import fcntl
import typing as t
from contextlib import contextmanager
if t.TYPE_CHECKING:
from io import TextIOWrapper
class LockTimeout(Exception):
pass
@@ -27,11 +31,13 @@ class FileLock:
unwanted and/or unexpected behaviour
"""
def __init__(self):
self.lockfd = None
def __init__(self) -> None:
self.lockfd: TextIOWrapper | None = None
@contextmanager
def lock_file(self, path, tmpdir, lock_timeout=None):
def lock_file(
self, path: os.PathLike, tmpdir: os.PathLike, lock_timeout: int | float | None = None
) -> t.Generator[None]:
"""
Context for lock acquisition
"""
@@ -41,7 +47,9 @@ class FileLock:
finally:
self.unlock()
def set_lock(self, path, tmpdir, lock_timeout=None):
def set_lock(
self, path: os.PathLike, tmpdir: os.PathLike, lock_timeout: int | float | None = None
) -> t.Literal[True]:
"""
Create a lock file based on path with flock to prevent other processes
using given path.
@@ -61,13 +69,13 @@ class FileLock:
self.lockfd = open(lock_path, "w")
if lock_timeout <= 0:
if lock_timeout is not None and lock_timeout <= 0:
fcntl.flock(self.lockfd, fcntl.LOCK_EX | fcntl.LOCK_NB)
os.chmod(lock_path, stat.S_IWRITE | stat.S_IREAD)
return True
if lock_timeout:
e_secs = 0
e_secs: float = 0
while e_secs < lock_timeout:
try:
fcntl.flock(self.lockfd, fcntl.LOCK_EX | fcntl.LOCK_NB)
@@ -86,7 +94,7 @@ class FileLock:
return True
def unlock(self):
def unlock(self) -> t.Literal[True]:
"""
Make sure lock file is available for everyone and Unlock the file descriptor
locked by set_lock