Model and Manager extensions

botocraft provides a number of mixins and decorators that are used to extend the functionality of the model and manager classes which are generated from the botocore definitions and botocraft definitions. These mixins and decorators are used to:

provide additional methods to those classes that are not provided by the AWS API
implement those methods in a more useful way
translate between what the AWS API provides and what we want to provide to the user.

Note

The mixins and decorators are not intended to be used directly by the user. They are intended to be used by the botocraft code generator to generate the service and manager classes.

Models

class botocraft.mixins.tags.TagsDict(*args, **kwargs)[source]

Bases: dict

A dict subclass that allows python dictionary style access to the Tags attribute for a model instance. Normally Tags is a list of dictionaries, but this class allows you to treat tags like a dictionary where the keys are the tag names and the values are the tag values.

This subclass does the following additional things:

overrides the __setitem__ method to set the the appropriate key and value in the Tags attribute of the model instance.
overrides the __delitem__ method to delete the appropriate key and value in the Tags attribute of the model instance.
overrides the pop method to delete the appropriate key and value in the Tags attribute of the model instance.

check() → None[source]

Check if we have been configured properly.

This means:

the instance attribute must be set.
self.instance must have an attribute named Tags.
the tag_class attribute must be set.
the tag_Key attribute must be set.
the tag_Value attribute must be set.

Returns:: True if we have been configured, False otherwise.

pop(key: str, default: Any | None = None) → Any[source]

Override the pop method to delete the tags in ourselves and in the Tags attribute of the model instance.

Parameters:: key – the key to delete.
Keyword Arguments:: default – the default value to return if the key is not found.
Returns:: Returns the value of the key if it exists, otherwise the default

instance: 'Boto3Model' | None: The model instance that we are setting tags for.

tag_Key: str | None: The name of the key attribute in the tag class.

tag_Value: str | None: The name of the value attribute in the tag class.

tag_class: Type | None: The boto3 tag class. This has to be configurable because once again boto3 has different tag classes for different services.

pydantic model botocraft.mixins.ecr.ImageInfo[source]

Bases: BaseModel

A class to hold information about a botocraft.services.ecr.Image that is not available from the boto3 library. We extract this information by pulling the image from the repository and inspecting it with the docker Python library.

Important

You must have the docker daemon running to use the methods that return this object.

Show JSON schema

{
   "title": "ImageInfo",
   "description": "A class to hold information about a :py:class:`botocraft.services.ecr.Image`\nthat is not available from the boto3 library.  We extract this information\nby pulling the image from the repository and inspecting it with the docker\nPython library.\n\nImportant:\n    You must have the docker daemon running to use the methods that return\n    this object.",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "platform": {
         "title": "Platform",
         "type": "string"
      },
      "architecture": {
         "title": "Architecture",
         "type": "string"
      },
      "size": {
         "title": "Size",
         "type": "integer"
      },
      "ports": {
         "additionalProperties": {
            "additionalProperties": true,
            "type": "object"
         },
         "default": {},
         "title": "Ports",
         "type": "object"
      },
      "docker_version": {
         "title": "Docker Version",
         "type": "string"
      },
      "user": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "User"
      },
      "created": {
         "format": "date-time",
         "title": "Created",
         "type": "string"
      }
   },
   "required": [
      "name",
      "platform",
      "architecture",
      "size",
      "docker_version",
      "created"
   ]
}

Fields:

architecture (str)
created (datetime.datetime)
docker_version (str)
name (str)
platform (str)
ports (dict[str, dict[str, typing.Any]])
size (int)
user (str | None)

field architecture: str [Required]: The architecture of the image

field created: datetime [Required]: When the image was created, as a UTC datetime object

field docker_version: str [Required]: Docker Version used to build the image

field name: str [Required]

field platform: str [Required]: The OS platform of the image

field ports: dict[str, dict[str, Any]] = {}: This is a dictionary of port mappings. The key is the port and the value is i’m not sure what

field size: int [Required]: Size of the image in bytes

field user: str | None = None: The user that the image runs as

Decorators

Application autoscaling

botocraft.mixins.application_autoscaling.scaling_policy_only(func: Callable[[...], PutScalingPolicyResponse]) → Callable[[...], ScalingPolicy][source]: Wraps application-autoscaling put_scaling_policy methods to return the actual botocraft.services.application_autoscaling.ScalingPolicy object instead of the response object.

botocraft.mixins.application_autoscaling.scalable_target_only(func: Callable[[...], str]) → Callable[[...], ScalableTarget][source]: Wraps application-autoscaling register_scalable_target methods to return the actual botocraft.services.application_autoscaling.ScalableTarget object instead of the ARN.

botocraft.mixins.application_autoscaling.scheduled_action_only(func: Callable[[...], None]) → Callable[[...], ScheduledAction][source]: Wraps application-autoscaling put_scheduled_action methods to return the actual botocraft.services.application_autoscaling.ScalableTarget object instead of the ARN.

EC2

botocraft.mixins.ec2.ec2_instance_only(func: Callable[[...], Reservation | None]) → Callable[[...], Instance | None][source]: Wraps a boto3 method that returns a list of Reservation objects to return a single Instance object instead.

ECR

botocraft.mixins.ecr.repo_list_images_ecr_images_only(func: Callable[[...], PrimaryBoto3ModelQuerySet]) → Callable[[...], PrimaryBoto3ModelQuerySet][source]: Convert a list of ECR image identifiers returned by botocraft.services.ecr.RepositoryManager.list_images into a list of botocraft.services.ecr.Image objects.

botocraft.mixins.ecr.image_list_images_ecr_images_only(func: Callable[[...], PrimaryBoto3ModelQuerySet]) → Callable[[...], list[ECRImage]][source]: Convert a list of ECR image identifiers returned by botocraft.services.ecr.Image.list into a list of botocraft.services.ecr.Image objects.

ECS

botocraft.mixins.ecs.ecs_services_only(func: Callable[[...], list[str]]) → Callable[[...], PrimaryBoto3ModelQuerySet][source]: Wraps botocraft.services.ecs.ServiceManager.list to return a PrimaryBoto3ModelQuerySet of botocraft.services.ecs.Service objects instead of only a list of ARNs.

botocraft.mixins.ecs.ecs_clusters_only(func: Callable[[...], list[str]]) → Callable[[...], PrimaryBoto3ModelQuerySet][source]: Wraps botocraft.services.ecs.ClusterManager.list to return a list of botocraft.services.abstract.PrimaryBoto3ModelQuerySet of botocraft.services.ecs.Cluster objects instead of only a list of ARNs.

botocraft.mixins.ecs.ecs_task_definitions_only(func: Callable[[...], list[str]]) → Callable[[...], PrimaryBoto3ModelQuerySet][source]: Decorator to convert a list of ECS task definition identifiers to a list of botocraft.services.ecs.TaskDefinition objects.

botocraft.mixins.ecs.ecs_container_instances_only(func: Callable[[...], list[str]]) → Callable[[...], PrimaryBoto3ModelQuerySet][source]: Decorator to convert a list of ECS container instance arns to a botocraft.services.abstract.PrimaryBoto3ModelQuerySet of botocraft.services.ecs.ContainerInstance objects.

botocraft.mixins.ecs.ecs_task_populate_taskDefinition(func: Callable[[...], Task | None]) → Callable[[...], Task | None][source]

Wraps botocraft.services.ecs.TaskManager.get to populate the botocraft.services.ecs.Task.taskDefinition attribute.

We set the taskDefinition attribute to the task family and revision in the format <family>:<revision>. taskDefinition is an extra field that we add to the botocraft.services.ecs.Task object that is not in the original botocore shape, but is useful for our purposes.

botocraft.mixins.ecs.ecs_task_populate_taskDefinitions(func: Callable[[...], PrimaryBoto3ModelQuerySet]) → Callable[[...], PrimaryBoto3ModelQuerySet][source]

Wraps botocraft.services.ecs.TaskManager.get_many to populate the botocraft.services.ecs.Task.taskDefinition attribute on each task.

We set the taskDefinition attribute to the task family and revision in the format <family>:<revision>. taskDefinition is an extra field that we add to the botocraft.services.ecs.Task object that is not in the original botocore shape, but is useful for our purposes.

botocraft.mixins.ecs.ecs_tasks_only(func: Callable[[...], list[str]]) → Callable[[...], PrimaryBoto3ModelQuerySet][source]: Wrap botocraft.services.ecs.TaskManager.list to return a list of botocraft.services.ecs.Task objects instead of only a list of ARNs.

Manager Mixins

EC2

class botocraft.mixins.ec2.AMIManagerMixin[source]

A mixin is used on botocraft.services.ec2.AMIManager to add miscellaneous methods to the class that are not normally part of the object.

in_use(owners: list[str] | None = None, tags: dict[str, str] | None = None, created_since: datetime | None = None, amis: list['AMI'] | None = None) → PrimaryBoto3ModelQuerySet[source]

Return a list of AMIs that are currently in use by a running or stopped instance, or by an autoscaling group’s launch configuration or launch template.

If amis is specified, all other filters are ignored, because we expect the user to know which AMIs they are submitting.

Important

We’re not checking for AMIs in the following places:

CloudFormation templates
OpsWorks stacks
Elastic Beanstalk environments
Launch Template Versions not in use by an autoscaling group

Keyword Arguments:

owners – Scopes the results to AMIs with the specified owners. You can specify a combination of Amazon Web Services account IDs, self, amazon, and aws-marketplace. If you omit this parameter, the results include all images for which you have launch permissions, regardless of ownership. If not specified, the default is self.
tags – Filters the AMIs to those who match the these tags.
created_since – Filters the AMIs to those created since this date.
amis – Filters the AMIs to those in this list. All other filters are ignored if this is specified.

MAX_AMI_FILTER_SIZE: Final[int] = 200: The maximum number of filters that can be added to to AMI.objects.list.

class botocraft.mixins.ec2.EC2TagsManagerMixin[source]

A mixin is used on on botocraft.services.ec2.InstanceManager to convert the odd EC2 tag list to a TagSpecification object.

convert_tags(tags: list['Tag'] | None, resource_type: Literal['capacity-reservation', 'client-vpn-endpoint', 'customer-gateway', 'carrier-gateway', 'coip-pool', 'dedicated-host', 'dhcp-options', 'egress-only-internet-gateway', 'elastic-ip', 'elastic-gpu', 'export-image-task', 'export-instance-task', 'fleet', 'fpga-image', 'host-reservation', 'image', 'import-image-task', 'import-snapshot-task', 'instance', 'instance-event-window', 'internet-gateway', 'ipam', 'ipam-pool', 'ipam-scope', 'ipv4pool-ec2', 'ipv6pool-ec2', 'key-pair', 'launch-template', 'local-gateway', 'local-gateway-route-table', 'local-gateway-virtual-interface', 'local-gateway-virtual-interface-group', 'local-gateway-route-table-vpc-association', 'local-gateway-route-table-virtual-interface-group-association', 'natgateway', 'network-acl', 'network-interface', 'network-insights-analysis', 'network-insights-path', 'network-insights-access-scope', 'network-insights-access-scope-analysis', 'placement-group', 'prefix-list', 'replace-root-volume-task', 'reserved-instances', 'route-table', 'security-group', 'security-group-rule', 'snapshot', 'spot-fleet-request', 'spot-instances-request', 'subnet', 'subnet-cidr-reservation', 'traffic-mirror-filter', 'traffic-mirror-session', 'traffic-mirror-target', 'transit-gateway', 'transit-gateway-attachment', 'transit-gateway-connect-peer', 'transit-gateway-multicast-domain', 'transit-gateway-policy-table', 'transit-gateway-route-table', 'transit-gateway-route-table-announcement', 'volume', 'vpc', 'vpc-endpoint', 'vpc-endpoint-connection', 'vpc-endpoint-service', 'vpc-endpoint-service-permission', 'vpc-peering-connection', 'vpn-connection', 'vpn-gateway', 'vpc-flow-log', 'capacity-reservation-fleet', 'traffic-mirror-filter-rule', 'vpc-endpoint-connection-device-type', 'verified-access-instance', 'verified-access-group', 'verified-access-endpoint', 'verified-access-policy', 'verified-access-trust-provider', 'vpn-connection-device-type', 'vpc-block-public-access-exclusion', 'ipam-resource-discovery', 'ipam-resource-discovery-association', 'instance-connect-endpoint']) → TagSpecification | None[source]

Given a TagList, convert it to a TagSpecification with ResourceType of resource_type.

Parameters:

tags – the list of Tag objects to convert.
resource_type – the EC2 resource type.

Returns:

A TagSpecification object, or None if tags is None.

Model Mixins

Any

class botocraft.mixins.tags.TagsDictMixin(*args, **kwargs)[source]

A mixin that adds a tags property to a model instance. This is intended to be used with Boto3Model classes that have a Tags attribute.

The attribute that has the AWS tags list must be named Tags. If your model stores them in another attribute (for example, ECS objects use tags instead), you will need to override the attribute name by setting the botocraft.sync.models.ModelAttributeDefinition.rename attribute to be Tags for that attribute.

Important

One important assumption this mixin makes is the caller modifies tags via either self.tags or self.Tags, but not both at the same time. If you do use both at the same time, you will get unexpected results.

tag_class: Type | None = None: The boto3 tag class. This has to be configurable because once again boto3 has different tag classes for different services.

property tags: TagsDict

Get the tags for the model instance.

Returns:: The tags for the model instance.

Application AutoScaling

class botocraft.mixins.application_autoscaling.ScalableTargetModelMixin[source]

A mixin class that extends ScalableTarget.

property resource: Service

Return the resource object for this scalable target.

Currently only the ecs service namespace is supported, which returns an Service object.

AutoScaling

class botocraft.mixins.autoscaling.AutoScalingGroupModelMixin[source]

A mixin for AutoScalingGroup that adds some convenience methods.

Sometimes we like full Instance objects instead of the AutoScalingInstanceReference objects that get listed on AutoScalingGroup.Instances.

EC2 is weird and doesn’t have an easy way to list instances in an autoscaling group except to use one of the Filter parameters on describe_instances. Instances that are part of an autoscaling group have a tag called aws:autoscaling:groupName whose value is the name of the autoscaling group. We can use this to filter instances that belong to a particular autoscaling group.

This method is too specialized at the moment to be included as one of the transformers for model related objects, so we’ll just add it as a mixin.

scale(desired_count: int, wait: bool = False, max_attempts: int = 40) → None[source]

Scale the autoscaling group to the desired count.

Parameters:

desired_count – The number of tasks to run.

Keyword Arguments:

wait – If True, wait for the service to reach the desired count.
max_attempts – The maximum number of attempts to make before giving up

wait_until_stable(max_attempts: int = 40, delay: int = 15) → None[source]

Since there is no waiter for this, we’ll use this method to wait until the autoscaling group is stable.

Raises:

botocore.exceptions.WaiterError – If the autoscaling group is not stable after max_attempts.

Keyword Arguments:

max_attempts – The maximum number of attempts to make before giving up.
delay – The number of seconds to wait between attempts.

property ec2_instances: PrimaryBoto3ModelQuerySet: Return the running Instance objects that belong to this group, if any.

property is_stable: bool

Return True if the autoscaling group is stable, False otherwise.

An autoscaling group is considered stable if all instances are running and healthy and the DesiredCapacity is equal to the number of instances.

EC2

class botocraft.mixins.ec2.InstanceModelMixin[source]

Used on botocraft.services.ec2.Instance to add miscellaneous methods to the class that are not normally part of the object.

close_tunnel(host: str, local_port: int | None = None) → None[source]

Close one or more tunnels to host. This will terminate the SSM session(s) and SSH process(es) that were opened by start_tunnel.

If local_port is not specified, all tunnels to the host will be closed. If local_port is specified, only the tunnel to that port will be closed.

Parameters:: host – Either an IP address of the remote host to connect to, or a hostname of that host.
Keyword Arguments:: local_port – The local port to close. If not specified, all tunnels to the host will be closed.
Raises:: ValueError – If no tunnels are found for the given host or port.

open_tunnel(host: str, remote_port: int, local_port: int | None = None, profile: str | None = None) → int[source]

Open a tunnel to the instance using SSM and SSH. This is useful for connecting to a database or other service on the instance that is in a private subnet. This will open a tunnel from the local_port on the local machine through the instance to the remote_port on host.

If local_port is not specified, a random port will be chosen starting from between 8800 and 65535.

Parameters:

host – either an IP address of the remote host to connect to, or a hostname of that host
remote_port – The remote port to connect to.

Keyword Arguments:

local_port – The local port to connect to.
profile – The AWS profile to use. If not specified, the default profile will be used.

Raises:

ValueError – If the local port is already in use.
RuntimeError – If the instance is not connected to SSM or if there is an error when starting the session or SSH tunnel.

Returns:

The local port that was used for the tunnel.

tunnel(host: str, remote_port: int, local_port: int | None = None, profile: str | None = None)[source]

A context manager for opening and closing a tunnel.

This will call open_tunnel when entering the context and close_tunnel when exiting the context.

Parameters:

host – The remote host to connect to (IP or hostname).
remote_port – The remote port to connect to.
local_port – The local port to use. If None, an unused port will be chosen.
profile – The AWS profile to use. If None, the default profile will be used.

Raises:

ValueError – If the local port is already in use.
RuntimeError – If the instance is not connected to SSM or if there is an error when starting the session or SSH tunnel.

Yields:

The local port used for the tunnel.

class botocraft.mixins.ec2.AMIModelMixin[source]

property in_use: bool: Return True if the AMI is in use by a running or stopped instance.

property vulnerabilities: PrimaryBoto3ModelQuerySet

Return a list of vulnerabilities for the instance. This is a convenience method to get the vulnerabilities for the instance.

Returns:: A list of Finding objects.

class botocraft.mixins.ec2.SecurityGroupModelMixin[source]

A mixin is used on botocraft.services.ec2.SecurityGroup to enhance the .save() method to allow for managing ingress and egress rules at the same time as saving the security group.

Normally this is done with several boto3 calls, but this mixin allows for a single call to .save() to create a security group and manage the rules.

save(**kwargs)[source]: Save the model. For security groups, ingress rules are managed via separate boto3 calls than the security group itself. This override of the save method will allow the user to create a security group and add ingress rules in one step.

ECR

class botocraft.mixins.ecr.RepositoryMixin[source]

get_image(imageId: ImageIdentifier) → ECRImage | None[source]

Get an image object for a given repository and image identifier.

Parameters:: imageId – The image ID or tag to describe. The format of the imageId reference is imageTag=tag or imageDigest=digest

property images: PrimaryBoto3ModelQuerySet: Get a list of images for a given repository.

class botocraft.mixins.ecr.ECRImageMixin[source]

Add a bunch of support for inspecting ECR images and getting information from them that AWS does not provide. This is done by using the docker Python library to pull the image and inspect it.

Note

I don’t love doing this because it is not pure AWS, which was my intention for botocraft, but I need these features for business purposes and they are not available in the boto3 library.

clean() → None[source]

Remove the image from our local docker storage, if it exists.

Raises:: RuntimeError – If the docker daemon is not running.

clean_other_versions() → None[source]

Remove the all images for this repository except for the one with our version.

Raises:: RuntimeError – If the docker daemon is not running.

services(status: Literal['ACTIVE', 'INACTIVE', 'ALL'] | None = 'ACTIVE', tags: dict[str, str] | None = None, verbose: bool = False) → PrimaryBoto3ModelQuerySet[source]

Return a list of ECS Services that use this image.

Warning

This will be quite slow if you have a lot of families and revisions, because the only way to deal with this is to get all the task definition families, and then look at each revision to see if one of its containers uses this image. Then look through all our services to see if there is a service that uses that task definition.

Parameters:

status – The status of the task definition to filter by. Valid values are ACTIVE, INACTIVE, or ALL. The default is ACTIVE.
tags – A dictionary of tags to filter task definitions and services by. The default is an empty dictionary.
verbose – If True, print out status messages as we work.

Returns:

A list of ECS Services that use this image.

task_definitions(status: Literal['ACTIVE', 'INACTIVE', 'ALL'] | None = 'ACTIVE', tags: dict[str, str] | None = None, verbose: bool = False) → PrimaryBoto3ModelQuerySet[source]

Return a list of ECS task definitions that use this image.

Warning

This will be quite slow if you have a lot of families and revisions, because the only way to deal with this is to get all the task definition families, and then look at each revision to see if one of its containers uses this image. There is no way to filter the task definitions by image.

Parameters:

status – The status of the task definition to filter by. Valid values are ACTIVE, INACTIVE, or ALL. The default is ACTIVE.
tags – A dictionary of tags to filter by. The default is an empty dictionary.
verbose – If True, print out the task definition family and revision that uses this image. The default is False.

Returns:

A list of ECS task definitions that use this image.

property docker_client: ECRDockerClient

Return a docker client, logged into our ECR registry.

Raises:: RuntimeError – If the docker daemon is not running.
Returns:: A botocraft.mixins.ecr.ECRDockerClient object, which has a docker client, username, password, and registry.

property docker_image: Image

Return the docker.models.images.Image object for this image.

Raises:: RuntimeError – If the docker daemon is not running.

property dockerd_is_running: bool

Check if the docker daemon is running.

We need dockerd to be running to perform these operations:

docker_client
pull
is_pulled
info
docker_image
history
clean
clean_other_versions

property history: list[dict[str, Any]]

Return the build history for this image. You can use this to reconstruct most of the Dockerfile that was used to build the image. You won’t have the FROM line, but you can get most of the rest of it.

Raises:: RuntimeError – If the docker daemon is not running.

property image_name: str: Return just the image name, excluding the registry.

property info: ImageInfo

Return information about the image. We’re doing this by pulling the image from the repository and inspecting it.

Note

I’d love to get the base image for this image, but there is no direct way to do it. You would to look up the layers for the image, get the sha256 hash of the first layer (which is the base image), then look in in various repositories to find the image that has the same layer, then get that image’s name. That seems stupid hard to do, especially if the base image is in the ECR registry of another AWS account.

Raises:: RuntimeError – If the docker daemon is not running.
Returns:: A botocraft.services.ecr.ImageInfo object.

property is_pulled: bool

Check if the image is pulled.

Returns:: True if the image is pulled, False otherwise.

property name: str: Get the name of the image.

property version: str: Get the version of the image.

property vulnerabilities: PrimaryBoto3ModelQuerySet

Return a list of vulnerabilities for this image. This is done by using the AWS Inspector2 service to scan the image and return the vulnerabilities.

Note

The AWS Inspector service is not instantaneous, but runs occasionally. This doesn’t matter much for us, because we are using the ECR immutable images, so we can just get the vulnerabilities for the image we are using.

Warning

If this image was just pushed, then the scan may not have run yet. In that case, you will need to wait for the scan to run before you can get the vulnerabilities.

Returns:: A list of vulnerabilities for this image.

ECS

class botocraft.mixins.ecs.ECSServiceModelMixin[source]

A mixin for botocraft.services.ecs.Service that adds some additional methods that we can’t auto generate.

scale(desired_count: int, wait: bool = False) → None[source]

Scale the service to the desired count. If wait is True, this will wait for the service to reach the desired count using the services_stable boto3 waiter.

Parameters:: desired_count – The number of tasks to run.
Keyword Arguments:: wait – If True, wait for the service to reach the desired count.

wait_until_stable(max_attempts: int = 40, delay: int = 15) → None[source]

Wait until the service is stable.

Raises:

botocore.exceptions.WaiterError – if the service is not stable after max_attempts, or some other error occurred.

Keyword Arguments:

max_attempts – The maximum number of attempts to make before giving up.
delay – The number of seconds to wait between attempts.

property container_instances: PrimaryBoto3ModelQuerySet: Return the botocraft.services.ecs.ContainerInstance objects which are running our tasks for the service.

property is_stable: bool: Return whether the service is stable or not.

property load_balancers: PrimaryBoto3ModelQuerySet: Return the LoadBalancer objects that are associated with the service.

property required_cpu: int: The required CPU for the service in CPU shares. One full CPU is equivalent to 1024 CPU shares.

property required_memory: int: Return the required memory for the service in MiB.

class botocraft.mixins.ecs.ECSContainerInstanceModelMixin[source]

property free_cpu: int: Return the free CPU shares on the container instance. One full CPU is equivalent to 1024 CPU shares.

property free_ram: int: Return the free RAM in MiB on the container instance.