Skip to content

Commit

Permalink
LookupHandler is now a Generic abstract base class (#2524)
Browse files Browse the repository at this point in the history
  • Loading branch information
ITProKyle authored Aug 21, 2024
1 parent 9961015 commit ee36279
Show file tree
Hide file tree
Showing 24 changed files with 96 additions and 188 deletions.
10 changes: 5 additions & 5 deletions docs/source/cfngin/lookups/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -119,10 +119,10 @@ If using boto3 in a lookup, use :meth:`context.get_session() <runway.context.Cfn
if TYPE_CHECKING:
from runway.cfngin.providers.aws.default import Provider
from runway.context import CfnginContext, RunwayContext
from runway.context import CfnginContext
class MylookupLookup(LookupHandler):
class MylookupLookup(LookupHandler["CfnginContext"]):
"""My lookup."""
TYPE_NAME: ClassVar[str] = "my_lookup"
Expand All @@ -132,9 +132,9 @@ If using boto3 in a lookup, use :meth:`context.get_session() <runway.context.Cfn
def handle(
cls,
value: str,
context: CfnginContext | RunwayContext,
*_args: Any,
provider: Provider | None = None,
context: CfnginContext,
*,
provider: Provider,
**_kwargs: Any
) -> str:
"""Do something.
Expand Down
9 changes: 4 additions & 5 deletions runway/cfngin/lookups/handlers/ami.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
"""AMI lookup."""

# pyright: reportIncompatibleMethodOverride=none
from __future__ import annotations

import operator
Expand Down Expand Up @@ -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.
Expand All @@ -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:
Expand All @@ -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")

Expand Down
111 changes: 24 additions & 87 deletions runway/cfngin/lookups/handlers/awslambda.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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.
Expand All @@ -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(
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -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.
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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.
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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.
Expand Down
3 changes: 1 addition & 2 deletions runway/cfngin/lookups/handlers/default.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
"""Lookup to provide a default value."""

# pyright: reportIncompatibleMethodOverride=none
from __future__ import annotations

from typing import TYPE_CHECKING, Any, ClassVar
Expand All @@ -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"
Expand Down
3 changes: 1 addition & 2 deletions runway/cfngin/lookups/handlers/dynamodb.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
"""DynamoDB lookup."""

# pyright: reportIncompatibleMethodOverride=none
from __future__ import annotations

import re
Expand Down Expand Up @@ -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"
Expand Down
5 changes: 2 additions & 3 deletions runway/cfngin/lookups/handlers/envvar.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
"""Environment variable lookup."""

# pyright: reportIncompatibleMethodOverride=none
import logging
import os
from typing import Any, ClassVar
Expand All @@ -11,15 +10,15 @@
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"
TYPE_NAME: ClassVar[str] = "envvar"
"""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:
Expand Down
Loading

0 comments on commit ee36279

Please sign in to comment.