diff --git a/function_app.py b/function_app.py index fda26cd..5d2472d 100644 --- a/function_app.py +++ b/function_app.py @@ -9,9 +9,9 @@ import lzma import os import tempfile +from collections.abc import Generator from email.utils import formatdate from pathlib import Path -from typing import Generator, Optional import azure.functions as func import pydpkg @@ -36,9 +36,9 @@ @contextlib.contextmanager -def temporary_filename() -> Generator[str, None, None]: +def temporary_filename() -> Generator[str]: """Create a temporary file and return the filename.""" - temporary_name: Optional[str] = None + temporary_name: str | None = None try: with tempfile.NamedTemporaryFile(delete=False) as f: temporary_name = f.name diff --git a/ruff.toml b/ruff.toml index 4a487ad..b149b87 100644 --- a/ruff.toml +++ b/ruff.toml @@ -14,11 +14,7 @@ indent-width = 4 target-version = "py313" [lint] -# Select explicitly rather than using extend-select, so that the rule set is -# pinned against changes to ruff's defaults (0.16 grew the default set from 59 -# to 413 rules, which silently enabled ~360 new rules here). -select = [ - "F", # pyflakes (previously picked up from ruff's defaults) +extend-select = [ "E", "I", # isort "D", # pydocstyle diff --git a/src/apt_package_function/__init__.py b/src/apt_package_function/__init__.py index ae0c7cc..bd50f9c 100644 --- a/src/apt_package_function/__init__.py +++ b/src/apt_package_function/__init__.py @@ -8,6 +8,8 @@ from pathlib import Path from typing import TextIO +log = logging.getLogger(__name__) + def common_logging(name: str, filename: str, stream: TextIO = sys.stdout) -> None: """Set up common logging.""" @@ -21,7 +23,7 @@ def common_logging(name: str, filename: str, stream: TextIO = sys.stdout) -> Non log_path = Path(log_filename) # Get the current time as a timestamp - timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + timestamp = datetime.now().astimezone().strftime("%Y-%m-%d_%H-%M-%S") rootdir = Path(__file__).parent.parent.parent logsdir = rootdir / "logs" @@ -49,4 +51,4 @@ def common_logging(name: str, filename: str, stream: TextIO = sys.stdout) -> Non root_logger.setLevel(logging.DEBUG) logging.getLogger("urllib3").setLevel(logging.INFO) - logging.info("Logging to %s", logspath) + log.info("Logging to %s", logspath) diff --git a/src/apt_package_function/azcmd.py b/src/apt_package_function/azcmd.py index a66f653..78d2d90 100644 --- a/src/apt_package_function/azcmd.py +++ b/src/apt_package_function/azcmd.py @@ -5,7 +5,7 @@ import json import logging import subprocess -from typing import Any, Dict, List, Optional +from typing import Any log = logging.getLogger(__name__) log.addHandler(logging.NullHandler()) @@ -16,13 +16,13 @@ class AzCmd: OUTPUT: str - def __init__(self, cmd: List[str], subscription: Optional[str] = None) -> None: + def __init__(self, cmd: list[str], subscription: str | None = None) -> None: """Create an AzCmd object""" self.cmd = cmd if subscription: self.cmd = [*cmd, "--subscription", subscription] - def _run_cmd(self, cmd: List[str]) -> Any: # noqa: ANN401 + def _run_cmd(self, cmd: list[str]) -> Any: # noqa: ANN401 """Runs a command and may return output""" raise NotImplementedError @@ -54,7 +54,7 @@ def run(self) -> None: """Run the Azure CLI command""" self._az_cmd() - def _run_cmd(self, cmd: List[str]) -> None: + def _run_cmd(self, cmd: list[str]) -> None: """Run a command but don't capture the output""" subprocess.run(cmd, check=True) @@ -69,21 +69,21 @@ def run(self) -> Any: # noqa: ANN401 data = self._az_cmd() return json.loads(data) - def _run_cmd(self, cmd: List[str]) -> str: + def _run_cmd(self, cmd: list[str]) -> str: return subprocess.check_output(cmd, encoding="utf-8") - def run_expect_dict(self) -> Dict[str, Any]: + def run_expect_dict(self) -> dict[str, Any]: """Run the Azure CLI command and return the result as a dictionary""" - data: Dict[str, Any] = self.run() + data: dict[str, Any] = self.run() if not isinstance(data, dict): - raise ValueError( + raise TypeError( f"Expected a dictionary, got {data.__class__.__name__}: {data}" ) return data - def run_expect_list(self) -> List[str]: + def run_expect_list(self) -> list[str]: """Run the Azure CLI command and return the result as a list of strings""" - data: Dict[str, Any] = self.run() + data: list[str] = self.run() if not isinstance(data, list): - raise ValueError(f"Expected a list, got {data}") + raise TypeError(f"Expected a list, got {data}") return data diff --git a/src/apt_package_function/bicep_deployment.py b/src/apt_package_function/bicep_deployment.py index 26c171f..37caa31 100644 --- a/src/apt_package_function/bicep_deployment.py +++ b/src/apt_package_function/bicep_deployment.py @@ -6,7 +6,7 @@ import tempfile from contextlib import ExitStack from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any from apt_package_function.azcmd import AzCmdJson, AzCmdNone @@ -22,10 +22,10 @@ def __init__( deployment_name: str, resource_group_name: str, template_file: Path, - parameters: Dict[str, Any], + parameters: dict[str, Any], description: str, - subscription: Optional[str] = None, - secure_parameters: Optional[Dict[str, str]] = None, + subscription: str | None = None, + secure_parameters: dict[str, str] | None = None, ) -> None: """Create a BicepDeployment object. @@ -85,7 +85,7 @@ def create(self) -> None: cmd.run() log.info("Finished deploying %s", self.description) - def outputs(self) -> Dict[str, Any]: + def outputs(self) -> dict[str, Any]: """Get the outputs of the deployment.""" cmd = AzCmdJson( [ diff --git a/src/apt_package_function/create_resources.py b/src/apt_package_function/create_resources.py index 5a0c477..3dd4a99 100644 --- a/src/apt_package_function/create_resources.py +++ b/src/apt_package_function/create_resources.py @@ -1,4 +1,3 @@ -#!/usr/bin/env python3 # Copyright (c) Alianza, Inc. All rights reserved. # Licensed under the MIT License. """Creates resources for the apt package function in Azure.""" diff --git a/src/apt_package_function/func_app.py b/src/apt_package_function/func_app.py index fa5770d..b6584de 100644 --- a/src/apt_package_function/func_app.py +++ b/src/apt_package_function/func_app.py @@ -11,7 +11,7 @@ from pathlib import Path from subprocess import CalledProcessError from types import TracebackType -from typing import Optional, Type +from typing import Self from zipfile import ZipFile from apt_package_function.azcmd import AzCmdJson, AzCmdNone @@ -45,7 +45,7 @@ def __init__( name: str, resource_group: str, output_path: Path, - subscription: Optional[str] = None, + subscription: str | None = None, ) -> None: """Create a FuncApp object.""" self.name = name @@ -95,15 +95,15 @@ def wait_for_event_trigger(self) -> None: time.sleep(5) - def __enter__(self) -> "FuncApp": + def __enter__(self) -> Self: """Return the object for use in a context manager.""" return self def __exit__( self, - _exc_type: Optional[Type[BaseException]], - _exc_value: Optional[BaseException], - _exc_traceback: Optional[TracebackType], + _exc_type: type[BaseException] | None, + _exc_value: BaseException | None, + _exc_traceback: TracebackType | None, ) -> None: """Clean up the object.""" if self.output_path.exists(): @@ -118,10 +118,14 @@ class FuncAppZip(FuncApp): """Class for managing zipped function apps.""" def __init__( - self, name: str, resource_group: str, subscription: Optional[str] = None + self, name: str, resource_group: str, subscription: str | None = None ) -> None: """Create a FuncAppZip object.""" - self.tempfile = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + # delete=False so the path outlives the handle: we only want a unique + # temp path to build the zip into. FuncApp.__exit__ unlinks it. + self.tempfile = tempfile.NamedTemporaryFile( # noqa: SIM115 + suffix=".zip", delete=False + ) super().__init__( name, resource_group, Path(self.tempfile.name), subscription=subscription ) @@ -170,10 +174,14 @@ class FuncAppBundle(FuncApp): _DEPLOY_POLL_INTERVAL_S = 15 def __init__( - self, name: str, resource_group: str, subscription: Optional[str] = None + self, name: str, resource_group: str, subscription: str | None = None ) -> None: """Create a FuncAppBundle object.""" - self.tempfile = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + # delete=False so the path outlives the handle: we only want a unique + # temp path to build the zip into. FuncApp.__exit__ unlinks it. + self.tempfile = tempfile.NamedTemporaryFile( # noqa: SIM115 + suffix=".zip", delete=False + ) super().__init__( name, resource_group, Path(self.tempfile.name), subscription=subscription ) @@ -200,7 +208,7 @@ def deploy(self) -> None: f"https://{self.name}.scm.azurewebsites.net/api/publish" "?type=zip&RemoteBuild=true" ) - request = urllib.request.Request( # noqa: S310 + request = urllib.request.Request( url, data=data, method="POST", @@ -220,7 +228,7 @@ def _wait_for_deployment(self, token: str) -> None: url = f"https://{self.name}.scm.azurewebsites.net/api/deployments/latest" deadline = time.monotonic() + self._DEPLOY_TIMEOUT_S while time.monotonic() < deadline: - request = urllib.request.Request( # noqa: S310 + request = urllib.request.Request( url, headers={"Authorization": f"Bearer {token}"} ) with urllib.request.urlopen(request) as response: # noqa: S310 diff --git a/src/apt_package_function/resource_group.py b/src/apt_package_function/resource_group.py index b0b3b8a..d29078c 100644 --- a/src/apt_package_function/resource_group.py +++ b/src/apt_package_function/resource_group.py @@ -3,7 +3,6 @@ """Manages resource groups.""" import logging -from typing import Optional from apt_package_function.azcmd import AzCmdNone @@ -12,7 +11,7 @@ def create_rg( - resource_group: str, location: str, subscription: Optional[str] = None + resource_group: str, location: str, subscription: str | None = None ) -> None: """Create a resource group.""" log.debug("Creating resource group %s in location %s", resource_group, location) diff --git a/src/apt_package_function/signing.py b/src/apt_package_function/signing.py index cc0a015..dc578e7 100644 --- a/src/apt_package_function/signing.py +++ b/src/apt_package_function/signing.py @@ -36,7 +36,7 @@ def load_private_key(path: str) -> str: try: key, _ = pgpy.PGPKey.from_blob(blob) - except Exception as e: # noqa: BLE001 - pgpy raises a variety of errors + except Exception as e: # pgpy raises a variety of errors raise ValueError(f"Could not parse a PGP key from {path}: {e}") from e if not key.is_public and key.is_protected: