The Service Layer

The infrastructure service layer: the platform’s only way to reach its infrastructure.

infrastructure gives the platform each service, whichever cloud provider implements it:

from smarter.apps.infrastructure.services import infrastructure

zone, created = infrastructure.dns.get_or_create_zone("example.com")
certificate_id, _ = infrastructure.certificates.get_or_create_certificate("example.com")
infrastructure.kubernetes.apply_manifest(manifest)
infrastructure.email.send_email(subject="Hello", body="...", to="user@example.com")

dns: DNSService, implemented by the cloud provider, e.g. AWS Route53.
certificates: CertificateService, implemented by the cloud provider, e.g. AWS Certificate Manager.
kubernetes: KubernetesService, implemented with kubectl, and the cloud provider’s kubeconfig.
email: EmailService, implemented with SMTP.
provider: the CloudProvider of smarter_settings.cloud_provider.

Each service is resolved when it is used, not when this module is imported, so that tests can replace it, with configure_provider(), configure_kubernetes() and configure_email(), and so that importing the platform never reaches the network.

Tests of code that uses infrastructure can also patch it where it is used, e.g. patch("smarter.apps.llmclient.tasks.verify_custom_domain.infrastructure").

class smarter.apps.infrastructure.services.Certificate(id, domain_name, status, validation_records=<factory>)[source]

Bases: object

A TLS certificate, for a domain and its subdomains.

__init__(id, domain_name, status, validation_records=<factory>)

domain_name: str: The certificate’s domain, e.g. example.com.

id: str: The provider’s id of the certificate, e.g. an AWS ACM certificate ARN.

property is_issued: bool

status: str: See CertificateStatus.

validation_records: list[DNSRecord]: The DNS records that prove control of the domain, once the provider has generated them.

class smarter.apps.infrastructure.services.CertificateService(provider_name, dns, *args, **kwargs)[source]

Bases: InfrastructureService

The TLS certificate service of a cloud provider.

Parameters:

provider_name (str) – The name of the provider, e.g. aws.
dns (DNSService) – The DNS service in which to create validation records.

__init__(provider_name, dns, *args, **kwargs)[source]

billable_certificates: bool = False

Whether the provider bills for certificates.

AWS ACM’s public certificates are free.

certificate_status(certificate_id)[source]

Return a certificate’s status, e.g. PENDING_VALIDATION or ISSUED.

See CertificateStatus.

Return type:: str

create_validation_records(certificate_id)[source]

Create the DNS records that validate a certificate, in its domain’s zone.

The zone is created if it does not exist. The provider can only read the records once the domain is delegated to the zone.

Parameters:: certificate_id (str) – The provider’s id of the certificate.
Return type:: list[DNSRecord]
Returns:: The validation records.

delete_certificate(certificate_id)[source]

Delete a certificate.

Parameters:: certificate_id (str) – The provider’s id of the certificate.
Return type:: bool
Returns:: True if it was deleted, False if it did not exist.

error_class: alias of CertificateServiceError

get_certificate(certificate_id)[source]

Return a certificate.

Parameters:: certificate_id (str) – The provider’s id of the certificate.
Raises:: CertificateNotFound – If the certificate does not exist.
Return type:: Certificate

get_certificate_id(domain_name)[source]

Return the id of a domain’s certificate.

Parameters:: domain_name (str) – The certificate’s domain, e.g. example.com.
Return type:: Optional[str]
Returns:: The id, or None if the domain has no certificate.

get_or_create_certificate(domain_name)[source]

Return the id of a domain’s certificate, and request one if it has none.

The certificate covers the domain and its subdomains, e.g. example.com and *.example.com. It is not issued until its validation records exist, see create_validation_records().

Parameters:: domain_name (str) – The certificate’s domain.
Return type:: tuple[str, bool]
Returns:: The certificate’s id, and whether it was requested.

is_issued(certificate_id)[source]

Whether a certificate is issued, i.e. its domain is validated.

Return type:: bool

issue_wait_attempts: int = 20: How many times wait_until_issued() checks the certificate.

issue_wait_seconds: float = 30: Seconds between the checks.

service_name: str = 'certificates': The name of the service in signals and logs, see InfrastructureServiceNames.

validation_wait_attempts: int = 120: How many times to look for a new certificate’s validation records, before CertificateTimeout.

validation_wait_seconds: float = 5

Seconds between the attempts.

A provider generates validation records in seconds.

wait_for_validation_records(certificate_id)[source]

Return a certificate once its provider has generated its validation records.

Parameters:

certificate_id (str) – The provider’s id of the certificate.

Raises:

CertificateTimeout – If the records are not generated in time.
CertificateNotFound – If the certificate does not exist, after the last attempt.

Return type:

Certificate

wait_until_issued(certificate_id)[source]

Wait for a certificate to be issued.

Parameters:: certificate_id (str) – The provider’s id of the certificate.
Return type:: bool
Returns:: True if it is issued, False if it is not after issue_wait_attempts checks.

class smarter.apps.infrastructure.services.DNSRecord(name, type, ttl=None, values=<factory>, alias=None)[source]

Bases: object

A DNS record set: the values of one name and type.

__init__(name, type, ttl=None, values=<factory>, alias=None)

alias: dict[str, Any] | None = None: A provider-specific alias target, e.g. an AWS load balancer, in place of values.

name: str: The record’s name, e.g. api.example.com.

same_target(other)[source]

Whether two records point at the same values, or the same alias.

Return type:: bool

ttl: int | None = None

Seconds.

None for an alias record.

type: str: The record’s type, e.g. A, CNAME, NS or TXT.

values: list[str]: The record’s values, e.g. IP addresses.

class smarter.apps.infrastructure.services.DNSService(provider_name, *args, **kwargs)[source]

Bases: InfrastructureService

The DNS service of a cloud provider.

Parameters:: provider_name (str) – The name of the provider, e.g. aws.

billable_zones: bool = True: Whether the provider bills for zones, e.g. AWS Route53 bills each hosted zone monthly.

create_domain_a_record(hostname, api_host_domain, zone_id=None)[source]

Point a host at the same target as a parent domain, by copying the parent’s A record.

e.g. an LLMClient’s host, example.3141-5926-5359.api.smarter.sh, points at the load balancer of api.smarter.sh.

Parameters:

hostname (str) – The host, e.g. example.3141-5926-5359.api.smarter.sh.
api_host_domain (str) – The parent domain whose A record is copied, e.g. api.smarter.sh. The record is created in its zone, unless zone_id is given.
zone_id (Optional[str]) – The zone in which to create the record, e.g. a custom domain’s.

Return type:

tuple[DNSRecord, bool]

Returns:

The record, and whether it was created.

Raises:

DNSZoneNotFound – If the parent domain has no A record.

delete_record(zone_id, name, record_type)[source]

Delete a record.

Parameters:

zone_id (str) – The provider’s id of the zone.
name (str) – The record’s name.
record_type (str) – The record’s type.

Return type:

bool

Returns:

True if the record was deleted, False if it did not exist.

delete_zone(domain)[source]

Delete the zone of a domain, and all of its records.

This cannot be undone.

Parameters:: domain (str) – The zone’s domain.
Return type:: bool
Returns:: True if the zone was deleted, False if it did not exist.

property environment_api_domain: str

The environment’s API domain, as it exists in DNS, e.g. local.api.example.com.

In the local environment, smarter_settings.environment_api_domain is a localhost domain, which DNS cannot serve, so this is its proxy domain.

error_class: alias of DNSServiceError

get_environment_a_record(domain=None)[source]

Return the A record of a domain, in its own zone: by default, the environment’s domain.

The A record of a Smarter environment’s domain points at its load balancer. The platform copies it to the hosts that it serves, e.g. each LLMClient’s.

Parameters:: domain (Optional[str]) – The domain, by default smarter_settings.environment_platform_domain.
Return type:: Optional[DNSRecord]
Returns:: The A record, or None if the domain has no zone or no A record.

get_name_servers(zone_id)[source]

Return the name servers of a zone, e.g. for a customer to delegate their domain to.

Parameters:: zone_id (str) – The provider’s id of the zone.
Return type:: list[str]
Returns:: The name servers, without trailing dots.
Raises:: DNSZoneNotFound – If the zone does not exist.

get_or_create_record(zone_id, name, record_type, ttl=600, values=None, alias=None)[source]

Return a record, and create it, or update its values, if it does not match.

Parameters:

zone_id (str) – The provider’s id of the zone.
name (str) – The record’s name.
record_type (str) – The record’s type.
ttl (Optional[int]) – Seconds. Ignored for an alias record.
values (Optional[list[str]]) – The record’s values, e.g. IP addresses.
alias (Optional[dict[str, Any]]) – A provider-specific alias target, in place of values.

Return type:

tuple[DNSRecord, bool]

Returns:

The record, and whether it was created, rather than found or updated.

Raises:

DNSRecordTimeout – If the record does not appear in the zone in time.

get_or_create_zone(domain)[source]

Return the zone of a domain, and create it if it does not exist.

A new zone is billable in most clouds, so it is announced with billable_resource_creating and billable_resource_created.

Parameters:: domain (str) – The zone’s domain, e.g. example.com.
Return type:: tuple[DNSZone, bool]
Returns:: The zone, and whether it was created.

get_record(zone_id, name, record_type)[source]

Return a record of a zone.

Parameters:

zone_id (str) – The provider’s id of the zone.
name (str) – The record’s name, e.g. api.example.com.
record_type (str) – The record’s type, e.g. A.

Return type:

Optional[DNSRecord]

Returns:

The record, or None if it does not exist.

get_zone(domain)[source]

Return the zone of a domain.

Parameters:: domain (str) – The zone’s domain, e.g. example.com.
Return type:: Optional[DNSZone]
Returns:: The zone, or None if it does not exist.

get_zone_by_id(zone_id)[source]

Return a zone by its id.

Parameters:: zone_id (str) – The provider’s id of the zone.
Return type:: Optional[DNSZone]
Returns:: The zone, with its name servers, or None if it does not exist.

list_records(zone_id)[source]

Return a zone’s records.

Return type:: list[DNSRecord]

record_wait_attempts: int = 10: How many times to look for a new record, before DNSRecordTimeout.

record_wait_seconds: float = 15: Seconds between the attempts.

resolve_domain(domain)[source]

Validate a domain, and replace a local environment’s API domain with its proxy domain.

Parameters:: domain (str) – A domain, e.g. example.api.localhost:9357.
Return type:: str
Returns:: The domain as it exists in DNS.
Raises:: SmarterValueError – If the domain is invalid, or is a local host.

resolve_record_name(name)[source]

Resolve a record’s name, as resolve_domain() does, but without validating it as a host name.

Record names may contain labels that host names may not, e.g. _acme-challenge.example.com.

Return type:: str

service_name: str = 'dns': The name of the service in signals and logs, see InfrastructureServiceNames.

class smarter.apps.infrastructure.services.DNSZone(id, name, name_servers=<factory>)[source]

Bases: object

A DNS zone, e.g. an AWS Route53 hosted zone.

__init__(id, name, name_servers=<factory>)

id: str: The provider’s id of the zone, e.g. Z148QEXAMPLE8V.

name: str: The zone’s domain, e.g. example.com.

name_servers: list[str]: The zone’s authoritative name servers, which a parent domain delegates the zone to.

class smarter.apps.infrastructure.services.EmailService(provider_name, *args, **kwargs)[source]

Bases: InfrastructureService

The platform’s outgoing email.

static admin_bcc(recipients)[source]

The blind copy to the platform’s admin, smarter_settings.email_admin.

Parameters:: recipients (list[str]) – The email’s recipients, who need no copy.
Return type:: list[str]
Returns:: The admin’s address, or nothing if there is no valid admin address, or the admin is a recipient.

error_class: alias of EmailServiceError

message(subject, body, to, html=False, from_email=None)[source]

Return a message, from smarter_settings.smtp_from_email unless from_email is given.

It has no Bcc header: blind copies are added to the envelope by _deliver().

Return type:: MIMEMultipart

send_email(subject, body, to, html=False, from_email=None, quiet=False, bcc_admin=True)[source]

Send an email.

Failures are logged, and announced with email_failed, rather than raised, so that email never breaks the request that sends it.

Parameters:

subject (str) – The subject.
body (str) – The body, HTML if html is True.
to (Union[str, List[str]]) – The recipient, or recipients. Invalid addresses are dropped.
html (bool) – True if the body is HTML.
from_email (Optional[str]) – The sender, by default smarter_settings.smtp_from_email.
quiet (bool) – True to only log what would have been sent.
bcc_admin (bool) – Send a blind copy to the platform’s admin, smarter_settings.email_admin. Pass False for an email that carries a secret, e.g. a password reset link, which only its recipient may see.

Return type:

bool

Returns:

True if the email was sent.

service_name: str = 'email': The name of the service in signals and logs, see InfrastructureServiceNames.

static validate_mail_list(emails, quiet=False)[source]

Return the valid email addresses of a list.

Parameters:

emails (Union[str, List[str]]) – An email address, or a list of them.
quiet (bool) – True to not log invalid addresses.

Return type:

Optional[List[str]]

Returns:

The valid addresses, or None if there are none.

class smarter.apps.infrastructure.services.InMemoryEmailService(ready=True, **kwargs)[source]

Bases: EmailService

Email that is kept in outbox, for tests and local development.

Parameters:: ready (bool) – Whether the service is ready.

__init__(ready=True, **kwargs)[source]

fail: str | None: Set to make delivery fail with this error.

outbox: list[SentEmail]

property ready: bool: Whether the service is authenticated and connected, i.e. it can be used.

class smarter.apps.infrastructure.services.InfrastructureService(provider_name, *args, **kwargs)[source]

Bases: ABC, SmarterHelperMixin

The base class of every infrastructure service.

Parameters:: provider_name (str) – The name of the cloud provider that implements the service, e.g. aws. Services that do not depend on a cloud, e.g. SMTP email, use their protocol’s name.

__init__(provider_name, *args, **kwargs)[source]

connection_state(connected, error=None)[source]

Record whether the service is connected to its backend.

Sends infrastructure_connected or infrastructure_connection_failed when the state changes, rather than on every check.

Parameters:

connected (bool) – Whether the service is connected.
error (Optional[str]) – Why it is not connected.

Return type:

bool

Returns:

connected.

created_resource(resource, resource_id=None)[source]

Send the signal that follows the creation of a resource.

Return type:: None

creating_resource(resource_type, resource_name, billable=False, **kwargs)[source]

Send the signal that precedes the creation of a resource.

Only billable resources have a “creating” signal, so that what costs money can be stopped or audited before it exists.

Return type:: dict[str, Any]
Returns:: The resource’s signal arguments, for created_resource().

destroyed_resource(resource)[source]

Send the signal that follows the destruction of a resource.

Return type:: None

destroying_resource(resource_type, resource_name, resource_id=None, billable=False)[source]

Send the signal that precedes the destruction of a resource.

Return type:: dict[str, Any]
Returns:: The resource’s signal arguments, for destroyed_resource().

error_class: alias of SmarterInfrastructureError

operation(name)[source]

Run one of the service’s operations, translating its errors.

An infrastructure error passes through. Any other error, e.g. a cloud SDK’s, is raised as the service’s error_class. Either way, infrastructure_operation_failed is sent.

with self.operation("create_zone"):
    response = self.client.create_hosted_zone(...)

Parameters:: name (str) – The operation’s name, e.g. create_zone.
Return type:: Iterator[None]

abstract property ready: bool: Whether the service is authenticated and connected, i.e. it can be used.

require_ready()[source]

Raise unless the service is ready.

Raises:: InfrastructureNotReadyError – If the service is not ready.
Return type:: None

send(signal, **kwargs)[source]

Send one of the infrastructure signals, with this service’s service and provider.

Return type:: None

service_name: str = 'service': The name of the service in signals and logs, see InfrastructureServiceNames.

class smarter.apps.infrastructure.services.InfrastructureServices[source]

Bases: object

The platform’s infrastructure services.

Use the infrastructure instance.

property certificates: CertificateService: The cloud provider’s TLS certificates.

property dns: DNSService: The cloud provider’s DNS.

property email: EmailService: The platform’s outgoing email.

property kubernetes: KubernetesService: The platform’s Kubernetes cluster.

property provider: CloudProvider: The cloud provider, smarter_settings.cloud_provider, unless one is configured.

property ready: bool: Whether the cloud provider is authenticated.

class smarter.apps.infrastructure.services.KubectlKubernetesService(provider=None, allow_in_tests=False, **kwargs)[source]

Bases: KubernetesService

The platform’s Kubernetes cluster, through kubectl.

The cluster is ready once kubectl is configured, with the provider’s update_kubeconfig(), and the environment’s namespace, smarter_settings.environment_namespace, exists.

Parameters:

provider (Optional[CloudProvider]) – The cloud provider that writes the kubeconfig. None for a cluster whose kubeconfig is already in place.
allow_in_tests (bool) – Allow kubectl in the unit tests, e.g. when subprocess is mocked.

__init__(provider=None, allow_in_tests=False, **kwargs)[source]

apply_manifest(manifest)[source]

Create or update the resources of a manifest, with kubectl apply.

Resources that provision billable cloud resources, see billable_resources(), are announced with the billable resource signals. Nothing is applied if the cluster is not ready.

Parameters:: manifest (str) – The manifest, which may have several YAML documents.
Raises:: KubernetesServiceError – If the cluster rejects it.
Return type:: None

property configured: bool: Whether kubectl is configured for the cluster.

delete_resource(kind, name, namespace)[source]

Delete a resource by name.

A resource that does not exist counts as deleted.

Return type:: bool

delete_resources(kinds, namespace, selector)[source]

Delete the resources of several kinds that match a label selector.

Idempotent. A selector is required, so that a namespace is never emptied by mistake.

Return type:: bool

get_pod_logs(namespace, selector, container=None, tail=200)[source]

Return the most recent log lines of the pods that match a label selector.

Return type:: Optional[str]

get_resource(kind, name, namespace)[source]

Return a resource, or None if it does not exist or the cluster is unavailable.

Return type:: Optional[dict]

property kubeconfig: dict: The platform’s kubeconfig file.

property kubeconfig_path: str: The path of the platform’s kubeconfig file.

list_resources(kind, namespace, selector=None)[source]

Return the resources of a kind, optionally those that match a label selector.

Return type:: list[dict]

property namespace_verified: bool: Whether the environment’s namespace exists.

property ready: bool: Whether the service is authenticated and connected, i.e. it can be used.

verify_namespace(namespace)[source]

Whether a namespace exists.

Return type:: bool

class smarter.apps.infrastructure.services.KubernetesService(provider_name, *args, **kwargs)[source]

Bases: InfrastructureService

The platform’s Kubernetes cluster.

Implementations provide the primitives: apply_manifest(), get_resource(), list_resources(), delete_resource(), delete_resources() and get_pod_logs(). The ingress operations that LLMClient deployments use are built on them.

abstractmethod apply_manifest(manifest)[source]

Create or update the resources of a manifest.

Raises:: KubernetesServiceError – If the cluster rejects them.
Return type:: None

certificate_wait_seconds: float = 60: Seconds between the checks of a cert-manager certificate, in verify_ingress_resources().

delete_certificate(name, namespace)[source]

Delete a cert-manager Certificate.

Return type:: bool

delete_ingress(name, namespace)[source]

Delete an Ingress.

Return type:: bool

delete_ingress_resources(hostname, namespace)[source]

Delete a host’s Ingress, its cert-manager Certificate, and its TLS Secret.

Return type:: tuple[bool, bool, bool]
Returns:: Whether the Ingress, the Certificate, and the Secret were deleted.

abstractmethod delete_resource(kind, name, namespace)[source]

Delete a resource by name.

A resource that does not exist counts as deleted.

Return type:: bool

abstractmethod delete_resources(kinds, namespace, selector)[source]

Delete the resources of several kinds that match a label selector.

Idempotent. A selector is required, so that a namespace is never emptied by mistake.

Return type:: bool

delete_secret(name, namespace)[source]

Delete a Secret.

Return type:: bool

error_class: alias of KubernetesServiceError

abstractmethod get_pod_logs(namespace, selector, container=None, tail=200)[source]

Return the most recent log lines of the pods that match a label selector.

Return type:: Optional[str]

abstractmethod get_resource(kind, name, namespace)[source]

Return a resource, or None if it does not exist or the cluster is unavailable.

Return type:: Optional[dict]

abstractmethod list_resources(kind, namespace, selector=None)[source]

Return the resources of a kind, optionally those that match a label selector.

Return type:: list[dict]

service_name: str = 'kubernetes': The name of the service in signals and logs, see InfrastructureServiceNames.

verify_certificate(name, namespace)[source]

Whether a cert-manager Certificate exists, and is Ready, i.e. issued.

Parameters:

name (str) – The Certificate’s name.
namespace (str) – The Certificate’s namespace.

Return type:

bool

verify_ingress(name, namespace)[source]

Whether an Ingress exists.

Return type:: bool

verify_ingress_resources(hostname, namespace, max_attempts=30)[source]

Verify that a host’s Ingress, its cert-manager Certificate, and its TLS Secret exist.

The Ingress is named after the host, and the Certificate and Secret <host>-tls.

Parameters:

hostname (str) – The host, e.g. example.3141-5926-5359.api.example.com.
namespace (str) – The namespace.
max_attempts (int) – How many times to check the Certificate, a minute apart. A Celery task passes 1, and checks again later, so that it does not block its worker.

Return type:

tuple[bool, bool, bool]

Returns:

Whether the Ingress, the Certificate, and the Secret are verified.

verify_secret(name, namespace)[source]

Whether a Secret exists.

Return type:: bool

class smarter.apps.infrastructure.services.SMTPEmailService(allow_in_tests=False, smtp_class=<class 'smtplib.SMTP'>, **kwargs)[source]

Bases: EmailService

Email, with SMTP, configured by the smtp_* settings of smarter_settings.

In the unit tests, nothing is sent, unless allow_in_tests is True, so that tests never email real people.

Parameters:

allow_in_tests (bool) – Send email in the unit tests.
smtp_class (Callable[..., SMTP]) – The SMTP client class, e.g. a fake.

__init__(allow_in_tests=False, smtp_class=<class 'smtplib.SMTP'>, **kwargs)[source]

property ready: bool: Whether the service is authenticated and connected, i.e. it can be used.

smarter.apps.infrastructure.services.configure_email(factory)[source]

Set the factory of the email service that get_email() returns.

Parameters:: factory (Optional[Callable[[], EmailService]]) – Returns the service, or None to restore the default, SMTPEmailService.
Return type:: None

smarter.apps.infrastructure.services.configure_kubernetes(factory)[source]

Set the factory of the Kubernetes service that get_kubernetes() returns.

Parameters:: factory (Optional[Callable[[], KubernetesService]]) – Returns the service, or None to restore the default, KubectlKubernetesService with the configured cloud provider.
Return type:: None

smarter.apps.infrastructure.services.get_email()[source]

Return the email service, which is created once.

Return type:: EmailService

smarter.apps.infrastructure.services.get_kubernetes()[source]

Return the Kubernetes service.

It is created once, and again if the cloud provider is reconfigured, so that it keeps its readiness, rather than configuring kubectl for each call.

Return type:: KubernetesService

smarter.apps.infrastructure.services.infrastructure = <smarter.apps.infrastructure.services.InfrastructureServices object>: The platform’s infrastructure services.

smarter.apps.infrastructure.services.refuse_in_unit_tests(what, allow_in_tests=False)[source]

Refuse to reach real infrastructure from the unit tests.

The Smarter containers’ cloud credentials and kubeconfig are real, and can create billable resources. A real service calls this before it reaches its backend. Tests install a fake, e.g. InMemoryProvider with configure_provider(), or pass allow_in_tests=True when they mock the backend themselves, or are tagged INFRASTRUCTURE.

Parameters:

what (str) – What is refused, for the error message.
allow_in_tests (bool) – True to allow it anyway.

Raises:

InfrastructureConfigurationError – In the unit tests, unless allowed.

Return type:

None

The base class of every infrastructure service.

InfrastructureService gives each service, whichever provider implements it:

its identity in signals and logs: service_name and provider_name.
the signals of smarter.apps.infrastructure.signals, sent with send(), and the lifecycle helpers that send them in the right order.
operation(), which translates a provider’s SDK errors into the service’s own exception, so that the platform never catches a provider’s exceptions.
the unit test guard: refuse_in_unit_tests().

class smarter.apps.infrastructure.services.base.InfrastructureService(provider_name, *args, **kwargs)[source]

Bases: ABC, SmarterHelperMixin

The base class of every infrastructure service.

Parameters:: provider_name (str) – The name of the cloud provider that implements the service, e.g. aws. Services that do not depend on a cloud, e.g. SMTP email, use their protocol’s name.

__init__(provider_name, *args, **kwargs)[source]

connection_state(connected, error=None)[source]

Record whether the service is connected to its backend.

Sends infrastructure_connected or infrastructure_connection_failed when the state changes, rather than on every check.

Parameters:

connected (bool) – Whether the service is connected.
error (Optional[str]) – Why it is not connected.

Return type:

bool

Returns:

connected.

created_resource(resource, resource_id=None)[source]

Send the signal that follows the creation of a resource.

Return type:: None

creating_resource(resource_type, resource_name, billable=False, **kwargs)[source]

Send the signal that precedes the creation of a resource.

Only billable resources have a “creating” signal, so that what costs money can be stopped or audited before it exists.

Return type:: dict[str, Any]
Returns:: The resource’s signal arguments, for created_resource().

destroyed_resource(resource)[source]

Send the signal that follows the destruction of a resource.

Return type:: None

destroying_resource(resource_type, resource_name, resource_id=None, billable=False)[source]

Send the signal that precedes the destruction of a resource.

Return type:: dict[str, Any]
Returns:: The resource’s signal arguments, for destroyed_resource().

error_class

The exception that operation() raises for a provider’s errors.

alias of SmarterInfrastructureError

operation(name)[source]

Run one of the service’s operations, translating its errors.

An infrastructure error passes through. Any other error, e.g. a cloud SDK’s, is raised as the service’s error_class. Either way, infrastructure_operation_failed is sent.

with self.operation("create_zone"):
    response = self.client.create_hosted_zone(...)

Parameters:: name (str) – The operation’s name, e.g. create_zone.
Return type:: Iterator[None]

abstract property ready: bool: Whether the service is authenticated and connected, i.e. it can be used.

require_ready()[source]

Raise unless the service is ready.

Raises:: InfrastructureNotReadyError – If the service is not ready.
Return type:: None

send(signal, **kwargs)[source]

Send one of the infrastructure signals, with this service’s service and provider.

Return type:: None

service_name: str = 'service': The name of the service in signals and logs, see InfrastructureServiceNames.

smarter.apps.infrastructure.services.base.refuse_in_unit_tests(what, allow_in_tests=False)[source]

Refuse to reach real infrastructure from the unit tests.

The Smarter containers’ cloud credentials and kubeconfig are real, and can create billable resources. A real service calls this before it reaches its backend. Tests install a fake, e.g. InMemoryProvider with configure_provider(), or pass allow_in_tests=True when they mock the backend themselves, or are tagged INFRASTRUCTURE.

Parameters:

what (str) – What is refused, for the error message.
allow_in_tests (bool) – True to allow it anyway.

Raises:

InfrastructureConfigurationError – In the unit tests, unless allowed.

Return type:

None