Hi, Im trying to set a default value for config’s ...
# ask-community
c
Hi, Im trying to set a default value for config’s variable but got the following error:
Copy code
dagster.core.errors.DagsterInvalidConfigError: Invalid default_value for Field.
    Error 1: Value at path root:download_3d_conformers_from_s3:config:file_names must not be None. Expected "[String]"
I define my op as following:
Copy code
@configured(download_files, config_schema={"output_path": str,
                                           "file_names": Field(Array(str), default_value=[], is_required=False)})
def download_3d_conformers_from_s3(config: dagster.utils.frozendict) -> Dict[str, Union[str, List[str]]]:
    return {"bucket": "my-bucket",
            "sub_folder": "pubchem/SDF/",
            "file_names": config["file_names"],
            "file_extension": "sdf",
            "output_path": config["output_path"]}

@op(required_resource_keys={'s3'},
    config_schema={"bucket": str,
                   "sub_folder": str,
                   "file_extension": str,
                   "output_path": str,
                   "file_names": Field(Array(str), default_value=[], is_required=False),
                   }
    )
def download_files(context: SolidExecutionContext) -> None:
    """
    Download files from a bucket in s3 and save them locally.
    :param context: contains config for the op with the following variables:
    bucket: bucket to download from.
    sub_folder: the name of the sub folder where the files exist.
    file_names: list of file names to download (without file extension).
    file_extension: file extension of the files.
    output_path: the path where to saves the files locally.
    """

    bucket_name =context.op_config["bucket"]
    file_names = context.op_config["file_names"]
    sub_folder = context.op_config["sub_folder"]
    file_ext = context.op_config["file_extension"]

    if len(file_names) == 0:
        keys = context.resources.s3.get_matching_keys(bucket_name=bucket_name, prefix=sub_folder, suffix=file_ext)
    else:
        keys = [f"{sub_folder}{name}.{file_ext}" for name in file_names]

    for key in keys:
        filename = os.path.join(context.op_config["output_path"], key.split(S3_PATH_SEPARATOR)[-1])
        context.resources.s3.download_file(filename, key, bucket_name)
what am I doing wrong ?
c
Hi Chen. What version of dagster are you running? I think a fix went out for this in dagster
0.14.2
c
0.14.3