Я пытаюсь использовать @mock_redshift
из moto для имитации соединения с AWS Redshift, я использую boto3 для создания тестового кластера, но когда я делаю запрос, я получаю следующую ошибку:
sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not translate host name "recon-test.cg034hpkmmjt.us-east-1.redshift.amazonaws.com" to address: Name or service not known
Код следующий:
@mock_redshift
@mock_s3
@pytest.mark.vcr
def test_extract_stp_both_dates():
client = boto3.client('redshift', region_name='us-east-1')
response = client.create_cluster(
DBName='recon',
ClusterIdentifier='recon-test',
ClusterType='single-node',
NodeType='ds2.xlarge',
MasterUsername='cuenca',
MasterUserPassword='password',
)
host = response['Cluster']['Endpoint']['Address']
port = response['Cluster']['Endpoint']['Port']
rs_client = RedshiftClient(
'recon',
'cuenca',
'password',
host,
port,
)
rs_client.s.execute("CREATE TABLE table_test (attr VARCHAR);") # The error is here
conn = boto3.resource('s3', region_name='us-east-1')
conn.create_bucket(Bucket=os.environ['S3_BUCKET'])
random.seed(1)
responses.add_passthru('https://')
extract('01/11/2018', '30/11/2018', rs_client=rs_client)
Это RedshiftClient
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker
class RedshiftClient:
def __init__(
self, database: str, user: str, password: str, host: str, port: str
):
self.connection_string = (
f'redshift+psycopg2://{user}:{password}@{host}:{port}/{database}'
)
self.engine = sa.create_engine(self.connection_string)
self.sessionmaker = sessionmaker(bind=self.engine)
self.s = self.sessionmaker()
def exec_query(self, query: str) -> list:
return self.s.execute(query).fetchall()