EC2_Handler.py

import boto3
import json

class EC2Handler:
  """
  Class used to handle operations with EC2 through boto3.
  """
  def __init__(self, region_name: str):
    self.client = boto3.resource('ec2', region_name)
  
  def list_ec2_on_region(self, filters: list = []):     
    '''
    This function returns a list of all the ec2 instances given a region.
    The arguments contains the status and tags with whitch we have to do the filtering on the instances.
    It returns a list of all the instances filtered.
    '''
    self.lista_ec2 = []
    if len(filters):
      formatted_filters = []
      for filter in filters:
        formatted_filters.append(
          {
            'Name': filter[0],
            'Values': [
              filter[1]
            ]
          }
        )

    for instance in self.client.instances.filter(Filters=formatted_filters):
      self.lista_ec2.append(instance)

  def start_ec2(self, instance: 'boto3.resources.factory.ec2.Instance'):
    '''
    Start the Ec2 instance passed as argument.
    Return error if not able to start it.
    '''
    try:
      instance.start()
      print(f"Instance with id {instance.id} is going to start.")
      assert (instance.state['Name'] == 'pending' or instance.state['Name'] == 'running')
    except AssertionError:
      raise BaseException(f"Unable to start instance with id {instance.id}.")
    return

      
  def stop_ec2(self, instance: 'boto3.resources.factory.ec2.Instance'):
    '''
    Stop the Ec2 instance passed as argument.
    Return error if not able to stop it.
    '''
    try:
      instance.stop()
      print(f"Instance with id {instance.id} is going to stop.")
      assert (instance.state['Name'] == 'stopping' or instance.state['Name'] == 'stopped')
    except AssertionError:
      raise BaseException(f'Unable to stop instance with id {instance.id}.')
    return

handler.py

"""
This Lambda is triggered by a CloudWhatch event. It start or stop certain ec2 
instances depending on the schedule:

1. Get a list of all ec2 instances given a region, filtered by state and tag.
2. Check if the list contains any instance
3. In case it contains instances start/stop them.

Lambda is executed via CloudWatch schedule at fixed schedules (i.e. every half hour or at set times).

If executed at 8.30am, it will query all the EC2 instances in a region that have tag Key=tag:Start & Value=8:30 
that are not already running, and attempt to start them. It will then query all the instances with
tag Key=tag:Stop & Value=8:30 and do the same.
"""

import time
import os
from EC2_Handler import EC2Handler

REGION = os.environ.get('REGION')

def lambda_handler(event, context):
  ec2 = EC2Handler(REGION)
  now = time.localtime()
  time_tag = f'{now.tm_hour}:{now.tm_min}'

  try:
    ec2.list_ec2_on_region([('tag:Start', time_tag), ('instance-state-name', 'stopped')])
  except Exception as e:
    print("Unable to list instances.")
    return

  try:
    assert len(ec2.lista_ec2)
    print(f'Found {len(ec2.lista_ec2)} instance(s) to start.')
  except:
    print('No instances marked for starting found OR all instances marked are already started')
    return
  for instance in ec2.lista_ec2:
    try:
      ec2.start_ec2(instance)
    except Exception as e:
      print(e)
      pass

  try:
    ec2.list_ec2_on_region([('tag:Stop', time_tag), ('instance-state-name', 'running')])
  except Exception as e:
    print("Unable to list instances.")
    return

  try:
    assert len(ec2.lista_ec2)
    print(f'Found {len(ec2.lista_ec2)} instance(s) to stop.')
  except:
    print('No instances marked for stopping found OR all instances marked are already stopped')
    return
  for instance in ec2.lista_ec2:
    try:
      ec2.stop_ec2(instance)
    except Exception as e:
      print(e)
      pass