Skip to content

Commit

Permalink
chore: upgrade to ruff 0.0.270 (#434)
Browse files Browse the repository at this point in the history
includes updates to address the breaking change in
astral-sh/ruff#4427 and fixes for the now
enabled rules:
- UP006 (non-pep585-annotation)
- UP007 (non-pep604-annotation)
  • Loading branch information
tekumara committed May 28, 2023
1 parent 85c77f4 commit e64f3cf
Show file tree
Hide file tree
Showing 10 changed files with 107 additions and 113 deletions.
6 changes: 1 addition & 5 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ dev = [
"pyfakefs~=5.1",
"pytest~=7.2",
"pytest-mock~=3.8",
"ruff~=0.0.259",
"ruff~=0.0.270",
"twine~=4.0",
]

Expand Down Expand Up @@ -99,10 +99,6 @@ src = ["src"]
fix = true
show-fixes = true

[tool.ruff.pyupgrade]
# Preserve types, even if a file imports `from __future__ import annotations`.
keep-runtime-typing = true

[tool.ruff.per-file-ignores]
# test functions don't need return types
"tests/*" = ["ANN201", "ANN202"]
Expand Down
34 changes: 17 additions & 17 deletions src/aec/command/ami.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from __future__ import annotations

from typing import TYPE_CHECKING, Any, Dict, List, NamedTuple, Optional, Sequence
from typing import TYPE_CHECKING, Any, NamedTuple, Sequence

import boto3

Expand All @@ -14,11 +14,11 @@


class Image(TypedDict):
Name: Optional[str]
Name: str | None
ImageId: str
CreationDate: str
RootDeviceName: Optional[str]
Size: Optional[int]
RootDeviceName: str | None
Size: int | None
SnapshotId: NotRequired[str]


Expand Down Expand Up @@ -59,9 +59,9 @@ def fetch(config: Config, ami: str) -> Image:

def _describe_images(
config: Config,
ident: Optional[str] = None,
owner: Optional[str] = None,
name_match: Optional[str] = None,
ident: str | None = None,
owner: str | None = None,
name_match: str | None = None,
) -> DescribeImagesResultTypeDef:
ec2_client = boto3.client("ec2", region_name=config.get("region", None))

Expand All @@ -77,7 +77,7 @@ def _describe_images(
elif isinstance(describe_images_owners, str):
owners_filter = [describe_images_owners]
else:
owners_filter: List[str] = describe_images_owners
owners_filter: list[str] = describe_images_owners

if name_match is None:
name_match = config.get("describe_images_name_match", None)
Expand All @@ -86,7 +86,7 @@ def _describe_images(
filters = [{"Name": "name", "Values": [f"{ident}"]}] if ident else []
match_desc = f" named {ident}" if ident else ""
else:
filters: List[FilterTypeDef] = [{"Name": "name", "Values": [f"*{name_match}*"]}]
filters: list[FilterTypeDef] = [{"Name": "name", "Values": [f"*{name_match}*"]}]
match_desc = f" with name containing {name_match}"

print(f"Describing images owned by {owners_filter}{match_desc}")
Expand All @@ -96,11 +96,11 @@ def _describe_images(

def describe(
config: Config,
ident: Optional[str] = None,
owner: Optional[str] = None,
name_match: Optional[str] = None,
ident: str | None = None,
owner: str | None = None,
name_match: str | None = None,
show_snapshot_id: bool = False,
) -> List[Image]:
) -> list[Image]:
"""List AMIs."""

response = _describe_images(config, ident, owner, name_match)
Expand All @@ -123,11 +123,11 @@ def describe(

def describe_tags(
config: Config,
ident: Optional[str] = None,
owner: Optional[str] = None,
name_match: Optional[str] = None,
ident: str | None = None,
owner: str | None = None,
name_match: str | None = None,
keys: Sequence[str] = [],
) -> List[Dict[str, Any]]:
) -> list[dict[str, Any]]:
"""List AMI images with their tags."""

response = _describe_images(config, ident, owner, name_match)
Expand Down
6 changes: 3 additions & 3 deletions src/aec/command/compute_optimizer.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations

from datetime import datetime
from typing import TYPE_CHECKING, Any, Dict, List
from typing import TYPE_CHECKING, Any

import boto3
import pytz
Expand All @@ -13,7 +13,7 @@
from aec.util.config import Config


def over_provisioned(config: Config) -> List[Dict[str, Any]]:
def over_provisioned(config: Config) -> list[dict[str, Any]]:
"""Show recommendations for over-provisioned EC2 instances."""

def util(metric: UtilizationMetricTypeDef) -> str:
Expand All @@ -40,7 +40,7 @@ def util(metric: UtilizationMetricTypeDef) -> str:
return recs


def describe_instances_uptime(config: Config) -> Dict[str, str]:
def describe_instances_uptime(config: Config) -> dict[str, str]:
"""List EC2 instance uptimes in the region."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))
Expand Down
86 changes: 43 additions & 43 deletions src/aec/command/ec2.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import os.path
from collections import defaultdict
from time import sleep
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, cast
from typing import TYPE_CHECKING, Any, Sequence, cast

import boto3
from typing_extensions import TypedDict
Expand Down Expand Up @@ -38,25 +38,25 @@ def is_ebs_optimizable(instance_type: str) -> bool:
class Instance(TypedDict, total=False):
InstanceId: str
State: str
Name: Optional[str]
Name: str | None
Type: str
DnsName: str
SubnetId: str
Volumes: List[str]
Volumes: list[str]


def launch(
config: Config,
name: str,
ami: Optional[str] = None,
template: Optional[str] = None,
volume_size: Optional[int] = None,
ami: str | None = None,
template: str | None = None,
volume_size: int | None = None,
encrypted: bool = True,
instance_type: Optional[str] = None,
key_name: Optional[str] = None,
userdata: Optional[str] = None,
instance_type: str | None = None,
key_name: str | None = None,
userdata: str | None = None,
wait_ssm: bool = False,
) -> List[Instance]:
) -> list[Instance]:
"""Launch a tagged EC2 instance with an EBS volume."""

template = template or config.get("launch_template", None)
Expand Down Expand Up @@ -122,7 +122,7 @@ def launch(
runargs["InstanceType"] = cast("InstanceTypeType", instance_type)
runargs["EbsOptimized"] = is_ebs_optimizable(instance_type)

tags: List[TagTypeDef] = [{"Key": "Name", "Value": name}]
tags: list[TagTypeDef] = [{"Key": "Name", "Value": name}]
additional_tags = config.get("additional_tags", {})
if additional_tags:
tags.extend([{"Key": k, "Value": v} for k, v in additional_tags.items()])
Expand Down Expand Up @@ -187,7 +187,7 @@ def launch(
return describe(config=config, ident=instance_id)


def _wait_ssm_agent_online(config: Config, instance_ids: List[str]) -> None:
def _wait_ssm_agent_online(config: Config, instance_ids: list[str]) -> None:
"""
Wait for ssm to come online.
Expand All @@ -208,13 +208,13 @@ def _wait_ssm_agent_online(config: Config, instance_ids: List[str]) -> None:

def describe(
config: Config,
ident: Optional[str] = None,
name_match: Optional[str] = None,
ident: str | None = None,
name_match: str | None = None,
include_terminated: bool = False,
show_running_only: bool = False,
sort_by: str = "State,Name",
columns: str = "InstanceId,State,Name,Type,DnsName,LaunchTime,ImageId",
) -> List[Instance]:
) -> list[Instance]:
"""List EC2 instances in the region."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))
Expand All @@ -223,7 +223,7 @@ def describe(
if show_running_only:
filters.append({"Name": "instance-state-name", "Values": ["pending", "running"]})

kwargs: Dict[str, Any] = {"MaxResults": 1000, "Filters": filters}
kwargs: dict[str, Any] = {"MaxResults": 1000, "Filters": filters}

response_fut = executor.submit(ec2_client.describe_instances, **kwargs)

Expand All @@ -235,7 +235,7 @@ def describe(
if "Volumes" in columns:
# fetch volume info
volumes_response: DescribeVolumesResultTypeDef = executor.submit(ec2_client.describe_volumes).result()
volumes: Dict[str, List[str]] = defaultdict(list)
volumes: dict[str, list[str]] = defaultdict(list)
for v in volumes_response["Volumes"]:
for a in v["Attachments"]:
volumes[a["InstanceId"]].append(f'Size={v["Size"]} GiB')
Expand All @@ -246,7 +246,7 @@ def describe(

# import json; print(json.dumps(response))

instances: List[Instance] = []
instances: list[Instance] = []
while True:
for r in response["Reservations"]:
for i in r["Instances"]:
Expand Down Expand Up @@ -284,11 +284,11 @@ def describe(

def describe_tags(
config: Config,
ident: Optional[str] = None,
name_match: Optional[str] = None,
ident: str | None = None,
name_match: str | None = None,
keys: Sequence[str] = [],
volumes: bool = False,
) -> List[Dict[str, Any]]:
) -> list[dict[str, Any]]:
"""List EC2 instances or volumes with their tags."""
if volumes:
return volume_tags(config, ident, name_match, keys)
Expand All @@ -298,14 +298,14 @@ def describe_tags(

def tag(
config: Config,
ident: Optional[str] = None,
name_match: Optional[str] = None,
ident: str | None = None,
name_match: str | None = None,
tags: Sequence[str] = [],
) -> List[Dict[str, Any]]:
) -> list[dict[str, Any]]:
"""Tag EC2 instance(s)."""
ec2_client = boto3.client("ec2", region_name=config.get("region", None))

tagdefs: List[TagTypeDef] = []
tagdefs: list[TagTypeDef] = []

for t in tags:
parts = t.split("=")
Expand All @@ -328,15 +328,15 @@ def tag(


def instance_tags(
config: Config, ident: Optional[str] = None, name_match: Optional[str] = None, keys: Sequence[str] = []
) -> List[Dict[str, Any]]:
config: Config, ident: str | None = None, name_match: str | None = None, keys: Sequence[str] = []
) -> list[dict[str, Any]]:
"""List EC2 instances with their tags."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))

response = ec2_client.describe_instances(Filters=to_filters(ident, name_match))

instances: List[Dict[str, Any]] = []
instances: list[dict[str, Any]] = []
for r in response["Reservations"]:
for i in r["Instances"]:
if i["State"]["Name"] != "terminated":
Expand All @@ -353,15 +353,15 @@ def instance_tags(


def volume_tags(
config: Config, ident: Optional[str] = None, name_match: Optional[str] = None, keys: Sequence[str] = []
) -> List[Dict[str, Any]]:
config: Config, ident: str | None = None, name_match: str | None = None, keys: Sequence[str] = []
) -> list[dict[str, Any]]:
"""List EC2 volumes with their tags."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))

response = ec2_client.describe_volumes(Filters=to_filters(ident, name_match))

volumes: List[Dict[str, Any]] = []
volumes: list[dict[str, Any]] = []
for v in response["Volumes"]:
vol = {"VolumeId": v["VolumeId"], "Name": util_tags.get_value(v, "Name")}
if not keys:
Expand All @@ -379,7 +379,7 @@ def start(
config: Config,
ident: str,
wait_ssm: bool = False,
) -> List[Instance]:
) -> list[Instance]:
"""Start EC2 instance."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))
Expand Down Expand Up @@ -411,7 +411,7 @@ def start(
return describe(config, ident)


def stop(config: Config, ident: str) -> List[Dict[str, Any]]:
def stop(config: Config, ident: str) -> list[dict[str, Any]]:
"""Stop EC2 instance."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))
Expand All @@ -430,7 +430,7 @@ def stop(config: Config, ident: str) -> List[Dict[str, Any]]:
return [{"State": i["CurrentState"]["Name"], "InstanceId": i["InstanceId"]} for i in response["StoppingInstances"]]


def terminate(config: Config, ident: str) -> List[Dict[str, Any]]:
def terminate(config: Config, ident: str) -> list[dict[str, Any]]:
"""Terminate EC2 instance."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))
Expand All @@ -451,7 +451,7 @@ def terminate(config: Config, ident: str) -> List[Dict[str, Any]]:
]


def modify(config: Config, ident: str, type: str) -> List[Instance]:
def modify(config: Config, ident: str, type: str) -> list[Instance]:
"""Change an instance's type."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))
Expand Down Expand Up @@ -506,7 +506,7 @@ def logs(config: Config, ident: str) -> str:
return response.get("Output", "No logs yet 馃様")


def templates(config: Config) -> List[Dict[str, Any]]:
def templates(config: Config) -> list[dict[str, Any]]:
"""Describe launch templates."""

ec2_client = boto3.client("ec2", region_name=config.get("region", None))
Expand All @@ -521,19 +521,19 @@ def templates(config: Config) -> List[Dict[str, Any]]:

def status(
config: Config,
ident: Optional[str] = None,
name_match: Optional[str] = None,
) -> List[Dict[str, Any]]:
ident: str | None = None,
name_match: str | None = None,
) -> list[dict[str, Any]]:
"""Describe instances status checks."""
ec2_client = boto3.client("ec2", region_name=config.get("region", None))

kwargs: Dict[str, Any] = {"MaxResults": 1000}
kwargs: dict[str, Any] = {"MaxResults": 1000}

response_fut = executor.submit(ec2_client.describe_instance_status, **kwargs)
instances = executor.submit(describe_running_instances_names, config).result()
response = response_fut.result()

def match(instance_id: str, instance_name: Optional[str]) -> bool:
def match(instance_id: str, instance_name: str | None) -> bool:
# describe_instance_status doesn't support name filters in the request so match here
if not ident and not name_match:
return True
Expand Down Expand Up @@ -579,7 +579,7 @@ def status_text(summary: InstanceStatusSummaryTypeDef, key: str = "reachability"
)


def user_data(config: Config, ident: str) -> Optional[str]:
def user_data(config: Config, ident: str) -> str | None:
"""Describe user data for an instance."""
ec2_client = boto3.client("ec2", region_name=config.get("region", None))

Expand All @@ -601,7 +601,7 @@ def user_data(config: Config, ident: str) -> Optional[str]:
return None


def to_filters(ident: Optional[str] = None, name_match: Optional[str] = None) -> List[FilterTypeDef]:
def to_filters(ident: str | None = None, name_match: str | None = None) -> list[FilterTypeDef]:
if ident and ident.startswith("i-"):
return [{"Name": "instance-id", "Values": [ident]}]
elif ident:
Expand Down

0 comments on commit e64f3cf

Please sign in to comment.