[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731:
URL: https://github.com/apache/airflow/pull/7731#discussion_r420367312



##
File path: airflow/providers/amazon/aws/operators/ec2_start_instance.py
##
@@ -0,0 +1,72 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2StartInstanceOperator(BaseOperator):
+"""
+Start AWS EC2 instance using boto3.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ("instance_id", "region_name")
+ui_color = "#eeaa11"
+ui_fgcolor = "#ff"

Review comment:
   maybe black could be better option for ec2 operator, but ec2 state 
sensor has darker background





This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731:
URL: https://github.com/apache/airflow/pull/7731#discussion_r420366644



##
File path: airflow/providers/amazon/aws/operators/ec2_start_instance.py
##
@@ -0,0 +1,72 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2StartInstanceOperator(BaseOperator):
+"""
+Start AWS EC2 instance using boto3.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ("instance_id", "region_name")
+ui_color = "#eeaa11"
+ui_fgcolor = "#ff"

Review comment:
   
![image](https://user-images.githubusercontent.com/18196182/81109578-f9c02280-8f22-11ea-8f9c-48bc204be170.png)
   
   I can change it





This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731:
URL: https://github.com/apache/airflow/pull/7731#discussion_r413786148



##
File path: airflow/providers/amazon/aws/operators/ec2_start_instance.py
##
@@ -0,0 +1,72 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2StartInstanceOperator(BaseOperator):
+"""
+Start AWS EC2 instance using boto3.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ["region_name"]

Review comment:
   all done :heavy_check_mark: waiting travis to build





This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r405791980
 
 

 ##
 File path: airflow/providers/amazon/aws/hooks/ec2.py
 ##
 @@ -0,0 +1,88 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+
+
+class EC2Hook(AwsBaseHook):
+"""
+Interact with AWS EC2 Service.
+
+Additional arguments (such as ``aws_conn_id``) may be specified and
+are passed down to the underlying AwsBaseHook.
+
+.. seealso::
+:class:`~airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook`
+"""
+
+def __init__(self,
+ *args,
+ **kwargs):
+super().__init__(resource_type="ec2", *args, **kwargs)
+
+def get_instance(self, instance_id: str):
+"""
+Get EC2 instance by id and return it.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:return: Instance object
+:rtype: ec2.Instance
+"""
+return self.get_conn().Instance(id=instance_id)
 
 Review comment:
   Oh, I have changed all other get_conn() methods with self.conn, but forgot 
this one :)
   thanks


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r396004322
 
 

 ##
 File path: airflow/providers/amazon/aws/hooks/ec2.py
 ##
 @@ -0,0 +1,98 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from typing import Optional
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+
+
+class EC2Hook(AwsBaseHook):
+"""
+Interact with AWS EC2 Service.
+"""
+
+def __init__(self,
+ region_name: Optional[str] = None,
+ *args,
+ **kwargs):
+self.region_name = region_name
+self.conn = None
+super().__init__(*args, **kwargs)
+
+def get_conn(self):
+"""
+Return self.conn, if it is None initialize ec2 resource object.
+
+:return: ec2 resource
+:rtype: boto3.resource
+"""
+# set self.conn in first call
+if not self.conn:
+self.conn = self.get_resource_type("ec2", self.region_name)
+return self.conn
+
+def get_instance(self, instance_id: str):
+"""
+Get EC2 instance by id and return it.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:return: Instance object
+:rtype: ec2.Instance
+"""
+return self.get_conn().Instance(id=instance_id)
+
+def get_instance_state(self, instance_id: str) -> str:
+"""
+Get EC2 instance state by id and return it.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:return: current state of the instance
+:rtype: str
+"""
+return self.get_instance(instance_id=instance_id).state["Name"]
+
+def wait_for_state(self,
+   instance_id: str,
+   target_state: str,
+   check_interval: float) -> None:
+"""
+Wait EC2 instance until its state is equal to the target_state.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param target_state: target state of instance
+:type target_state: str
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+:return: None
+:rtype: None
+"""
+instance_state = self.get_instance_state(
+instance_id=instance_id
+)
+while instance_state != target_state:
+self.log.info("instance state: %s", instance_state)
+time.sleep(check_interval)
+instance_state = self.get_instance_state(
+instance_id=instance_id
+)
 
 Review comment:
   Not sure, I heard that while True is a bad practice since it's hard to 
follow the loop condition, but the other one has some code duplication. So, I 
am not sure, what do you think about this, do you know any post related to this 
situation. I am facing this situation frequently and still not know what to do 
:(
   [example question frrom 
stackoverflow](https://stackoverflow.com/questions/743164/emulate-a-do-while-loop-in-python)


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r396003763
 
 

 ##
 File path: docs/operators-and-hooks-ref.rst
 ##
 @@ -360,12 +360,19 @@ These integrations allow you to perform various 
operations within the Amazon Web
  -
  -
 
-   * - `Amazon EC2 `__
+   * - `Amazon ECS `__
  -
  -
  - :mod:`airflow.providers.amazon.aws.operators.ecs`
  -
 
+   * - `Amazon EC2 `__
+ -
+ - :mod:`airflow.providers.amazon.aws.hooks.ec2`
+ - :mod:`airflow.providers.amazon.aws.operators.ec2_start_instance`,
+   :mod:`airflow.providers.amazon.aws.operators.ec2_stop_instance`,
+ - :mod:`airflow.providers.amazon.aws.sensors.ec2_instance_state`
+
 
 Review comment:
   Sure, I did not noticed that they are alphabetical


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392979402
 
 

 ##
 File path: airflow/providers/amazon/aws/operators/ec2_start_instance.py
 ##
 @@ -0,0 +1,88 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from typing import Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2StartInstanceOperator(BaseOperator):
+"""
+Start AWS EC2 instance using boto3.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ["region_name"]
+ui_color = "#eeaa11"
+ui_fgcolor = "#ff"
+
+@apply_defaults
+def __init__(self,
+ instance_id: str,
+ aws_conn_id: str = "aws_default",
+ region_name: Optional[str] = None,
+ check_interval: float = 15,
+ *args,
+ **kwargs):
+super().__init__(*args, **kwargs)
+self.instance_id = instance_id
+self.aws_conn_id = aws_conn_id
+self.region_name = region_name
+self.check_interval = check_interval
+
+def execute(self, context):
+ec2_hook = self.get_hook()
+self.log.info("Starting EC2 instance %s", self.instance_id)
+
+instance = ec2_hook.get_instance(instance_id=self.instance_id)
+instance.start()
+instance_state = ec2_hook.get_instance_state(
+instance_id=self.instance_id
+)
+
+while instance_state != "running":
+self.log.info("instance state: %s", instance_state)
+time.sleep(self.check_interval)
+instance_state = ec2_hook.get_instance_state(
+instance_id=self.instance_id
+)
+
+def get_hook(self):
+"""
+Return EC2Hook object.
+
+:return: ec2 hook
+:rtype: EC2Hook
+"""
+return EC2Hook(
+aws_conn_id=self.aws_conn_id,
+region_name=self.region_name
+)
 
 Review comment:
   You're right, a little unnecessary. I saw examples like this and thought "I 
should write like this". I will change it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392977905
 
 

 ##
 File path: airflow/providers/amazon/aws/operators/ec2_start_instance.py
 ##
 @@ -0,0 +1,88 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from typing import Optional
+
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2StartInstanceOperator(BaseOperator):
+"""
+Start AWS EC2 instance using boto3.
+
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ["region_name"]
+ui_color = "#eeaa11"
+ui_fgcolor = "#ff"
+
+@apply_defaults
+def __init__(self,
+ instance_id: str,
+ aws_conn_id: str = "aws_default",
+ region_name: Optional[str] = None,
+ check_interval: float = 15,
+ *args,
+ **kwargs):
+super().__init__(*args, **kwargs)
+self.instance_id = instance_id
+self.aws_conn_id = aws_conn_id
+self.region_name = region_name
+self.check_interval = check_interval
+
+def execute(self, context):
+ec2_hook = self.get_hook()
+self.log.info("Starting EC2 instance %s", self.instance_id)
+
+instance = ec2_hook.get_instance(instance_id=self.instance_id)
+instance.start()
+instance_state = ec2_hook.get_instance_state(
+instance_id=self.instance_id
+)
+
+while instance_state != "running":
+self.log.info("instance state: %s", instance_state)
+time.sleep(self.check_interval)
+instance_state = ec2_hook.get_instance_state(
+instance_id=self.instance_id
+)
 
 Review comment:
   Sure, I will.


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392914708
 
 

 ##
 File path: airflow/providers/amazon/aws/operators/ec2.py
 ##
 @@ -0,0 +1,102 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2Operator(BaseOperator):
+"""
+Manage AWS EC2 instance using boto3.
+Change instance state by applying given operation.
+
+:param operation: action to be taken on AWS EC2 instance
+valid values: "start", "stop"
+:type operation: str
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ["operation", "region_name"]
+ui_color = "#eeaa11"
+ui_fgcolor = "#ff"
+valid_operations = ["start", "stop"]
+operation_target_state_map = {
+"start": "running",
+"stop": "stopped",
+}
+
+@apply_defaults
+def __init__(self,
+ operation: str,
+ instance_id: str,
+ aws_conn_id: str = "aws_default",
+ region_name: Optional[str] = None,
+ check_interval: float = 15,
+ *args,
+ **kwargs):
+if operation not in self.valid_operations:
+raise AirflowException(f"Invalid operation: {operation}")
 
 Review comment:
   This part removed, after spliting EC2StartInstanceOperator and 
EC2StopInstanceOperator


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392914054
 
 

 ##
 File path: airflow/providers/amazon/aws/sensors/ec2.py
 ##
 @@ -0,0 +1,78 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.sensors.base_sensor_operator import BaseSensorOperator
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2Sensor(BaseSensorOperator):
+"""
+Check the state of the AWS EC2 instance until
+state of the instance become equal to the target state.
+
+:param target_state: target state of instances
+:type target_state: str
+:param instance_id: id of the AWS EC2 Instances
+:type instance_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+"""
+
+template_fields = ["target_state", "region_name"]
+ui_color = "#cc8811"
+ui_fgcolor = "#ff"
+valid_states = ["running", "stopped"]
+
+@apply_defaults
+def __init__(self,
+ target_state: str,
+ instance_id: str,
+ aws_conn_id: str = "aws_default",
+ region_name: Optional[str] = None,
+ *args,
+ **kwargs):
+if target_state not in self.valid_states:
+raise AirflowException(f"Invalid target_state: {target_state}")
+super().__init__(*args, **kwargs)
+self.target_state = target_state
+self.instance_id = instance_id
+self.aws_conn_id = aws_conn_id
+self.region_name = region_name
+self.hook = self.get_hook()
 
 Review comment:
   I didn't know that, fixed, thank you. 
   P.S.: There are some bad practices in other operators.


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392914383
 
 

 ##
 File path: airflow/providers/amazon/aws/operators/ec2.py
 ##
 @@ -0,0 +1,102 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2Operator(BaseOperator):
+"""
+Manage AWS EC2 instance using boto3.
+Change instance state by applying given operation.
+
+:param operation: action to be taken on AWS EC2 instance
+valid values: "start", "stop"
+:type operation: str
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ["operation", "region_name"]
+ui_color = "#eeaa11"
+ui_fgcolor = "#ff"
+valid_operations = ["start", "stop"]
+operation_target_state_map = {
+"start": "running",
+"stop": "stopped",
+}
+
+@apply_defaults
+def __init__(self,
+ operation: str,
+ instance_id: str,
+ aws_conn_id: str = "aws_default",
+ region_name: Optional[str] = None,
+ check_interval: float = 15,
+ *args,
+ **kwargs):
+if operation not in self.valid_operations:
+raise AirflowException(f"Invalid operation: {operation}")
+super().__init__(*args, **kwargs)
+self.operation = operation
+self.instance_id = instance_id
+self.aws_conn_id = aws_conn_id
+self.region_name = region_name
+self.check_interval = check_interval
+self.target_state = self.operation_target_state_map[self.operation]
+self.hook = self.get_hook()
 
 Review comment:
   fixed


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392913117
 
 

 ##
 File path: airflow/providers/amazon/aws/sensors/ec2.py
 ##
 @@ -0,0 +1,78 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.sensors.base_sensor_operator import BaseSensorOperator
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2Sensor(BaseSensorOperator):
+"""
+Check the state of the AWS EC2 instance until
+state of the instance become equal to the target state.
+
+:param target_state: target state of instances
+:type target_state: str
+:param instance_id: id of the AWS EC2 Instances
+:type instance_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+"""
+
+template_fields = ["target_state", "region_name"]
+ui_color = "#cc8811"
+ui_fgcolor = "#ff"
+valid_states = ["running", "stopped"]
 
 Review comment:
   added, thx


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392912965
 
 

 ##
 File path: airflow/providers/amazon/aws/operators/ec2.py
 ##
 @@ -0,0 +1,102 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2Operator(BaseOperator):
 
 Review comment:
   I have split it into 2 operators: EC2StartInstanceOperator and 
EC2StopInstanceOperator


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392705410
 
 

 ##
 File path: airflow/providers/amazon/aws/hooks/ec2.py
 ##
 @@ -0,0 +1,48 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Optional
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+
+
+class EC2Hook(AwsBaseHook):
+"""
+Interact with AWS EC2 Service.
+"""
+
+def __init__(self,
+ region_name: Optional[str] = None,
+ *args,
+ **kwargs):
+self.region_name = region_name
+self.conn = None
+super().__init__(*args, **kwargs)
+
+def get_conn(self):
+"""
+Return self.conn, if it is None initialize ec2 resource object.
+
+:return: ec2 resource
+:rtype: boto3.resource
+"""
+# set self.conn in first call
+if not self.conn:
+self.conn = self.get_resource_type("ec2", self.region_name)
+return self.conn
 
 Review comment:
   You are right, I think I should add a method named get_instance(instance_id) 
or get_instance_state(instance_id) to thi EC2Hook class. Then I can call this 
from operator/sensor. What do you think?


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392705410
 
 

 ##
 File path: airflow/providers/amazon/aws/hooks/ec2.py
 ##
 @@ -0,0 +1,48 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import Optional
+
+from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
+
+
+class EC2Hook(AwsBaseHook):
+"""
+Interact with AWS EC2 Service.
+"""
+
+def __init__(self,
+ region_name: Optional[str] = None,
+ *args,
+ **kwargs):
+self.region_name = region_name
+self.conn = None
+super().__init__(*args, **kwargs)
+
+def get_conn(self):
+"""
+Return self.conn, if it is None initialize ec2 resource object.
+
+:return: ec2 resource
+:rtype: boto3.resource
+"""
+# set self.conn in first call
+if not self.conn:
+self.conn = self.get_resource_type("ec2", self.region_name)
+return self.conn
 
 Review comment:
   You are right, I think I should add a method named get_instance(instance_id) 
or get_instance_state(instance_id) to this EC2Hook class. Then I can call this 
from operator/sensor. What do you think?


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] [airflow] mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create EC2 Hook, Operator and Sensor

[GitBox]

mustafagok commented on a change in pull request #7731: [AIRFLOW-7068] Create 
EC2 Hook, Operator and Sensor
URL: https://github.com/apache/airflow/pull/7731#discussion_r392705107
 
 

 ##
 File path: airflow/providers/amazon/aws/operators/ec2.py
 ##
 @@ -0,0 +1,102 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from typing import Optional
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.amazon.aws.hooks.ec2 import EC2Hook
+from airflow.utils.decorators import apply_defaults
+
+
+class EC2Operator(BaseOperator):
+"""
+Manage AWS EC2 instance using boto3.
+Change instance state by applying given operation.
+
+:param operation: action to be taken on AWS EC2 instance
+valid values: "start", "stop"
+:type operation: str
+:param instance_id: id of the AWS EC2 instance
+:type instance_id: str
+:param aws_conn_id: aws connection to use
+:type aws_conn_id: str
+:param region_name: (optional) aws region name associated with the client
+:type region_name: Optional[str]
+:param check_interval: time in seconds that the job should wait in
+between each instance state checks until operation is completed
+:type check_interval: float
+"""
+
+template_fields = ["operation", "region_name"]
+ui_color = "#eeaa11"
+ui_fgcolor = "#ff"
+valid_operations = ["start", "stop"]
+operation_target_state_map = {
+"start": "running",
+"stop": "stopped",
+}
+
+@apply_defaults
+def __init__(self,
+ operation: str,
+ instance_id: str,
+ aws_conn_id: str = "aws_default",
+ region_name: Optional[str] = None,
+ check_interval: float = 15,
+ *args,
+ **kwargs):
+if operation not in self.valid_operations:
+raise AirflowException(f"Invalid operation: {operation}")
+super().__init__(*args, **kwargs)
+self.operation = operation
+self.instance_id = instance_id
+self.aws_conn_id = aws_conn_id
+self.region_name = region_name
+self.check_interval = check_interval
+self.target_state = self.operation_target_state_map[self.operation]
+self.hook = self.get_hook()
+
+def execute(self, context):
+self.log.info("Executing: %s %s", self.operation, self.instance_id)
+
+instance = self.hook.get_conn().Instance(id=self.instance_id)
+
+if self.operation == "start":
 
 Review comment:
   I can write EC2StartInstanceOperator and EC2StopInstanceOperator, but these 
operators will be identical except 2 lines. What do you suggest?


This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services