Andres Crucetta
04/11/2023, 4:52 PM@pytest.fixture
def data_config():
"""
Op configuration parameters.
"""
return DataTransferConfig(
raw_table="RAW_CERPASS_MAC",
azure_stage="STAGE_CERPASS_MAC",
file_format="PLAIN_TEXT"
)
@pytest.fixture
def db_resource():
"""
Snowflake database resource.
"""
return DatabaseResource(
user=EnvVar("SNOWFLAKE_USER"),
account=EnvVar("SNOWFLAKE_ACCOUNT"),
password=EnvVar("SNOWFLAKE_PASSWORD"),
warehouse=EnvVar("SNOWFLAKE_WAREHOUSE_NAME"),
)
def test_load_sftp_to_snowflake(data_config: DataTransferConfig,
db_resource: DatabaseResource):
"""
Test the copy into op.
"""
# Load environment variables
context = build_op_context(
config=data_config)
result = load_sftp_to_snowflake(context)
assert result is True
class DataTransferConfig(Config):
raw_table: str
azure_stage: str
file_format: str
@op()
def load_sftp_to_snowflake(config: DataTransferConfig, db_conn: DatabaseResource) -> bool:
"""
Copy the file into Snowflake
"""
# Get the snowflake connection
<http://logging.info|logging.info>("Copying file into Snowflake")
query = f"""
COPY INTO {config.raw_table}
FROM @{config.azure_stage}
CREDENTIALS = (
AZURE_SAS_TOKEN='{EnvVar('AZURE_SAS_TOKEN')}'
)
FILE_FORMAT = (FORMAT_NAME = {config.file_format})
"""
try:
db_conn.execute(query)
except Exception as e:
logging.error(f"Error while copying file into Snowflake: {e}")
return False
<http://logging.info|logging.info>("File copied into Snowflake")
return True
What's the best way to test an Operation with Pythonic resources and configs? I can't find testing documentation for this new way of defining parametersschrockn
04/11/2023, 6:09 PMload_sftp_to_snowflake
directly in your test
e.g.
load_sftp_to_snowflake(db_conn=db_resource, config=data_config)
Andres Crucetta
04/11/2023, 6:19 PMschrockn
04/11/2023, 7:45 PMAndres Crucetta
04/11/2023, 8:26 PM