How to use the kedro.io.DataCatalog.from_config function in kedro

To help you get started, we’ve selected a few kedro examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github quantumblacklabs / kedro / tests / contrib / io / catalog_with_default / test_data_catalog_with_default.py View on Github external
def test_from_sane_config_default(sane_config, dummy_dataframe, tmpdir):
    catalog = DataCatalog.from_config(
        sane_config["catalog"], sane_config["credentials"]
    )
    catalog_with_default = DataCatalogWithDefault.from_data_catalog(
        catalog, default_csv
    )
    path = str(tmpdir.mkdir("sub").join("missing.csv"))
    catalog_with_default.save(path, dummy_dataframe)
    reloaded_df = catalog_with_default.load(path)
    assert dummy_dataframe.equals(reloaded_df)
github quantumblacklabs / kedro / tests / io / test_data_catalog.py View on Github external
def test_from_sane_config_load_versions_warn(self, sane_config):
        sane_config["catalog"]["boats"]["versioned"] = True
        version = generate_timestamp()
        load_version = {"non-boart": version}
        pattern = r"\`load_versions\` keys \[non-boart\] are not found in the catalog\."
        with pytest.warns(UserWarning, match=pattern):
            DataCatalog.from_config(**sane_config, load_versions=load_version)
github quantumblacklabs / kedro / tests / io / test_data_catalog.py View on Github external
def test_config_invalid_arguments(self, sane_config):
        """Check the error if the data set config contains invalid arguments"""
        sane_config["catalog"]["boats"]["save_and_load_args"] = False
        pattern = (
            r"DataSet 'boats' must only contain arguments valid for "
            r"the constructor of `.*CSVLocalDataSet`"
        )
        with pytest.raises(DataSetError, match=pattern):
            DataCatalog.from_config(**sane_config)
github quantumblacklabs / kedro / tests / io / test_data_catalog.py View on Github external
def test_config_missing_class(self, sane_config):
        """Check the error if the type points to nonexistent class"""
        sane_config["catalog"]["boats"]["type"] = "kedro.io.CSVLocalDataSetInvalid"

        pattern = (
            "An exception occurred when parsing config for DataSet `boats`:\n"
            "Class `kedro.io.CSVLocalDataSetInvalid` not found"
        )
        with pytest.raises(DataSetError, match=re.escape(pattern)):
            DataCatalog.from_config(**sane_config)
github quantumblacklabs / kedro / tests / io / test_data_catalog.py View on Github external
def test_idempotent_catalog(self, sane_config):
        """Test that data catalog instantiations are idempotent"""
        _ = DataCatalog.from_config(**sane_config)  # NOQA
        catalog = DataCatalog.from_config(**sane_config)
        assert catalog
github quantumblacklabs / kedro / tests / io / test_data_catalog.py View on Github external
def test_idempotent_catalog(self, sane_config):
        """Test that data catalog instantiations are idempotent"""
        _ = DataCatalog.from_config(**sane_config)  # NOQA
        catalog = DataCatalog.from_config(**sane_config)
        assert catalog
github quantumblacklabs / kedro / tests / io / test_data_catalog.py View on Github external
def test_link_credentials(self, sane_config, mocker):
        """Test credentials being linked to the relevant data set"""
        mock_client = mocker.patch("kedro.io.csv_s3.S3FileSystem")

        DataCatalog.from_config(**sane_config)

        expected_client_kwargs = {
            "aws_access_key_id": sane_config["credentials"]["s3_credentials"][
                "aws_access_key_id"
            ],
            "aws_secret_access_key": sane_config["credentials"]["s3_credentials"][
                "aws_secret_access_key"
            ],
        }
        mock_client.assert_called_once_with(client_kwargs=expected_client_kwargs)
github tipresias / tipresias / data_science / src / machine_learning / run.py View on Github external
DataCatalog defined in `catalog.yml`.

    """
    conf_logging = config.get("logging*", "logging*/**")
    logging.config.dictConfig(conf_logging)
    conf_catalog = config.get("catalog*", "catalog*/**")

    try:
        conf_creds = config.get("credentials*", "credentials*/**")
    except MissingConfigException:
        warn("Your Kedro project is missing a credentials file!")
        conf_creds = None

    conf_params = config.get("parameters*", "parameters*/**")
    logging.config.dictConfig(conf_logging)
    catalog = DataCatalog.from_config(conf_catalog, conf_creds)
    catalog.add_feed_dict({"parameters": conf_params})
    return catalog
github quantumblacklabs / kedro / kedro / context / context.py View on Github external
def _create_catalog(  # pylint: disable=no-self-use,too-many-arguments
        self,
        conf_catalog: Dict[str, Any],
        conf_creds: Dict[str, Any],
        save_version: str = None,
        journal: Journal = None,
        load_versions: Dict[str, str] = None,
    ) -> DataCatalog:
        """A factory method for the DataCatalog instantiation.

        Returns:
            DataCatalog defined in `catalog.yml`.

        """
        return DataCatalog.from_config(
            conf_catalog,
            conf_creds,
            save_version=save_version,
            journal=journal,
            load_versions=load_versions,
        )