Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: upgrade to ruff 0.0.270 #434

Merged
merged 1 commit into from
May 28, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
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