https://dagster.io/ logo
#ask-community
Title
# ask-community
a

Andres Crucetta

04/11/2023, 4:52 PM
Copy code
@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 parameters
s

schrockn

04/11/2023, 6:09 PM
Hey Andres you should be able to invoke
load_sftp_to_snowflake
directly in your test e.g.
load_sftp_to_snowflake(db_conn=db_resource, config=data_config)
a

Andres Crucetta

04/11/2023, 6:19 PM
Thanks @schrockn! Do I still need to bind the resources to the jobs in 1.2? The blog post with Dagster 1.3 says we should for 1.2 but the actual pythonic resource article in the docs doesn't mention it I'm trying to give dagster a good chance, we have a demo today with the team and I want to use the pythonic config to set it up right from the beginning. So far it's been relatively straightforward. I've had a few issues switching to this new config/resource method and how it handles environmentals/definitions
s

schrockn

04/11/2023, 7:45 PM
Yes you still need to bind resources to jobs in 1.2. That change ships with 1.3
👍 1
a

Andres Crucetta

04/11/2023, 8:26 PM
Got it! And just to confirm this will be only with resources, doesn't apply to configuration