Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions function_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
6 changes: 1 addition & 5 deletions ruff.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 4 additions & 2 deletions src/apt_package_function/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand All @@ -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"
Expand Down Expand Up @@ -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)
22 changes: 11 additions & 11 deletions src/apt_package_function/azcmd.py
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand All @@ -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

Expand Down Expand Up @@ -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)

Expand All @@ -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
10 changes: 5 additions & 5 deletions src/apt_package_function/bicep_deployment.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.

Expand Down Expand Up @@ -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(
[
Expand Down
1 change: 0 additions & 1 deletion src/apt_package_function/create_resources.py
Original file line number Diff line number Diff line change
@@ -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."""
Expand Down
32 changes: 20 additions & 12 deletions src/apt_package_function/func_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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():
Expand All @@ -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
)
Expand Down Expand Up @@ -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
)
Expand All @@ -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",
Expand All @@ -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
Expand Down
3 changes: 1 addition & 2 deletions src/apt_package_function/resource_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
"""Manages resource groups."""

import logging
from typing import Optional

from apt_package_function.azcmd import AzCmdNone

Expand All @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion src/apt_package_function/signing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down