"""
The AWS cloud provider.
:class:`AWSProvider` replaces ``smarter.common.helpers.aws_helpers.AWSInfrastructureConfig``. It
authenticates with AWS, with ``smarter_settings``' AWS credentials, and implements:
- DNS, with Route53: :class:`~smarter.apps.infrastructure.providers.aws.dns.Route53DNSService`.
- TLS certificates, with ACM:
:class:`~smarter.apps.infrastructure.providers.aws.certificates.ACMCertificateService`.
- the Kubernetes cluster's kubeconfig and description, with EKS.
Of the AWS services that the old helpers wrapped, only these are used by the platform. API
Gateway, DynamoDB, IAM, Lambda, RDS, Rekognition and S3 were not, and were removed.
In the unit tests, it refuses to reach AWS, unless ``allow_in_tests`` is True, because the
Smarter containers' AWS credentials are real.
"""
from typing import Any, Optional
import boto3
from smarter.lib import logging
from smarter.lib.django.waffle import SmarterWaffleSwitches
from ...const import CloudProviders
from ...exceptions import InfrastructureConfigurationError
from ...services.base import refuse_in_unit_tests
from ..base import CloudProvider
from .certificates import ACMCertificateService
from .dns import Route53DNSService
from .helpers.base import AWSBase
from .helpers.eks import AWSEks
logger = logging.getSmarterLogger(__name__, any_switches=[SmarterWaffleSwitches.INFRASTRUCTURE_LOGGING])
[docs]
class AWSProvider(CloudProvider):
"""
Amazon Web Services.
:param allow_in_tests: Allow it to reach AWS in the unit tests, e.g. in a test tagged
:data:`~smarter.lib.unittest.runner.INFRASTRUCTURE`.
"""
name = CloudProviders.AWS
[docs]
def __init__(self, allow_in_tests: bool = False, **kwargs):
super().__init__(allow_in_tests=allow_in_tests, **kwargs)
self._base: Optional[AWSBase] = None
self._eks: Optional[AWSEks] = None
self._dns: Optional[Route53DNSService] = None
self._certificates: Optional[ACMCertificateService] = None
[docs]
def require_live(self) -> None:
"""
Refuse to reach AWS from the unit tests, unless allowed.
:raises InfrastructureConfigurationError: In the unit tests, unless allowed.
"""
refuse_in_unit_tests("AWS", self.allow_in_tests)
@property
def base(self) -> AWSBase:
"""The low-level AWS helper that authenticates, created when it is first used."""
self.require_live()
if self._base is None:
self._base = AWSBase()
return self._base
@property
def session(self) -> Optional[boto3.Session]:
"""
The boto3 session, for AWS-specific backends that need clients of their own.
e.g. :class:`~smarter.apps.llmhost.services.nodegroups.EKSNodeGroupBackend`.
"""
return self.base.aws_session
@property
def eks(self) -> AWSEks:
"""The low-level EKS helper."""
self.require_live()
if self._eks is None:
self._eks = AWSEks()
return self._eks
@property
def identity(self) -> Optional[dict[str, Any]]:
try:
return self.base.identity
except InfrastructureConfigurationError:
return None
@property
def ready(self) -> bool:
try:
self.require_live()
except InfrastructureConfigurationError:
return False
try:
identity = self.base.identity
# pylint: disable=broad-except
except Exception as e:
return self.authentication_state(None, error=str(e))
return self.authentication_state(identity, error="could not fetch the AWS identity")
@property
def account_id(self) -> Optional[str]:
identity = self.identity
return identity.get("Account") if isinstance(identity, dict) else None
@property
def sdk_version(self) -> str:
return boto3.__version__
@property
def dns(self) -> Route53DNSService:
if self._dns is None:
self._dns = Route53DNSService(self)
return self._dns
@property
def certificates(self) -> ACMCertificateService:
if self._certificates is None:
self._certificates = ACMCertificateService(self, self.dns)
return self._certificates
[docs]
def update_kubeconfig(self) -> bool:
if not self.ready:
logger.warning("%s AWS is not ready, so the kubeconfig cannot be updated.", self.formatted_class_name)
return False
return self.eks.update_kubeconfig()
[docs]
def get_kubernetes_cluster_info(self) -> dict[str, Any]:
self.require_ready()
with self.operation("get_kubernetes_cluster_info"):
return self.eks.get_kubernetes_info()
__all__ = ["AWSProvider"]