Implement Async `RedshiftOperator` about astronomer-providers HOT 11 CLOSED

Based on the existing sync version of AWS Redshift, these are all the following operator classes which needs Async versions.

• RedshiftSQLOperator
• AwsRedshiftClusterSensor
• Redshift cluster management operators (RedshiftResumeClusterOperator, RedshiftPauseClusterOperator)
• S3ToRedshiftOperator

RedshiftSQLOperator and AwsRedshiftClusterSensor uses python redshift_connector library to connect to redshift db, and execute the queries these needs Async version implementation.

• RedshiftSQLHookAsync
• RedshiftTrigger
• RedshiftSQLOperatorAsync
• RedshiftHookAsync
• RedshiftResumeClusterOperator, RedshiftPauseClusterOperator
• AwsRedshiftClusterSensorAsync
• S3ToRedshiftOperatorAsync (Separated it to #67)

S3ToRedshiftOperator uses S3Hook which uses python boto3 library to connect to s3 and copy file from s3 to redshift by copy command these needs S3HookAsync async version which will be a part of #14 implementation once its done we need to implement the S3ToRedshiftOperator async version

from astronomer-providers.

For RedshiftResumeClusterOperator, RedshiftPauseClusterOperator and AwsRedshiftClusterSensorAsync I can able to connect to redshift cluster via aiobotocore python async library I can able to pause and resume cluster and get the cluster status

from astronomer-providers.

On the RedshiftSQLOperatorAsync, RedshiftSQLHookAsync I couldn't find async version of redshift_connectorand any other library to connect with redshift and execute the query, SO I am planning to use the syn_to_async function to make the async connection and need to dig around the redshift connector library whether it helps us to track the query status.

In RedshiftSQLHook there is function get_sqlalchemy_engine() which uses sqlalchemy-redshift library to connect to the redshift cluster, currently its override the super class function I am not sure this function is used in the RedshiftSQLHook

Need to come up with solution for RedshiftSQLHookAsync connection

from astronomer-providers.

@sunank200 add this code as common function in awshooksasync class

import asyncio
from aiobotocore.session import get_session

def get_client():
session = get_session()
return session.create_client('redshift', region_name='us-east-1',
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
aws_access_key_id=AWS_ACCESS_KEY_ID)

how to use the get_client
async def go():
cluster_identifier = "astro-redshift-cluster-1"
async with get_client() as client:
response = await client.describe_clusters(ClusterIdentifier=cluster_identifier)
print(response)

from astronomer-providers.

Added code for RedshiftResumeClusterOperatorAsync and RedshiftPauseClusterOperatorAsync into async-redshift branch.

Added

• RedshiftResumeClusterOperatorAsync, RedshiftPauseClusterOperatorAsync
• RedshiftHookAsync
• RedshiftClusterTrigger

commit id: 9ac0c33

from astronomer-providers.

Added AwsRedshiftClusterSensorAsync

commit id: dfb5756

from astronomer-providers.

Added example Dags for redshift cluster management operators and sensor operators

commit id: a3fdfe2

from astronomer-providers.

completed and gave PR for the following operators

• RedshiftResumeClusterOperator
• RedshiftPauseClusterOperator
• RedshiftClusterSensorAsync

from astronomer-providers.

Next up: RedshiftSQLOperator

from astronomer-providers.

Raised PR on RedshiftSQLOperator

from astronomer-providers.

Async versions for the below operators have been implemented and merged into main branch. Closing this story.

1. RedshiftPauseClusterOperator
2. RedshiftResumeClusterOperator
3. RedshiftSQLOperator
4. RedshiftClusterSensor

from astronomer-providers.