20220928.5

Build # 20220928.5

Build Type

push

circleci

Committed by GitHub

Commit Message

<a href="https://github.com/great-expectations/great_expectations/commit/<a class=hub.com/great-expectations/great_expectations/commit/<a class="double-link" href="https://git"><a class=hub.com/great-expectations/great_expectations/commit/<a class="double-link" href="https://git"><a class=hub.com/great-expectations/great_expectations/commit/<a class="double-link" href="https://git"><a class=hub.com/great-expectations/great_expectations/commit/0f02bce8afe5a3555a7d8f0037c578245ec568c0">0f02bce8a">&lt;a href=&quot;https://github.com/great-expectations/great_expectations/commit/</a><a class="double-link" href="https://github.com/great-expectations/great_expectations/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/great-expectations/great_expectations/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/great-expectations/great_expectations/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/great-expectations/great_expectations/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/great-expectations/great_expectations/commit/0f02bce8afe5a3555a7d8f0037c578245ec568c0">0f02bce8a</a><a href="https://github.com/great-expectations/great_expectations/commit/0f02bce8afe5a3555a7d8f0037c578245ec568c0">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/great-expectations/great_expectations/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/great-expectations/great_expectations/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/great-expectations/great_expectations/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/great-expectations/great_expectations/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/great-expectations/great_expectations/commit/0f02bce8afe5a3555a7d8f0037c578245ec568c0&quot;&gt;0f02bce8a&lt;/a&gt;&lt;a href=&quot;https://github.com/great-expectations/great_expectations/commit/0f02bce8afe5a3555a7d8f0037c578245ec568c0&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/great-expectations/great_ex... (continued)

Run Details

19620 of 49613 relevant lines covered (39.55%)

0.4 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

29.91

/great_expectations/datasource/batch_kwargs_generator/subdir_reader_batch_kwargs_generator.py

import logging
import os
import warnings
from typing import Dict, Iterable

from great_expectations.datasource.batch_kwargs_generator.batch_kwargs_generator import (
    BatchKwargsGenerator,
)
from great_expectations.datasource.types import PathBatchKwargs
from great_expectations.exceptions import BatchKwargsError

logger = logging.getLogger(__name__)

KNOWN_EXTENSIONS = [
    ".csv",
    ".tsv",
    ".parquet",
    ".parq",
    ".pqt",
    ".xls",
    ".xlsx",
    ".json",
    ".csv.gz",
    ".tsv.gz",
    ".feather",
    ".pkl",
]


class SubdirReaderBatchKwargsGenerator(BatchKwargsGenerator):
    """The SubdirReaderBatchKwargsGenerator inspects a filesystem and produces path-based batch_kwargs.

    SubdirReaderBatchKwargsGenerator recognizes data assets using two criteria:
      - for files directly in 'base_directory' with recognized extensions (.csv, .tsv, .parquet, .xls, .xlsx, .json
        .csv.gz, tsv.gz, .feather, .pkl), it uses the name of the file without the extension
      - for other files or directories in 'base_directory', is uses the file or directory name

    SubdirReaderBatchKwargsGenerator sees all files inside a directory of base_directory as batches of one datasource.

    SubdirReaderBatchKwargsGenerator can also include configured reader_options which will be added to batch_kwargs generated
    by this generator.
    """

    _default_reader_options: Dict = {}
    recognized_batch_parameters = {"data_asset_name", "partition_id"}

    def __init__(
        self,
        name="default",
        datasource=None,
        base_directory="/data",
        reader_options=None,
        known_extensions=None,
        reader_method=None,
    ) -> None:
        super().__init__(name, datasource=datasource)
        if reader_options is None:
            reader_options = self._default_reader_options

        if known_extensions is None:
            known_extensions = KNOWN_EXTENSIONS

        self._known_extensions = known_extensions
        self._reader_options = reader_options
        self._reader_method = reader_method
        self._base_directory = base_directory

    @property
    def reader_options(self):
        return self._reader_options

    @property
    def known_extensions(self):
        return self._known_extensions

    @property
    def reader_method(self):
        return self._reader_method

    @property
    def base_directory(self):
        # If base directory is a relative path, interpret it as relative to the data context's
        # context root directory (parent directory of great_expectation dir)
        if os.path.isabs(self._base_directory) or self._datasource.data_context is None:
            return self._base_directory
        else:
            return os.path.join(
                self._datasource.data_context.root_directory, self._base_directory
            )

    def get_available_data_asset_names(self):
        if not os.path.isdir(self.base_directory):
            return {"names": [], "is_complete_list": True}
        known_assets = self._get_valid_file_options(base_directory=self.base_directory)
        return {"names": known_assets, "is_complete_list": True}

    # TODO: deprecate generator_asset argument
    def get_available_partition_ids(self, generator_asset=None, data_asset_name=None):
        assert (generator_asset and not data_asset_name) or (
            not generator_asset and data_asset_name
        ), "Please provide either generator_asset or data_asset_name."
        if generator_asset:
            # deprecated-v0.11.0
            warnings.warn(
                "The 'generator_asset' argument is deprecated as of v0.11.0 and will be removed in v0.16. "
                "Please use 'data_asset_name' instead.",
                DeprecationWarning,
            )
            data_asset_name = generator_asset

        # If the generator asset names a single known *file*, return ONLY that
        for extension in self.known_extensions:
            if os.path.isfile(
                os.path.join(self.base_directory, data_asset_name + extension)
            ):
                return [data_asset_name]
        if os.path.isfile(os.path.join(self.base_directory, data_asset_name)):
            return [data_asset_name]

        # Otherwise, subdir files are partition ids
        return [
            path
            for (path, type) in self._get_valid_file_options(
                base_directory=os.path.join(self.base_directory, data_asset_name)
            )
        ]

    def _build_batch_kwargs(self, batch_parameters):
        """

        Args:
            batch_parameters:

        Returns:
            batch_kwargs

        """
        try:
            data_asset_name = batch_parameters.pop("data_asset_name")
        except KeyError:
            raise BatchKwargsError(
                "Unable to build BatchKwargs: no name provided in batch_parameters.",
                batch_kwargs=batch_parameters,
            )

        if "partition_id" in batch_parameters:
            partition_id = batch_parameters.pop("partition_id")
            # Find the path
            path = None
            for extension in self.known_extensions:
                if os.path.isfile(
                    os.path.join(
                        self.base_directory, data_asset_name, partition_id + extension
                    )
                ):
                    path = os.path.join(
                        self.base_directory, data_asset_name, partition_id + extension
                    )

            if path is None:
                logger.warning(
                    "Unable to find path with the provided partition; searching for asset-name partitions."
                )
                # Fall through to this case in the event that there is not a subdir available, or if partition_id was
                # not provided
                if os.path.isfile(os.path.join(self.base_directory, data_asset_name)):
                    path = os.path.join(self.base_directory, data_asset_name)

                for extension in self.known_extensions:
                    if os.path.isfile(
                        os.path.join(self.base_directory, data_asset_name + extension)
                    ):
                        path = os.path.join(
                            self.base_directory, data_asset_name + extension
                        )

            if path is None:
                raise BatchKwargsError(
                    f"Unable to build batch kwargs from for asset '{data_asset_name}'",
                    batch_parameters,
                )
            return self._build_batch_kwargs_from_path(path, **batch_parameters)

        else:
            return self.yield_batch_kwargs(
                data_asset_name=data_asset_name, **batch_parameters
            )

    def _get_valid_file_options(self, base_directory=None):
        valid_options = []
        if base_directory is None:
            base_directory = self.base_directory
        file_options = os.listdir(base_directory)
        for file_option in file_options:
            for extension in self.known_extensions:
                if (
                    file_option.endswith(extension)
                    and not file_option.startswith(".")
                    and (file_option[: -len(extension)], "file") not in valid_options
                ):
                    valid_options.append((file_option[: -len(extension)], "file"))
                elif os.path.isdir(os.path.join(self.base_directory, file_option)):
                    # Make sure there's at least one valid file inside the subdir
                    subdir_options = self._get_valid_file_options(
                        base_directory=os.path.join(base_directory, file_option)
                    )
                    if (
                        len(subdir_options) > 0
                        and (file_option, "directory") not in valid_options
                    ):
                        valid_options.append((file_option, "directory"))
        return valid_options

    def _get_iterator(self, data_asset_name, reader_options=None, limit=None):
        logger.debug(
            f"Beginning SubdirReaderBatchKwargsGenerator _get_iterator for data_asset_name: {data_asset_name}"
        )
        # If the data asset is a file, then return the path.
        # Otherwise, use files in a subdir as batches
        if os.path.isdir(os.path.join(self.base_directory, data_asset_name)):
            subdir_options = os.listdir(
                os.path.join(self.base_directory, data_asset_name)
            )
            batches = []
            for file_option in subdir_options:
                for extension in self.known_extensions:
                    if file_option.endswith(extension) and not file_option.startswith(
                        "."
                    ):
                        batches.append(
                            os.path.join(
                                self.base_directory, data_asset_name, file_option
                            )
                        )

            return self._build_batch_kwargs_path_iter(
                batches, reader_options=reader_options, limit=limit
            )
        else:
            for extension in self.known_extensions:
                path = os.path.join(self.base_directory, data_asset_name + extension)
                if os.path.isfile(path):
                    return iter(
                        [
                            self._build_batch_kwargs_from_path(
                                path, reader_options=reader_options, limit=limit
                            )
                        ]
                    )
            # If we haven't returned yet, raise
            raise BatchKwargsError(
                "No valid files found when searching {:s} using configured known_extensions: "
                "{:s} ".format(
                    os.path.join(self.base_directory, data_asset_name),
                    ", ".join(map(str, self.known_extensions)),
                ),
                batch_kwargs=PathBatchKwargs(
                    path=os.path.join(self.base_directory, data_asset_name)
                ),
            )

    def _build_batch_kwargs_path_iter(
        self, path_list, reader_options=None, limit=None
    ) -> Iterable[PathBatchKwargs]:
        for path in path_list:
            yield self._build_batch_kwargs_from_path(
                path, reader_options=reader_options, limit=limit
            )

    def _build_batch_kwargs_from_path(
        self, path, reader_method=None, reader_options=None, limit=None
    ):
        batch_kwargs = self._datasource.process_batch_parameters(
            reader_method=reader_method or self.reader_method,
            reader_options=reader_options or self.reader_options,
            limit=limit,
        )
        batch_kwargs["path"] = path
        batch_kwargs["datasource"] = self._datasource.name
        return PathBatchKwargs(batch_kwargs)

1	import logging	3✔
2	import os	3✔
3	import warnings	3✔
4	from typing import Dict, Iterable	3✔
5
6	from great_expectations.datasource.batch_kwargs_generator.batch_kwargs_generator import (	3✔
7	BatchKwargsGenerator,
8	)
9	from great_expectations.datasource.types import PathBatchKwargs	3✔
10	from great_expectations.exceptions import BatchKwargsError	3✔
11
12	logger = logging.getLogger(__name__)	3✔
13
14	KNOWN_EXTENSIONS = [	3✔
15	".csv",
16	".tsv",
17	".parquet",
18	".parq",
19	".pqt",
20	".xls",
21	".xlsx",
22	".json",
23	".csv.gz",
24	".tsv.gz",
25	".feather",
26	".pkl",
27	]
28
29
30	class SubdirReaderBatchKwargsGenerator(BatchKwargsGenerator):	3✔
31	"""The SubdirReaderBatchKwargsGenerator inspects a filesystem and produces path-based batch_kwargs.
32
33	SubdirReaderBatchKwargsGenerator recognizes data assets using two criteria:
34	- for files directly in 'base_directory' with recognized extensions (.csv, .tsv, .parquet, .xls, .xlsx, .json
35	.csv.gz, tsv.gz, .feather, .pkl), it uses the name of the file without the extension
36	- for other files or directories in 'base_directory', is uses the file or directory name
37
38	SubdirReaderBatchKwargsGenerator sees all files inside a directory of base_directory as batches of one datasource.
39
40	SubdirReaderBatchKwargsGenerator can also include configured reader_options which will be added to batch_kwargs generated
41	by this generator.
42	"""
43
44	_default_reader_options: Dict = {}	3✔
45	recognized_batch_parameters = {"data_asset_name", "partition_id"}	3✔
46
47	def __init__(	3✔
48	self,
49	name="default",
50	datasource=None,
51	base_directory="/data",
52	reader_options=None,
53	known_extensions=None,
54	reader_method=None,
55	) -> None:
56	super().__init__(name, datasource=datasource)	3✔
57	if reader_options is None:	3✔
58	reader_options = self._default_reader_options	×
59
60	if known_extensions is None:	3✔
61	known_extensions = KNOWN_EXTENSIONS	3✔
62
63	self._known_extensions = known_extensions	3✔
64	self._reader_options = reader_options	3✔
65	self._reader_method = reader_method	3✔
66	self._base_directory = base_directory	3✔
67
68	@property	3✔
69	def reader_options(self):	1✔
70	return self._reader_options	×
71
72	@property	3✔
73	def known_extensions(self):	1✔
74	return self._known_extensions	×
75
76	@property	3✔
77	def reader_method(self):	1✔
78	return self._reader_method	×
79
80	@property	3✔
81	def base_directory(self):	1✔
82	# If base directory is a relative path, interpret it as relative to the data context's
83	# context root directory (parent directory of great_expectation dir)
84	if os.path.isabs(self._base_directory) or self._datasource.data_context is None:	×
85	return self._base_directory	×
86	else:
87	return os.path.join(	×
88	self._datasource.data_context.root_directory, self._base_directory
89	)
90
91	def get_available_data_asset_names(self):	3✔
92	if not os.path.isdir(self.base_directory):	×
93	return {"names": [], "is_complete_list": True}	×
94	known_assets = self._get_valid_file_options(base_directory=self.base_directory)	×
95	return {"names": known_assets, "is_complete_list": True}	×
96
97	# TODO: deprecate generator_asset argument
98	def get_available_partition_ids(self, generator_asset=None, data_asset_name=None):	3✔
99	assert (generator_asset and not data_asset_name) or (	×
100	not generator_asset and data_asset_name
101	), "Please provide either generator_asset or data_asset_name."
102	if generator_asset:	×
103	# deprecated-v0.11.0
104	warnings.warn(	×
105	"The 'generator_asset' argument is deprecated as of v0.11.0 and will be removed in v0.16. "
106	"Please use 'data_asset_name' instead.",
107	DeprecationWarning,
108	)
109	data_asset_name = generator_asset	×
110
111	# If the generator asset names a single known file, return ONLY that
112	for extension in self.known_extensions:	×
113	if os.path.isfile(	×
114	os.path.join(self.base_directory, data_asset_name + extension)
115	):
116	return [data_asset_name]	×
117	if os.path.isfile(os.path.join(self.base_directory, data_asset_name)):	×
118	return [data_asset_name]	×
119
120	# Otherwise, subdir files are partition ids
121	return [	×
122	path
123	for (path, type) in self._get_valid_file_options(
124	base_directory=os.path.join(self.base_directory, data_asset_name)
125	)
126	]
127
128	def _build_batch_kwargs(self, batch_parameters):	3✔
129	"""
130
131	Args:
132	batch_parameters:
133
134	Returns:
135	batch_kwargs
136
137	"""
138	try:	×
139	data_asset_name = batch_parameters.pop("data_asset_name")	×
140	except KeyError:	×
141	raise BatchKwargsError(	×
142	"Unable to build BatchKwargs: no name provided in batch_parameters.",
143	batch_kwargs=batch_parameters,
144	)
145
146	if "partition_id" in batch_parameters:	×
147	partition_id = batch_parameters.pop("partition_id")	×
148	# Find the path
149	path = None	×
150	for extension in self.known_extensions:	×
151	if os.path.isfile(	×
152	os.path.join(
153	self.base_directory, data_asset_name, partition_id + extension
154	)
155	):
156	path = os.path.join(	×
157	self.base_directory, data_asset_name, partition_id + extension
158	)
159
160	if path is None:	×
161	logger.warning(	×
162	"Unable to find path with the provided partition; searching for asset-name partitions."
163	)
164	# Fall through to this case in the event that there is not a subdir available, or if partition_id was
165	# not provided
166	if os.path.isfile(os.path.join(self.base_directory, data_asset_name)):	×
167	path = os.path.join(self.base_directory, data_asset_name)	×
168
169	for extension in self.known_extensions:	×
170	if os.path.isfile(	×
171	os.path.join(self.base_directory, data_asset_name + extension)
172	):
173	path = os.path.join(	×
174	self.base_directory, data_asset_name + extension
175	)
176
177	if path is None:	×
178	raise BatchKwargsError(	×
179	f"Unable to build batch kwargs from for asset '{data_asset_name}'",
180	batch_parameters,
181	)
182	return self._build_batch_kwargs_from_path(path, **batch_parameters)	×
183
184	else:
185	return self.yield_batch_kwargs(	×
186	data_asset_name=data_asset_name, **batch_parameters
187	)
188
189	def _get_valid_file_options(self, base_directory=None):	3✔
190	valid_options = []	×
191	if base_directory is None:	×
192	base_directory = self.base_directory	×
193	file_options = os.listdir(base_directory)	×
194	for file_option in file_options:	×
195	for extension in self.known_extensions:	×
196	if (	×
197	file_option.endswith(extension)
198	and not file_option.startswith(".")
199	and (file_option[: -len(extension)], "file") not in valid_options
200	):
201	valid_options.append((file_option[: -len(extension)], "file"))	×
202	elif os.path.isdir(os.path.join(self.base_directory, file_option)):	×
203	# Make sure there's at least one valid file inside the subdir
204	subdir_options = self._get_valid_file_options(	×
205	base_directory=os.path.join(base_directory, file_option)
206	)
207	if (	×
208	len(subdir_options) > 0
209	and (file_option, "directory") not in valid_options
210	):
211	valid_options.append((file_option, "directory"))	×
212	return valid_options	×
213
214	def _get_iterator(self, data_asset_name, reader_options=None, limit=None):	3✔
215	logger.debug(	×
216	f"Beginning SubdirReaderBatchKwargsGenerator _get_iterator for data_asset_name: {data_asset_name}"
217	)
218	# If the data asset is a file, then return the path.
219	# Otherwise, use files in a subdir as batches
220	if os.path.isdir(os.path.join(self.base_directory, data_asset_name)):	×
221	subdir_options = os.listdir(	×
222	os.path.join(self.base_directory, data_asset_name)
223	)
224	batches = []	×
225	for file_option in subdir_options:	×
226	for extension in self.known_extensions:	×
227	if file_option.endswith(extension) and not file_option.startswith(	×
228	"."
229	):
230	batches.append(	×
231	os.path.join(
232	self.base_directory, data_asset_name, file_option
233	)
234	)
235
236	return self._build_batch_kwargs_path_iter(	×
237	batches, reader_options=reader_options, limit=limit
238	)
239	else:
240	for extension in self.known_extensions:	×
241	path = os.path.join(self.base_directory, data_asset_name + extension)	×
242	if os.path.isfile(path):	×
243	return iter(	×
244	[
245	self._build_batch_kwargs_from_path(
246	path, reader_options=reader_options, limit=limit
247	)
248	]
249	)
250	# If we haven't returned yet, raise
251	raise BatchKwargsError(	×
252	"No valid files found when searching {:s} using configured known_extensions: "
253	"{:s} ".format(
254	os.path.join(self.base_directory, data_asset_name),
255	", ".join(map(str, self.known_extensions)),
256	),
257	batch_kwargs=PathBatchKwargs(
258	path=os.path.join(self.base_directory, data_asset_name)
259	),
260	)
261
262	def _build_batch_kwargs_path_iter(	3✔
263	self, path_list, reader_options=None, limit=None
264	) -> Iterable[PathBatchKwargs]:
265	for path in path_list:	×
266	yield self._build_batch_kwargs_from_path(	×
267	path, reader_options=reader_options, limit=limit
268	)
269
270	def _build_batch_kwargs_from_path(	3✔
271	self, path, reader_method=None, reader_options=None, limit=None
272	):
273	batch_kwargs = self._datasource.process_batch_parameters(	×
274	reader_method=reader_method or self.reader_method,
275	reader_options=reader_options or self.reader_options,
276	limit=limit,
277	)
278	batch_kwargs["path"] = path	×
279	batch_kwargs["datasource"] = self._datasource.name	×
280	return PathBatchKwargs(batch_kwargs)	×

great-expectations / great_expectations / 20220928.5

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous