You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/06/08 20:50:39 UTC
[GitHub] [airflow] vincbeck commented on a diff in pull request #24099: Add AWS operators to create and delete RDS Database
vincbeck commented on code in PR #24099:
URL: https://github.com/apache/airflow/pull/24099#discussion_r892854210
##########
tests/system/providers/amazon/aws/rds/example_rds_db_instance.py:
##########
@@ -0,0 +1,75 @@
+#
+# 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 datetime import datetime
+from os import getenv
+
+from airflow import DAG
+from airflow.models.baseoperator import chain
+from airflow.providers.amazon.aws.operators.rds import (
+ RdsCreateDbInstanceOperator,
+ RdsDeleteDbInstanceOperator,
+)
+from tests.system.providers.amazon.aws.utils import set_env_id
+
+ENV_ID = set_env_id()
+DAG_ID = "example_rds_instance"
+
+RDS_DB_IDENTIFIER = getenv("RDS_DB_IDENTIFIER", "database-identifier")
+RDS_USERNAME = getenv("RDS_USERNAME", "database_username")
+# NEVER store your production password in plaintext in a DAG like this.
+# Use Airflow Secrets or a secret manager for this in production.
+RDS_PASSWORD = getenv("RDS_PASSWORD", "database_password")
Review Comment:
```suggestion
RDS_DB_IDENTIFIER = f'{ENV_ID}-database'
RDS_USERNAME = 'database_username'
# NEVER store your production password in plaintext in a DAG like this.
# Use Airflow Secrets or a secret manager for this in production.
RDS_PASSWORD = 'database_password'
```
##########
tests/system/providers/amazon/aws/rds/example_rds_db_instance.py:
##########
@@ -0,0 +1,75 @@
+#
Review Comment:
One of the convention we (AWS) are following is we name the file with the same name we use as dag ID. Here `example_rds_instance` makes sense so we should rename this file `example_rds_instance`
##########
airflow/providers/amazon/aws/operators/rds.py:
##########
@@ -551,6 +551,94 @@ def execute(self, context: 'Context') -> str:
return json.dumps(delete_subscription, default=str)
+class RdsCreateDbInstanceOperator(RdsBaseOperator):
+ """
+ Creates an RDS DB instance
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:RdsCreateDbInstanceOperator`
+
+ :param db_instance_identifier: The DB instance identifier, must start with a letter and
+ contain from 1 to 63 letters, numbers, or hyphens
+ :param db_instance_class: The compute and memory capacity of the DB instance, for example db.m5.large
+ :param engine: The name of the database engine to be used for this instance
+ :param rds_kwargs: Named arguments to pass to boto3 RDS client function ``create_db_instance``
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ *,
+ db_instance_identifier: str,
+ db_instance_class: str,
+ engine: str,
+ rds_kwargs: Optional[Dict] = None,
+ aws_conn_id: str = "aws_default",
+ **kwargs,
+ ):
+ super().__init__(aws_conn_id=aws_conn_id, **kwargs)
+
+ self.db_instance_identifier = db_instance_identifier
+ self.db_instance_class = db_instance_class
+ self.engine = engine
+ self.rds_kwargs = rds_kwargs or {}
+
+ def execute(self, context: 'Context') -> str:
+ self.log.info(f"Creating new DB instance {self.db_instance_identifier}")
+
+ create_db_instance = self.hook.conn.create_db_instance(
+ DBInstanceIdentifier=self.db_instance_identifier,
+ DBInstanceClass=self.db_instance_class,
+ Engine=self.engine,
+ **self.rds_kwargs,
+ )
+ self.hook.conn.get_waiter("db_instance_available").wait(
+ DBInstanceIdentifier=self.db_instance_identifier
+ )
+
+ return json.dumps(create_db_instance, default=str)
+
+
+class RdsDeleteDbInstanceOperator(RdsBaseOperator):
+ """
+ Deletes an RDS DB Instance
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:RdsDeleteDbInstanceOperator`
+
+ :param db_instance_identifier: The DB instance identifier for the DB instance to be deleted
+ :param rds_kwargs: Named arguments to pass to boto3 RDS client function ``delete_db_instance``
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ *,
+ db_instance_identifier: str,
+ rds_kwargs: Optional[Dict] = None,
+ aws_conn_id: str = "aws_default",
+ **kwargs,
+ ):
+ super().__init__(aws_conn_id=aws_conn_id, **kwargs)
+ self.db_instance_identifier = db_instance_identifier
+ self.rds_kwargs = rds_kwargs or {}
+
+ def execute(self, context: 'Context') -> str:
+ self.log.info(f"Deleting DB instance {self.db_instance_identifier}")
+
+ delete_db_instance = self.hook.conn.delete_db_instance(
+ DBInstanceIdentifier=self.db_instance_identifier,
+ **self.rds_kwargs,
+ )
+ self.hook.conn.get_waiter("db_instance_deleted").wait(
+ DBInstanceIdentifier=self.db_instance_identifier
+ )
Review Comment:
Same here
##########
airflow/providers/amazon/aws/operators/rds.py:
##########
@@ -551,6 +551,94 @@ def execute(self, context: 'Context') -> str:
return json.dumps(delete_subscription, default=str)
+class RdsCreateDbInstanceOperator(RdsBaseOperator):
+ """
+ Creates an RDS DB instance
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:RdsCreateDbInstanceOperator`
+
+ :param db_instance_identifier: The DB instance identifier, must start with a letter and
+ contain from 1 to 63 letters, numbers, or hyphens
+ :param db_instance_class: The compute and memory capacity of the DB instance, for example db.m5.large
+ :param engine: The name of the database engine to be used for this instance
+ :param rds_kwargs: Named arguments to pass to boto3 RDS client function ``create_db_instance``
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ """
+
+ def __init__(
+ self,
+ *,
+ db_instance_identifier: str,
+ db_instance_class: str,
+ engine: str,
+ rds_kwargs: Optional[Dict] = None,
+ aws_conn_id: str = "aws_default",
+ **kwargs,
+ ):
+ super().__init__(aws_conn_id=aws_conn_id, **kwargs)
+
+ self.db_instance_identifier = db_instance_identifier
+ self.db_instance_class = db_instance_class
+ self.engine = engine
+ self.rds_kwargs = rds_kwargs or {}
+
+ def execute(self, context: 'Context') -> str:
+ self.log.info(f"Creating new DB instance {self.db_instance_identifier}")
+
+ create_db_instance = self.hook.conn.create_db_instance(
+ DBInstanceIdentifier=self.db_instance_identifier,
+ DBInstanceClass=self.db_instance_class,
+ Engine=self.engine,
+ **self.rds_kwargs,
+ )
+ self.hook.conn.get_waiter("db_instance_available").wait(
+ DBInstanceIdentifier=self.db_instance_identifier
+ )
Review Comment:
Should be have this logic under a flag? In other words, do we want to wait the db to be available by default or do we want to leave this option to the user?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org