diff --git a/docs/source/cfngin/lookups/index.rst b/docs/source/cfngin/lookups/index.rst index 3243a150d..54630d39e 100644 --- a/docs/source/cfngin/lookups/index.rst +++ b/docs/source/cfngin/lookups/index.rst @@ -119,10 +119,10 @@ If using boto3 in a lookup, use :meth:`context.get_session() str: """Do something. diff --git a/runway/cfngin/lookups/handlers/ami.py b/runway/cfngin/lookups/handlers/ami.py index f709fe129..2b3ca2bd9 100644 --- a/runway/cfngin/lookups/handlers/ami.py +++ b/runway/cfngin/lookups/handlers/ami.py @@ -1,6 +1,5 @@ """AMI lookup.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import operator @@ -59,14 +58,14 @@ def __init__(self, search_string: str) -> None: super().__init__(f"Unable to find ec2 image with search string: {search_string}") -class AmiLookup(LookupHandler): +class AmiLookup(LookupHandler["CfnginContext"]): """AMI lookup.""" TYPE_NAME: ClassVar[str] = "ami" """Name that the Lookup is registered as.""" @classmethod - def parse(cls, value: str) -> tuple[str, dict[str, str]]: + def parse_query(cls, value: str) -> tuple[str, dict[str, str]]: """Parse the value passed to the lookup. This overrides the default parsing to account for special requirements. @@ -93,7 +92,7 @@ def parse(cls, value: str) -> tuple[str, dict[str, str]]: return args.pop("name_regex"), args @classmethod - def handle(cls, value: str, context: CfnginContext, *__args: Any, **__kwargs: Any) -> str: + def handle(cls, value: str, context: CfnginContext, **_kwargs: Any) -> str: """Fetch the most recent AMI Id using a filter. Args: @@ -113,7 +112,7 @@ def handle(cls, value: str, context: CfnginContext, *__args: Any, **__kwargs: An AMI lookup. """ - query, raw_args = cls.parse(value) + query, raw_args = cls.parse_query(value) args = ArgsDataModel.model_validate(raw_args) ec2 = context.get_session(region=args.region).client("ec2") diff --git a/runway/cfngin/lookups/handlers/awslambda.py b/runway/cfngin/lookups/handlers/awslambda.py index e34aa7b33..69b2f01bf 100644 --- a/runway/cfngin/lookups/handlers/awslambda.py +++ b/runway/cfngin/lookups/handlers/awslambda.py @@ -20,20 +20,19 @@ from ....lookups.handlers.base import LookupHandler from ....utils import load_object_from_string -from ...exceptions import CfnginOnlyLookupError if TYPE_CHECKING: from ....config import CfnginConfig from ....config.models.cfngin import CfnginHookDefinitionModel - from ....context import CfnginContext, RunwayContext + from ....context import CfnginContext from ...hooks.awslambda.base_classes import AwsLambdaHook from ...hooks.awslambda.models.responses import AwsLambdaHookDeployResponse LOGGER = logging.getLogger(__name__) -class AwsLambdaLookup(LookupHandler): +class AwsLambdaLookup(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda" @@ -112,11 +111,7 @@ def get_required_hook_definition( @classmethod def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *_args: Any, - **_kwargs: Any, + cls, value: str, context: CfnginContext, **_kwargs: Any ) -> AwsLambdaHookDeployResponse: """Retrieve metadata for an AWS Lambda deployment package. @@ -129,12 +124,8 @@ def handle( data model. """ - # `if isinstance(context, _RunwayContext)` without needing to import candidate - # importing candidate causes cyclic import - if "RunwayContext" in type(context).__name__: - raise CfnginOnlyLookupError(cls.TYPE_NAME) query, _ = cls.parse(value) - return cls.get_deployment_package_data(cast("CfnginContext", context), query) + return cls.get_deployment_package_data(context, query) @staticmethod def init_hook_class( @@ -167,19 +158,13 @@ def init_hook_class( ) return cast("AwsLambdaHook[Any]", kls(context, **hook_def.args)) - class Code(LookupHandler): + class Code(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.Code" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, - ) -> Code: + def handle(cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any) -> Code: """Retrieve metadata for an AWS Lambda deployment package. Args: @@ -201,19 +186,13 @@ def handle( ) ) - class CodeSha256(LookupHandler): + class CodeSha256(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.CodeSha256" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, - ) -> str: + def handle(cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any) -> str: """Retrieve metadata for an AWS Lambda deployment package. Args: @@ -229,18 +208,14 @@ def handle( """ return AwsLambdaLookup.handle(value, context, *args, **kwargs).code_sha256 - class CompatibleArchitectures(LookupHandler): + class CompatibleArchitectures(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.CompatibleArchitectures" @classmethod def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, + cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any ) -> list[str] | None: """Retrieve metadata for an AWS Lambda deployment package. @@ -261,19 +236,13 @@ def handle( **lookup_args, ) - class CompatibleRuntimes(LookupHandler): + class CompatibleRuntimes(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.CompatibleRuntimes" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, - ) -> Any: + def handle(cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any) -> Any: """Retrieve metadata for an AWS Lambda deployment package. Args: @@ -293,19 +262,13 @@ def handle( **lookup_args, ) - class Content(LookupHandler): + class Content(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.Content" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, - ) -> Content: + def handle(cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any) -> Content: """Retrieve metadata for an AWS Lambda deployment package. Args: @@ -327,18 +290,14 @@ def handle( ) ) - class LicenseInfo(LookupHandler): + class LicenseInfo(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.LicenseInfo" @classmethod def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, + cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any ) -> str | None: """Retrieve metadata for an AWS Lambda deployment package. @@ -359,19 +318,13 @@ def handle( **lookup_args, ) - class Runtime(LookupHandler): + class Runtime(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.Runtime" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, - ) -> str: + def handle(cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any) -> str: """Retrieve metadata for an AWS Lambda deployment package. Args: @@ -387,19 +340,13 @@ def handle( """ return AwsLambdaLookup.handle(value, context, *args, **kwargs).runtime - class S3Bucket(LookupHandler): + class S3Bucket(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.S3Bucket" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, - ) -> str: + def handle(cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any) -> str: """Retrieve metadata for an AWS Lambda deployment package. Args: @@ -416,19 +363,13 @@ def handle( """ return AwsLambdaLookup.handle(value, context, *args, **kwargs).bucket_name - class S3Key(LookupHandler): + class S3Key(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.S3Key" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, - ) -> str: + def handle(cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any) -> str: """Retrieve metadata for an AWS Lambda deployment package. Args: @@ -445,18 +386,14 @@ def handle( """ return AwsLambdaLookup.handle(value, context, *args, **kwargs).object_key - class S3ObjectVersion(LookupHandler): + class S3ObjectVersion(LookupHandler["CfnginContext"]): """Lookup for AwsLambdaHook responses.""" TYPE_NAME: ClassVar[str] = "awslambda.S3ObjectVersion" @classmethod def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *args: Any, - **kwargs: Any, + cls, value: str, context: CfnginContext, *args: Any, **kwargs: Any ) -> str | None: """Retrieve metadata for an AWS Lambda deployment package. diff --git a/runway/cfngin/lookups/handlers/default.py b/runway/cfngin/lookups/handlers/default.py index 413dc38ae..fe96906a1 100644 --- a/runway/cfngin/lookups/handlers/default.py +++ b/runway/cfngin/lookups/handlers/default.py @@ -1,6 +1,5 @@ """Lookup to provide a default value.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations from typing import TYPE_CHECKING, Any, ClassVar @@ -12,7 +11,7 @@ from ....context import CfnginContext -class DefaultLookup(LookupHandler): +class DefaultLookup(LookupHandler["CfnginContext"]): """Lookup to provide a default value.""" TYPE_NAME: ClassVar[str] = "default" diff --git a/runway/cfngin/lookups/handlers/dynamodb.py b/runway/cfngin/lookups/handlers/dynamodb.py index 22c7ce34c..f980972a5 100644 --- a/runway/cfngin/lookups/handlers/dynamodb.py +++ b/runway/cfngin/lookups/handlers/dynamodb.py @@ -1,6 +1,5 @@ """DynamoDB lookup.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import re @@ -87,7 +86,7 @@ def item_key(self) -> dict[str, AttributeValueTypeDef]: } -class DynamodbLookup(LookupHandler): +class DynamodbLookup(LookupHandler["CfnginContext"]): """DynamoDB lookup.""" TYPE_NAME: ClassVar[str] = "dynamodb" diff --git a/runway/cfngin/lookups/handlers/envvar.py b/runway/cfngin/lookups/handlers/envvar.py index a22ec5f81..8f2989a09 100644 --- a/runway/cfngin/lookups/handlers/envvar.py +++ b/runway/cfngin/lookups/handlers/envvar.py @@ -1,6 +1,5 @@ """Environment variable lookup.""" -# pyright: reportIncompatibleMethodOverride=none import logging import os from typing import Any, ClassVar @@ -11,7 +10,7 @@ LOGGER = logging.getLogger(__name__) -class EnvvarLookup(LookupHandler): +class EnvvarLookup(LookupHandler[Any]): """Environment variable lookup.""" DEPRECATION_MSG = "envvar Lookup has been deprecated; use the env lookup instead" @@ -19,7 +18,7 @@ class EnvvarLookup(LookupHandler): """Name that the Lookup is registered as.""" @classmethod - def handle(cls, value: str, **_: Any) -> str: + def handle(cls, value: str, *_args: Any, **_: Any) -> str: """Retrieve an environment variable. Args: diff --git a/runway/cfngin/lookups/handlers/file.py b/runway/cfngin/lookups/handlers/file.py index 56df09811..2f5f57c43 100644 --- a/runway/cfngin/lookups/handlers/file.py +++ b/runway/cfngin/lookups/handlers/file.py @@ -1,6 +1,5 @@ """File lookup.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import base64 @@ -47,7 +46,7 @@ def _validate_supported_codec(cls, v: str) -> str: raise ValueError(f"Codec '{v}' must be one of: {', '.join(CODECS)}") -class FileLookup(LookupHandler): +class FileLookup(LookupHandler[Any]): """File lookup.""" TYPE_NAME: ClassVar[str] = "file" @@ -81,7 +80,7 @@ def parse(cls, value: str) -> tuple[str, ParsedArgsTypeDef]: return read_value_from_path(data_or_path), args @classmethod - def handle(cls, value: str, **_: Any) -> Any: + def handle(cls, value: str, *_args: Any, **_kwargs: Any) -> Any: """Translate a filename into the file contents.""" data, raw_args = cls.parse(value) args = ArgsDataModel.model_validate(raw_args) diff --git a/runway/cfngin/lookups/handlers/hook_data.py b/runway/cfngin/lookups/handlers/hook_data.py index c86ac4d6b..38e7ee2a4 100644 --- a/runway/cfngin/lookups/handlers/hook_data.py +++ b/runway/cfngin/lookups/handlers/hook_data.py @@ -1,6 +1,5 @@ """Hook data lookup.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import logging @@ -18,7 +17,7 @@ LOGGER = logging.getLogger(__name__) -class HookDataLookup(LookupHandler): +class HookDataLookup(LookupHandler["CfnginContext"]): """Hook data lookup.""" TYPE_NAME: ClassVar[str] = "hook_data" diff --git a/runway/cfngin/lookups/handlers/kms.py b/runway/cfngin/lookups/handlers/kms.py index b71ecc137..4eafb5f4d 100644 --- a/runway/cfngin/lookups/handlers/kms.py +++ b/runway/cfngin/lookups/handlers/kms.py @@ -1,6 +1,5 @@ """AWS KMS lookup.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import codecs @@ -18,7 +17,7 @@ LOGGER = logging.getLogger(__name__) -class KmsLookup(LookupHandler): +class KmsLookup(LookupHandler["CfnginContext"]): """AWS KMS lookup.""" DEPRECATION_MSG = ( diff --git a/runway/cfngin/lookups/handlers/output.py b/runway/cfngin/lookups/handlers/output.py index c0a2fe9db..575ad3811 100644 --- a/runway/cfngin/lookups/handlers/output.py +++ b/runway/cfngin/lookups/handlers/output.py @@ -1,6 +1,5 @@ """AWS CloudFormation Output lookup.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import logging @@ -28,7 +27,7 @@ class OutputQuery(NamedTuple): output_name: str -class OutputLookup(LookupHandler): +class OutputLookup(LookupHandler["CfnginContext"]): """AWS CloudFormation Output lookup.""" DEPRECATION_MSG = ( diff --git a/runway/cfngin/lookups/handlers/rxref.py b/runway/cfngin/lookups/handlers/rxref.py index 8d96ff7e9..a03db93e2 100644 --- a/runway/cfngin/lookups/handlers/rxref.py +++ b/runway/cfngin/lookups/handlers/rxref.py @@ -1,6 +1,5 @@ """Handler for fetching outputs from a stack in the current namespace.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import logging @@ -19,7 +18,7 @@ LOGGER = logging.getLogger(__name__) -class RxrefLookup(LookupHandler): +class RxrefLookup(LookupHandler["CfnginContext"]): """Rxref lookup.""" DEPRECATION_MSG = ( @@ -42,7 +41,7 @@ def legacy_parse(cls, value: str) -> tuple[OutputQuery, ParsedArgsTypeDef]: return deconstruct(value), {} @classmethod - def handle(cls, value: str, context: CfnginContext, provider: Provider, **_: Any) -> Any: + def handle(cls, value: str, context: CfnginContext, *, provider: Provider, **_: Any) -> Any: """Fetch an output from the designated stack in the current namespace. The ``output`` lookup supports fetching outputs from stacks created diff --git a/runway/cfngin/lookups/handlers/split.py b/runway/cfngin/lookups/handlers/split.py index 74f53f405..c3239359a 100644 --- a/runway/cfngin/lookups/handlers/split.py +++ b/runway/cfngin/lookups/handlers/split.py @@ -1,19 +1,18 @@ """Split lookup.""" -# pyright: reportIncompatibleMethodOverride=none from typing import Any, ClassVar from ....lookups.handlers.base import LookupHandler -class SplitLookup(LookupHandler): +class SplitLookup(LookupHandler[Any]): """Split lookup.""" TYPE_NAME: ClassVar[str] = "split" """Name that the Lookup is registered as.""" @classmethod - def handle(cls, value: str, **_: Any) -> list[str]: + def handle(cls, value: str, *_args: Any, **_kwargs: Any) -> list[str]: """Split the supplied string on the given delimiter, providing a list. Args: diff --git a/runway/cfngin/lookups/handlers/xref.py b/runway/cfngin/lookups/handlers/xref.py index 2143e1a06..85bf02b75 100644 --- a/runway/cfngin/lookups/handlers/xref.py +++ b/runway/cfngin/lookups/handlers/xref.py @@ -1,6 +1,5 @@ """Handler for fetching outputs from fully qualified stacks.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import logging @@ -18,7 +17,7 @@ XREF_PERSISTENT_STATE = {"has_warned": False} -class XrefLookup(LookupHandler): +class XrefLookup(LookupHandler[Any]): """Xref lookup.""" DEPRECATION_MSG = "xref Lookup has been deprecated; use the cfn lookup instead" @@ -26,7 +25,7 @@ class XrefLookup(LookupHandler): """Name that the Lookup is registered as.""" @classmethod - def handle(cls, value: str, provider: Provider, **_: Any) -> str: + def handle(cls, value: str, *_args: Any, provider: Provider, **_kwargs: Any) -> str: """Fetch an output from the designated, fully qualified stack. The `output` handler supports fetching outputs from stacks created diff --git a/runway/cfngin/lookups/registry.py b/runway/cfngin/lookups/registry.py index bf72124a0..ad4c84edd 100644 --- a/runway/cfngin/lookups/registry.py +++ b/runway/cfngin/lookups/registry.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from typing import cast +from typing import Any, cast from ...lookups.handlers.base import LookupHandler from ...lookups.handlers.cfn import CfnLookup @@ -25,11 +25,13 @@ from .handlers.split import SplitLookup from .handlers.xref import XrefLookup -CFNGIN_LOOKUP_HANDLERS: dict[str, type[LookupHandler]] = {} +CFNGIN_LOOKUP_HANDLERS: dict[str, type[LookupHandler[Any]]] = {} LOGGER = logging.getLogger(__name__) -def register_lookup_handler(lookup_type: str, handler_or_path: str | type[LookupHandler]) -> None: +def register_lookup_handler( + lookup_type: str, handler_or_path: str | type[LookupHandler[Any]] +) -> None: """Register a lookup handler. Args: diff --git a/runway/lookups/handlers/base.py b/runway/lookups/handlers/base.py index eab955b5d..d651c08b5 100644 --- a/runway/lookups/handlers/base.py +++ b/runway/lookups/handlers/base.py @@ -4,8 +4,9 @@ import json import logging +from abc import ABC, abstractmethod from collections.abc import Sequence -from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast +from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypedDict, TypeVar, cast, overload import yaml from troposphere import BaseAWSObject @@ -21,6 +22,12 @@ LOGGER = logging.getLogger(__name__) +ContextTypeVar = TypeVar( + "ContextTypeVar", "CfnginContext", "RunwayContext", "CfnginContext | RunwayContext" +) +"""Type variable for context type.""" + + TransformToTypeLiteral = Literal["bool", "str"] @@ -44,7 +51,7 @@ class ParsedArgsTypeDef(TypedDict, total=False): transform: TransformToTypeLiteral -class LookupHandler: +class LookupHandler(ABC, Generic[ContextTypeVar]): """Base class for lookup handlers.""" TYPE_NAME: ClassVar[str] @@ -115,19 +122,25 @@ def format_results( return value.data return value + @overload @classmethod + @abstractmethod def handle( - cls, - __value: str, - context: CfnginContext | RunwayContext, - *__args: Any, - provider: Provider | None = None, - **__kwargs: Any, - ) -> Any: + cls, value: str, context: ContextTypeVar, *, provider: Provider, **_kwargs: Any + ) -> Any: ... + + @overload + @classmethod + @abstractmethod + def handle(cls, value: str, context: ContextTypeVar, **_kwargs: Any) -> Any: ... + + @classmethod + @abstractmethod + def handle(cls, value: str, context: ContextTypeVar, **_kwargs: Any) -> Any: """Perform the lookup. Args: - __value: Parameter(s) given to the lookup. + value: Parameter(s) given to the lookup. context: The current context object. provider: CFNgin AWS provider. diff --git a/runway/lookups/handlers/cfn.py b/runway/lookups/handlers/cfn.py index 6747ae5d1..7c9d65ff9 100644 --- a/runway/lookups/handlers/cfn.py +++ b/runway/lookups/handlers/cfn.py @@ -6,7 +6,6 @@ """ -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import json @@ -36,7 +35,7 @@ class OutputQuery(NamedTuple): output_name: str -class CfnLookup(LookupHandler): +class CfnLookup(LookupHandler["CfnginContext | RunwayContext"]): """CloudFormation Stack Output lookup.""" TYPE_NAME: ClassVar[str] = "cfn" diff --git a/runway/lookups/handlers/ecr.py b/runway/lookups/handlers/ecr.py index b16624a3f..db471408e 100644 --- a/runway/lookups/handlers/ecr.py +++ b/runway/lookups/handlers/ecr.py @@ -16,7 +16,7 @@ LOGGER = logging.getLogger(__name__) -class EcrLookup(LookupHandler): +class EcrLookup(LookupHandler["CfnginContext | RunwayContext"]): """ECR Lookup.""" TYPE_NAME: ClassVar[str] = "ecr" @@ -33,13 +33,7 @@ def get_login_password(client: ECRClient) -> str: return password @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *__args: Any, - **__kwargs: Any, - ) -> Any: + def handle(cls, value: str, context: CfnginContext | RunwayContext, **_kwargs: Any) -> Any: """Retrieve a value from AWS Elastic Container Registry (ECR). Args: diff --git a/runway/lookups/handlers/env.py b/runway/lookups/handlers/env.py index 6eaa39422..d64fe0c9f 100644 --- a/runway/lookups/handlers/env.py +++ b/runway/lookups/handlers/env.py @@ -1,6 +1,5 @@ """Retrieve a value from an environment variable.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations from typing import TYPE_CHECKING, Any, ClassVar @@ -12,20 +11,14 @@ from ...context import CfnginContext, RunwayContext -class EnvLookup(LookupHandler): +class EnvLookup(LookupHandler["CfnginContext | RunwayContext"]): """Environment variable Lookup.""" TYPE_NAME: ClassVar[str] = "env" """Name that the Lookup is registered as.""" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *__args: Any, - **__kwargs: Any, - ) -> Any: + def handle(cls, value: str, context: CfnginContext | RunwayContext, **_kwargs: Any) -> Any: """Retrieve an environment variable. The value is retrieved from a copy of the current environment variables diff --git a/runway/lookups/handlers/random_string.py b/runway/lookups/handlers/random_string.py index d48aa2599..151405109 100644 --- a/runway/lookups/handlers/random_string.py +++ b/runway/lookups/handlers/random_string.py @@ -1,6 +1,5 @@ """Generate a random string.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import logging @@ -14,7 +13,6 @@ if TYPE_CHECKING: from collections.abc import Sequence - from ...context import CfnginContext, RunwayContext LOGGER = logging.getLogger(__name__) @@ -28,7 +26,7 @@ class ArgsDataModel(BaseModel): uppercase: bool = True -class RandomStringLookup(LookupHandler): +class RandomStringLookup(LookupHandler[Any]): """Random string lookup.""" TYPE_NAME: ClassVar[str] = "random.string" @@ -95,18 +93,11 @@ def ensure_has_one_of(cls, args: ArgsDataModel, value: str) -> bool: return sum(c(value) for c in checks) == len(checks) @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, # noqa: ARG003 - *__args: Any, - **__kwargs: Any, - ) -> Any: + def handle(cls, value: str, *_args: Any, **_kwargs: Any) -> Any: """Generate a random string. Args: value: The value passed to the Lookup. - context: The current context object. Raises: ValueError: Unable to find a value for the provided query and diff --git a/runway/lookups/handlers/ssm.py b/runway/lookups/handlers/ssm.py index 76ccc469e..8537db214 100644 --- a/runway/lookups/handlers/ssm.py +++ b/runway/lookups/handlers/ssm.py @@ -15,20 +15,14 @@ LOGGER = logging.getLogger(__name__) -class SsmLookup(LookupHandler): +class SsmLookup(LookupHandler["CfnginContext | RunwayContext"]): """SSM Parameter Store Lookup.""" TYPE_NAME: ClassVar[str] = "ssm" """Name that the Lookup is registered as.""" @classmethod - def handle( - cls, - value: str, - context: CfnginContext | RunwayContext, - *__args: Any, - **__kwargs: Any, - ) -> Any: + def handle(cls, value: str, context: CfnginContext | RunwayContext, **_kwargs: Any) -> Any: """Retrieve a value from SSM Parameter Store. Args: diff --git a/runway/lookups/handlers/var.py b/runway/lookups/handlers/var.py index 988eabbb1..ee6f98c60 100644 --- a/runway/lookups/handlers/var.py +++ b/runway/lookups/handlers/var.py @@ -1,6 +1,5 @@ """Retrieve a variable from the variables file or definition.""" -# pyright: reportIncompatibleMethodOverride=none from __future__ import annotations import logging @@ -17,14 +16,14 @@ TYPE_NAME = "var" -class VarLookup(LookupHandler): +class VarLookup(LookupHandler[Any]): """Variable definition Lookup.""" TYPE_NAME: ClassVar[str] = "var" """Name that the Lookup is registered as.""" @classmethod - def handle(cls, value: str, *__args: Any, variables: MutableMap, **__kwargs: Any) -> Any: + def handle(cls, value: str, *_args: Any, variables: MutableMap, **_kwargs: Any) -> Any: """Retrieve a variable from the variable definition. The value is retrieved from the variables passed to Runway using diff --git a/runway/lookups/registry.py b/runway/lookups/registry.py index 2c69015f5..4103397ca 100644 --- a/runway/lookups/registry.py +++ b/runway/lookups/registry.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from typing import cast +from typing import Any, cast from ..utils import load_object_from_string from .handlers.base import LookupHandler @@ -14,11 +14,13 @@ from .handlers.ssm import SsmLookup from .handlers.var import VarLookup -RUNWAY_LOOKUP_HANDLERS: dict[str, type[LookupHandler]] = {} +RUNWAY_LOOKUP_HANDLERS: dict[str, type[LookupHandler[Any]]] = {} LOGGER = logging.getLogger(__name__) -def register_lookup_handler(lookup_type: str, handler_or_path: str | type[LookupHandler]) -> None: +def register_lookup_handler( + lookup_type: str, handler_or_path: str | type[LookupHandler[Any]] +) -> None: """Register a lookup handler. Args: diff --git a/runway/templates/k8s-cfn-repo/k8s-workers.cfn/local_lookups/bootstrap_value.py b/runway/templates/k8s-cfn-repo/k8s-workers.cfn/local_lookups/bootstrap_value.py index 41a70884d..98cb3c2aa 100644 --- a/runway/templates/k8s-cfn-repo/k8s-workers.cfn/local_lookups/bootstrap_value.py +++ b/runway/templates/k8s-cfn-repo/k8s-workers.cfn/local_lookups/bootstrap_value.py @@ -29,7 +29,7 @@ class HookArgs(BaseModel): post_bootstrap: str -class BootstrapValue(LookupHandler): +class BootstrapValue(LookupHandler["CfnginContext"]): """Return the bootstrap value on creation otherwise the post_bootstrap. .. rubric:: Example @@ -42,12 +42,7 @@ class BootstrapValue(LookupHandler): @classmethod def handle( - cls, - value: str, - context: CfnginContext, - *_args: Any, - provider: Provider, - **_kwargs: Any, + cls, value: str, context: CfnginContext, *, provider: Provider, **_kwargs: Any ) -> str: """Handle lookup.""" query, raw_args = cls.parse(value) diff --git a/runway/variables.py b/runway/variables.py index e65142810..ce0810bd8 100644 --- a/runway/variables.py +++ b/runway/variables.py @@ -732,7 +732,7 @@ def __repr__(self) -> str: class VariableValueLookup(VariableValue): """A lookup variable value.""" - handler: type[LookupHandler] + handler: type[LookupHandler[Any]] lookup_name: VariableValueLiteral[str] lookup_query: VariableValue @@ -742,7 +742,7 @@ def __init__( self, lookup_name: VariableValueLiteral[str], lookup_query: str | VariableValue, - handler: type[LookupHandler] | None = None, + handler: type[LookupHandler[Any]] | None = None, variable_type: VariableTypeLiteralTypeDef = "cfngin", ) -> None: """Initialize class.