V3: type hints for operations

pyinfra/api/command.py

@@ -1,7 +1,9 @@
+from __future__ import annotations
+
 import shlex
 from inspect import getfullargspec
 from string import Formatter
-from typing import TYPE_CHECKING, Callable, Union
+from typing import IO, TYPE_CHECKING, Callable, Union
 
 import gevent
 from typing_extensions import Unpack
@@ -143,7 +145,7 @@ def execute(self, state: "State", host: "Host", connector_arguments: ConnectorAr
 class FileUploadCommand(PyinfraCommand):
     def __init__(
         self,
-        src: str,
+        src: str | IO,
         dest: str,
         remote_temp_filename=None,
         **kwargs: Unpack[ConnectorArguments],
@@ -173,7 +175,7 @@ class FileDownloadCommand(PyinfraCommand):
     def __init__(
         self,
         src: str,
-        dest: str,
+        dest: str | IO,
         remote_temp_filename=None,
         **kwargs: Unpack[ConnectorArguments],
     ):

pyinfra/api/util.py

@@ -109,16 +109,16 @@ def get_caller_frameinfo(frame_offset: int = 0):
 
 
 def get_operation_order_from_stack(state: "State"):
+
     stack_items = list(reversed(stack()))
 
+    i = 0
     # Find the *first* occurrence of our deploy file in the reversed stack
     if state.current_deploy_filename:
         for i, stack_item in enumerate(stack_items):
             frame = getframeinfo(stack_item[0])
             if frame.filename == state.current_deploy_filename:
                 break
-    else:
-        i = 0
 
     # Now generate a list of line numbers *following that file*
     line_numbers = []
@@ -139,7 +139,7 @@ def get_operation_order_from_stack(state: "State"):
     return line_numbers
 
 
-def get_template(filename_or_io: str):
+def get_template(filename_or_io: str | IO):
     """
     Gets a jinja2 ``Template`` object for the input filename or string, with caching
     based on the filename of the template, or the SHA1 of the input string.

pyinfra/facts/apk.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/apt.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 
 from pyinfra.api import FactBase

pyinfra/facts/brew.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 
 from pyinfra import logger

pyinfra/facts/bsdinit.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from .sysvinit import InitdStatus
 
 

pyinfra/facts/cargo.py

@@ -1,5 +1,7 @@
 # encoding: utf8
 
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/choco.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/deb.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 import shlex
 

pyinfra/facts/dnf.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util import make_cat_files_command

pyinfra/facts/docker.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 from pyinfra.api import FactBase

pyinfra/facts/files.py

@@ -2,6 +2,8 @@
 The files facts provide information about the filesystem and it's contents on the target host.
 """
 
+from __future__ import annotations
+
 import re
 import stat
 from datetime import datetime

pyinfra/facts/gem.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/gpg.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from urllib.parse import urlparse
 
 from pyinfra.api import FactBase

pyinfra/facts/launchd.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 

pyinfra/facts/lxd.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 from pyinfra.api import FactBase

pyinfra/facts/mysql.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 from collections import defaultdict
 
@@ -8,11 +10,11 @@
 
 
 def make_mysql_command(
-    database=None,
-    user=None,
-    password=None,
-    host=None,
-    port=None,
+    database: str | None = None,
+    user: str | None = None,
+    password: str | None = None,
+    host: str | None = None,
+    port: int | None = None,
     executable="mysql",
 ):
     target_bits = [executable]
@@ -37,7 +39,11 @@ def make_mysql_command(
     return StringCommand(*target_bits)
 
 
-def make_execute_mysql_command(command, ignore_errors=False, **mysql_kwargs):
+def make_execute_mysql_command(
+    command: str | StringCommand,
+    ignore_errors=False,
+    **mysql_kwargs,
+):
     commands_bits = [
         make_mysql_command(**mysql_kwargs),
         "-Be",

pyinfra/facts/npm.py

@@ -1,4 +1,5 @@
 # encoding: utf8
+from __future__ import annotations
 
 from pyinfra.api import FactBase
 

pyinfra/facts/openrc.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 
 from pyinfra.api import FactBase

pyinfra/facts/pacman.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import shlex
 
 from pyinfra.api import FactBase

pyinfra/facts/pip.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/pkg.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/pkgin.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/postgres.py

@@ -7,13 +7,13 @@
 
 
 def make_psql_command(
-    database=None,
-    user=None,
-    password=None,
-    host=None,
-    port=None,
+    database: str | None = None,
+    user: str | None = None,
+    password: str | None = None,
+    host: str | None = None,
+    port: str | int | None = None,
     executable="psql",
-):
+) -> StringCommand:
     target_bits: list[str] = []
 
     if password:

pyinfra/facts/postgresql.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from .postgres import PostgresDatabases, PostgresRoles
 
 

pyinfra/facts/rpm.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 import shlex
 

pyinfra/facts/server.py

@@ -5,7 +5,7 @@
 import shutil
 from datetime import datetime
 from tempfile import mkdtemp
-from typing import Dict, List, NewType, Optional, Union
+from typing import Dict, List, Optional, Union
 
 from dateutil.parser import parse as parse_date
 from distro import distro
@@ -350,8 +350,7 @@ class Groups(FactBase[List[str]]):
     command = "cat /etc/group"
     default = list
 
-    @staticmethod
-    def process(output) -> list[str]:
+    def process(self, output) -> list[str]:
         groups: list[str] = []
 
         for line in output:
@@ -361,9 +360,6 @@ def process(output) -> list[str]:
         return groups
 
 
-CrontabCommand = NewType("CrontabCommand", int)
-
-
 class CrontabDict(TypedDict):
     minute: NotRequired[Union[int, str]]
     hour: NotRequired[Union[int, str]]
@@ -374,7 +370,7 @@ class CrontabDict(TypedDict):
     special_time: NotRequired[str]
 
 
-class Crontab(FactBase[Dict[CrontabCommand, CrontabDict]]):
+class Crontab(FactBase[Dict[str, CrontabDict]]):
     """
     Returns a dictionary of cron command -> execution time.
 
@@ -404,9 +400,8 @@ def command(user=None):
             return "crontab -l -u {0} || true".format(user)
         return "crontab -l || true"
 
-    @staticmethod
-    def process(output):
-        crons: dict[Command, CrontabDict] = {}
+    def process(self, output):
+        crons: dict[str, CrontabDict] = {}
         current_comments = []
 
         for line in output:

pyinfra/facts/snap.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 
 from pyinfra.api import FactBase

pyinfra/facts/systemd.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 from typing import Dict, Iterable
 

pyinfra/facts/upstart.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import re
 
 from pyinfra.api import FactBase

pyinfra/facts/util/packaging.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import re
+from typing import Iterable
 
 
-def parse_packages(regex, output):
+def parse_packages(regex: str, output: Iterable[str]) -> dict[str, set[str]]:
     packages: dict[str, set[str]] = {}
 
     for line in output:

pyinfra/facts/vzctl.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 from pyinfra.api import FactBase

pyinfra/facts/xbps.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util.packaging import parse_packages

pyinfra/facts/yum.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util import make_cat_files_command

pyinfra/facts/zypper.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pyinfra.api import FactBase
 
 from .util import make_cat_files_command

pyinfra/operations/apk.py

@@ -2,6 +2,8 @@
 Manage apk packages.
 """
 
+from __future__ import annotations
+
 from pyinfra import host
 from pyinfra.api import operation
 from pyinfra.facts.apk import ApkPackages
@@ -40,7 +42,7 @@ def update():
 
 @operation()
 def packages(
-    packages=None,
+    packages: str | list[str] | None = None,
     present=True,
     latest=False,
     update=False,