From 240e064f0f4bb789f809a11c641706802d6a3ea7 Mon Sep 17 00:00:00 2001 From: Oliver Mannion <125105+tekumara@users.noreply.github.com> Date: Sun, 28 May 2023 21:34:44 +1000 Subject: [PATCH] . --- pyproject.toml | 6 +- src/aec/command/ami.py | 34 +++++------ src/aec/command/compute_optimizer.py | 6 +- src/aec/command/ec2.py | 86 ++++++++++++++-------------- src/aec/command/ssm.py | 42 +++++++------- src/aec/util/display.py | 8 +-- src/aec/util/docgen.py | 4 +- src/aec/util/ec2.py | 10 ++-- src/aec/util/ec2_types.py | 20 +++---- src/aec/util/tags.py | 4 +- 10 files changed, 107 insertions(+), 113 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 11be95f2..955b677c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", ] @@ -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"] diff --git a/src/aec/command/ami.py b/src/aec/command/ami.py index a6755957..3d150199 100644 --- a/src/aec/command/ami.py +++ b/src/aec/command/ami.py @@ -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 @@ -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] @@ -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)) @@ -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) @@ -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}") @@ -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) @@ -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) diff --git a/src/aec/command/compute_optimizer.py b/src/aec/command/compute_optimizer.py index ed81efc8..96ee5601 100644 --- a/src/aec/command/compute_optimizer.py +++ b/src/aec/command/compute_optimizer.py @@ -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 @@ -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: @@ -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)) diff --git a/src/aec/command/ec2.py b/src/aec/command/ec2.py index 2fab7da4..427d1f21 100755 --- a/src/aec/command/ec2.py +++ b/src/aec/command/ec2.py @@ -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 @@ -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) @@ -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()]) @@ -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. @@ -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)) @@ -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) @@ -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') @@ -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"]: @@ -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) @@ -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("=") @@ -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": @@ -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: @@ -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)) @@ -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)) @@ -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)) @@ -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)) @@ -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)) @@ -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 @@ -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)) @@ -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: diff --git a/src/aec/command/ssm.py b/src/aec/command/ssm.py index 263388bb..eddeab21 100644 --- a/src/aec/command/ssm.py +++ b/src/aec/command/ssm.py @@ -3,7 +3,7 @@ import codecs import sys import uuid -from typing import IO, TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Sequence, TypeVar, Union, cast +from typing import IO, TYPE_CHECKING, Any, Iterator, Sequence, TypeVar, cast import boto3 from botocore.exceptions import ClientError @@ -18,16 +18,16 @@ class Agent(TypedDict): ID: str - Name: Optional[str] + Name: str | None PingStatus: str Platform: str - AgentVersion: Optional[str] + AgentVersion: str | None def describe( config: Config, - ident: Optional[str] = None, - name_match: Optional[str] = None, + ident: str | None = None, + name_match: str | None = None, ) -> Iterator[Agent]: """List running instances with the SSM agent.""" @@ -45,7 +45,7 @@ def describe( else: filters = [] - kwargs: Dict[str, Any] = {"MaxResults": 50, "Filters": filters} + kwargs: dict[str, Any] = {"MaxResults": 50, "Filters": filters} client = boto3.client("ssm", region_name=config.get("region", None)) while True: response = client.describe_instance_information(**kwargs) @@ -67,7 +67,7 @@ def describe( break -def patch_summary(config: Config) -> Iterator[Dict[str, Any]]: +def patch_summary(config: Config) -> Iterator[dict[str, Any]]: """Patch summary for all instances that have run the patch baseline.""" instances_names = describe_instances_names(config) instance_ids = list(instances_names.keys()) @@ -92,7 +92,7 @@ def patch_summary(config: Config) -> Iterator[Dict[str, Any]]: } -def compliance_summary(config: Config) -> List[Dict[str, Any]]: +def compliance_summary(config: Config) -> list[dict[str, Any]]: """Compliance summary for running instances that have run the patch baseline.""" instances_names = describe_instances_names(config) @@ -115,15 +115,15 @@ def compliance_summary(config: Config) -> List[Dict[str, Any]]: def patch( - config: Config, operation: Literal["scan", "install"], idents: List[str], no_reboot: bool -) -> List[Dict[str, Optional[str]]]: + config: Config, operation: Literal["scan", "install"], idents: list[str], no_reboot: bool +) -> list[dict[str, str | None]]: """Scan or install AWS patch baseline.""" instance_ids = fetch_instance_ids(config, idents) client = boto3.client("ssm", region_name=config.get("region", None)) - kwargs: Dict[str, Any] = { + kwargs: dict[str, Any] = { "DocumentName": "AWS-RunPatchBaseline", "InstanceIds": instance_ids, } @@ -159,7 +159,7 @@ def patch( ] -def run(config: Config, idents: List[str]) -> List[Dict[str, Optional[str]]]: +def run(config: Config, idents: list[str]) -> list[dict[str, str | None]]: """ Run a shell script on instance(s). @@ -172,7 +172,7 @@ def run(config: Config, idents: List[str]) -> List[Dict[str, Optional[str]]]: script = sys.stdin.readlines() - kwargs: Dict[str, Any] = { + kwargs: dict[str, Any] = { "DocumentName": "AWS-RunShellScript", "InstanceIds": instance_ids, "Parameters": {"commands": script}, @@ -203,19 +203,19 @@ def run(config: Config, idents: List[str]) -> List[Dict[str, Optional[str]]]: E = TypeVar("E") -def first(xs: Optional[Sequence[E]]) -> Optional[E]: +def first(xs: Sequence[E] | None) -> E | None: if xs: return xs[0] else: return None -def commands(config: Config, ident: Optional[str] = None) -> Iterator[Dict[str, Union[str, int, None]]]: +def commands(config: Config, ident: str | None = None) -> Iterator[dict[str, str | int | None]]: """List commands by instance.""" client = boto3.client("ssm", region_name=config.get("region", None)) - kwargs: Dict[str, Any] = {"MaxResults": 50} + kwargs: dict[str, Any] = {"MaxResults": 50} if ident: kwargs["InstanceId"] = fetch_instance_id(config, ident) @@ -243,7 +243,7 @@ def commands(config: Config, ident: Optional[str] = None) -> Iterator[Dict[str, break -def invocations(config: Config, command_id: str) -> List[Dict[str, Any]]: +def invocations(config: Config, command_id: str) -> list[dict[str, Any]]: """List invocations of a command across instances.""" client = boto3.client("ssm", region_name=config.get("region", None)) @@ -323,12 +323,12 @@ def fetch_instance_id(config: Config, ident: str) -> str: raise ValueError(f"No instance named {ident}") from None -def fetch_instance_ids(config: Config, idents: List[str]) -> List[str]: +def fetch_instance_ids(config: Config, idents: list[str]) -> list[str]: if idents == ["all"]: return [i["ID"] for i in describe(config)] - ids: List[str] = [] - names: List[str] = [] + ids: list[str] = [] + names: list[str] = [] for i in idents: if i.startswith("i-"): @@ -349,7 +349,7 @@ def fetch_instance_ids(config: Config, idents: List[str]) -> List[str]: return ids -def name_filters(idents: Optional[List[str]] = None) -> List[InstanceInformationStringFilterTypeDef]: +def name_filters(idents: list[str] | None = None) -> list[InstanceInformationStringFilterTypeDef]: if idents and idents[0].startswith("i-"): return [{"Key": "InstanceIds", "Values": idents}] elif idents: diff --git a/src/aec/util/display.py b/src/aec/util/display.py index f9dc9de2..acd43296 100644 --- a/src/aec/util/display.py +++ b/src/aec/util/display.py @@ -4,7 +4,7 @@ import enum import json import sys -from typing import Any, Dict, Iterable, Iterator, List, Optional, Sequence, cast +from typing import Any, Iterable, Iterator, List, Sequence, cast from rich import box from rich.console import Console @@ -17,7 +17,7 @@ class OutputFormat(enum.Enum): csv = "csv" -def as_table(dicts: Sequence[Dict[str, Any]], keys: Optional[List[str]] = None) -> List[List[Optional[str]]]: +def as_table(dicts: Sequence[dict[str, Any]], keys: list[str] | None = None) -> list[list[str | None]]: """ Converts a list of dictionaries to a list of lists (table), ordered by specified keys. @@ -33,12 +33,12 @@ def as_table(dicts: Sequence[Dict[str, Any]], keys: Optional[List[str]] = None) return [keys] + [[str(d.get(f, "")) if d.get(f, "") else None for f in keys] for d in dicts] # type: ignore -def as_strings(values: Iterable[Any]) -> List[str]: +def as_strings(values: Iterable[Any]) -> list[str]: return [str(v) if v else "" for v in values] def pretty_print( - result: List[Dict[str, Any]] | Iterator[Dict[str, Any]] | Dict | str | None, + result: list[dict[str, Any]] | Iterator[dict[str, Any]] | dict | str | None, output_format: OutputFormat = OutputFormat.table, ) -> None: """print results as table/csv/json.""" diff --git a/src/aec/util/docgen.py b/src/aec/util/docgen.py index 7e91ebc4..779c837b 100644 --- a/src/aec/util/docgen.py +++ b/src/aec/util/docgen.py @@ -2,7 +2,7 @@ import io from contextlib import redirect_stdout -from typing import Any, Dict, Iterator, List +from typing import Any, Iterator from moto import mock_ec2 from moto.ec2 import ec2_backends @@ -32,7 +32,7 @@ def docs( cmd_name: str, - result: List[Dict[str, Any]] | Iterator[Dict[str, Any]] | Dict | str | None, + result: list[dict[str, Any]] | Iterator[dict[str, Any]] | dict | str | None, ) -> str: capture = io.StringIO() with redirect_stdout(capture): diff --git a/src/aec/util/ec2.py b/src/aec/util/ec2.py index cfd98578..38a4ecd1 100644 --- a/src/aec/util/ec2.py +++ b/src/aec/util/ec2.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Dict, Optional, Sequence +from typing import Sequence import boto3 @@ -9,14 +9,12 @@ from aec.util.ec2_types import DescribeArgs -def describe_running_instances_names(config: Config) -> Dict[str, Optional[str]]: +def describe_running_instances_names(config: Config) -> dict[str, str | None]: # 2x speed up (8 -> 4 secs) compared to listing all names return describe_instances_names(config, {"instance-state-name": ["running"]}) -def describe_instances_names( - config: Config, filters: Optional[Dict[str, Sequence[str]]] = None -) -> Dict[str, Optional[str]]: +def describe_instances_names(config: Config, filters: dict[str, Sequence[str]] | None = None) -> dict[str, str | None]: """Map of EC2 instance ids to names in the region.""" ec2_client = boto3.client("ec2", region_name=config.get("region", None)) @@ -24,7 +22,7 @@ def describe_instances_names( if filters: kwargs["Filters"] = [{"Name": k, "Values": v} for k, v in filters.items()] - instances: Dict[str, Optional[str]] = {} + instances: dict[str, str | None] = {} while True: response = ec2_client.describe_instances(**kwargs) diff --git a/src/aec/util/ec2_types.py b/src/aec/util/ec2_types.py index 04bf0145..f7bec026 100644 --- a/src/aec/util/ec2_types.py +++ b/src/aec/util/ec2_types.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TYPE_CHECKING, List, Sequence +from typing import TYPE_CHECKING, Sequence if TYPE_CHECKING: from mypy_boto3_ec2.literals import InstanceTypeType, ShutdownBehaviorType @@ -35,18 +35,18 @@ class _RequiredRunArgs(TypedDict, total=True): class RunArgs(_RequiredRunArgs, total=False): - BlockDeviceMappings: List[BlockDeviceMappingTypeDef] + BlockDeviceMappings: list[BlockDeviceMappingTypeDef] ImageId: str InstanceType: InstanceTypeType Ipv6AddressCount: int - Ipv6Addresses: List[InstanceIpv6AddressTypeDef] + Ipv6Addresses: list[InstanceIpv6AddressTypeDef] KernelId: str KeyName: str Monitoring: RunInstancesMonitoringEnabledTypeDef Placement: PlacementTypeDef RamdiskId: str - SecurityGroupIds: List[str] - SecurityGroups: List[str] + SecurityGroupIds: list[str] + SecurityGroups: list[str] SubnetId: str UserData: str AdditionalInfo: str @@ -56,18 +56,18 @@ class RunArgs(_RequiredRunArgs, total=False): EbsOptimized: bool IamInstanceProfile: IamInstanceProfileSpecificationTypeDef InstanceInitiatedShutdownBehavior: ShutdownBehaviorType - NetworkInterfaces: List[InstanceNetworkInterfaceSpecificationTypeDef] + NetworkInterfaces: list[InstanceNetworkInterfaceSpecificationTypeDef] PrivateIpAddress: str - ElasticGpuSpecification: List[ElasticGpuSpecificationTypeDef] - ElasticInferenceAccelerators: List[ElasticInferenceAcceleratorTypeDef] - TagSpecifications: List[TagSpecificationTypeDef] + ElasticGpuSpecification: list[ElasticGpuSpecificationTypeDef] + ElasticInferenceAccelerators: list[ElasticInferenceAcceleratorTypeDef] + TagSpecifications: list[TagSpecificationTypeDef] LaunchTemplate: LaunchTemplateSpecificationTypeDef InstanceMarketOptions: InstanceMarketOptionsRequestTypeDef CreditSpecification: CreditSpecificationRequestTypeDef CpuOptions: CpuOptionsRequestTypeDef CapacityReservationSpecification: CapacityReservationSpecificationTypeDef HibernationOptions: HibernationOptionsRequestTypeDef - LicenseSpecifications: List[LicenseConfigurationRequestTypeDef] + LicenseSpecifications: list[LicenseConfigurationRequestTypeDef] MetadataOptions: InstanceMetadataOptionsRequestTypeDef EnclaveOptions: EnclaveOptionsRequestTypeDef diff --git a/src/aec/util/tags.py b/src/aec/util/tags.py index a928cce9..426bdaa8 100644 --- a/src/aec/util/tags.py +++ b/src/aec/util/tags.py @@ -1,11 +1,11 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING if TYPE_CHECKING: from mypy_boto3_ec2.type_defs import ImageTypeDef, InstanceTypeDef, VolumeTypeDef -def get_value(resource: ImageTypeDef | InstanceTypeDef | VolumeTypeDef, key: str) -> Optional[str]: +def get_value(resource: ImageTypeDef | InstanceTypeDef | VolumeTypeDef, key: str) -> str | None: tag_value = [t["Value"] for t in resource.get("Tags", []) if t["Key"] == key] return tag_value[0] if tag_value else None