Verify gcs target

environment.yml

@@ -5,7 +5,7 @@ dependencies:
   - python=3.8
   - cdsapi
   - ecmwf-api-client
-  - apache-beam[gcp]
+  - apache-beam[gcp]==2.40.0
   - numpy
   - pandas
   - xarray

setup.py

@@ -14,10 +14,26 @@
 
 from setuptools import find_packages, setup
 
+beam_gcp_requirements = [
+    "google-cloud-bigquery==2.34.4",
+    "google-cloud-bigquery-storage==2.14.1",
+    "google-cloud-bigtable==1.7.2",
+    "google-cloud-core==1.7.3",
+    "google-cloud-datastore==1.15.5",
+    "google-cloud-dlp==3.8.0",
+    "google-cloud-language==1.3.2",
+    "google-cloud-pubsub==2.13.4",
+    "google-cloud-pubsublite==1.4.2",
+    "google-cloud-recommendations-ai==0.2.0",
+    "google-cloud-spanner==1.19.3",
+    "google-cloud-videointelligence==1.16.3",
+    "google-cloud-vision==1.0.2",
+    "apache-beam[gcp]==2.40.0",
+]
+
 weather_dl_requirements = [
     "cdsapi",
     "ecmwf-api-client",
-    "apache-beam[gcp]",
     "numpy>=1.19.1",
     "pandas",
     "xarray",
@@ -28,12 +44,10 @@
 ]
 
 weather_mv_requirements = [
-    "apache-beam[gcp]",
     "dataclasses",
     "numpy",
     "pandas",
     "xarray",
-    "google-cloud-storage==2.2.1",
     "cfgrib",
     "netcdf4",
     "geojson",
@@ -47,7 +61,6 @@
 ]
 
 weather_sp_requirements = [
-    "apache-beam[gcp]",
     "numpy>=1.20.3",
     "pygrib",
     "xarray",
@@ -67,7 +80,9 @@
     "metview",
 ]
 
-all_test_requirements = weather_dl_requirements + weather_mv_requirements + weather_sp_requirements + test_requirements
+all_test_requirements = beam_gcp_requirements + weather_dl_requirements + \
+                        weather_mv_requirements + weather_sp_requirements + \
+                        test_requirements
 
 setup(
     name='google-weather-tools',
@@ -97,7 +112,7 @@
 
     ],
     python_requires='>=3.7, <3.10',
-    install_requires=['apache-beam[gcp]'],
+    install_requires=['apache-beam[gcp]==2.40.0'],
     use_scm_version=True,
     setup_requires=['setuptools_scm'],
     scripts=['weather_dl/weather-dl', 'weather_dl/download-status',

weather_dl/download_pipeline/config.py

@@ -12,8 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import calendar
-import copy
+import copy as cp
 import dataclasses
+import itertools
 import typing as t
 
 Values = t.Union[t.List['Values'], t.Dict[str, 'Values'], bool, int, float, str]  # pytype: disable=not-supported-yet
@@ -78,7 +79,7 @@ def optimize_selection_partition(selection: t.Dict) -> t.Dict:
 
     Used to support custom syntax and optimizations, such as 'all'.
     """
-    selection_ = copy.deepcopy(selection)
+    selection_ = cp.deepcopy(selection)
 
     if 'day' in selection_.keys() and selection_['day'] == 'all':
         year, month = selection_['year'], selection_['month']
@@ -109,3 +110,46 @@ def optimize_selection_partition(selection: t.Dict) -> t.Dict:
         del selection_['year']
 
     return selection_
+
+
+def prepare_partitions(config: Config) -> t.Iterator[Config]:
+    """Iterate over client parameters, partitioning over `partition_keys`.
+
+    This produces a Cartesian-Cross over the range of keys.
+
+    For example, if the keys were 'year' and 'month', it would produce
+    an iterable like:
+        ( ('2020', '01'), ('2020', '02'), ('2020', '03'), ...)
+
+    Returns:
+        An iterator of `Config`s.
+    """
+    for option in itertools.product(*[config.selection[key] for key in config.partition_keys]):
+        yield _create_partition_config(option, config)
+
+
+def _create_partition_config(option: t.Tuple, config: Config) -> Config:
+    """Create a config for a single partition option.
+
+    Output a config dictionary, overriding the range of values for
+    each key with the partition instance in 'selection'.
+    Continuing the example from prepare_partitions, the selection section
+    would be:
+      { 'foo': ..., 'year': ['2020'], 'month': ['01'], ... }
+      { 'foo': ..., 'year': ['2020'], 'month': ['02'], ... }
+      { 'foo': ..., 'year': ['2020'], 'month': ['03'], ... }
+
+    Args:
+        option: A single item in the range of partition_keys.
+        config: The download config, including the parameters and selection sections.
+
+    Returns:
+        A configuration with that selects a single download partition.
+    """
+    copy = cp.deepcopy(config.selection)
+    out = cp.deepcopy(config)
+    for idx, key in enumerate(config.partition_keys):
+        copy[key] = [option[idx]]
+
+    out.selection = copy
+    return out

weather_dl/download_pipeline/parsers.py

@@ -18,17 +18,24 @@
 import copy as cp
 import datetime
 import json
+import logging
 import string
 import textwrap
 import typing as t
 import numpy as np
 from collections import OrderedDict
 from urllib.parse import urlparse
 
+from apache_beam.io.aws.s3io import S3IO
+from apache_beam.io.filesystems import FileSystems
+from apache_beam.io.gcp.gcsio import GcsIO
+
 from .clients import CLIENTS
-from .config import Config
+from .config import Config, prepare_partitions
 from .manifest import MANIFESTS, Manifest, Location, NoOpManifest
 
+logger = logging.getLogger(__name__)
+
 
 def date(candidate: str) -> datetime.date:
     """Converts ECMWF-format date strings into a `datetime.date`.
@@ -326,6 +333,29 @@ def _number_of_replacements(s: t.Text):
     return len(set(format_names)) + num_empty_names
 
 
+def _validate_target_path(config: Config):
+    """Checks accessibility of target locations before execution start"""
+    for partition_conf in prepare_partitions(config):
+        target = prepare_target_name(partition_conf)
+        parsed = urlparse(target)
+        if FileSystems.exists(target):
+            raise ValueError(f"Path {target} already exists.")
+        try:
+            if parsed.scheme == "gs":
+                GcsIO().open(target, 'w').close()
+            elif parsed.scheme == "s3":
+                S3IO().open(target, 'w').close()
+            elif parsed.scheme == "" or parsed.scheme == "fs":
+                open(target, 'w').close()
+            else:
+                logger.warning(f"Unable to verify path {target} writeable."
+                               f"Persistent IO errors when writing to the target"
+                               f"path during execution may cause deadlocks.")
+        except Exception:
+            raise ValueError(f"Unable to write to {target}")
+        FileSystems.delete([target])
+
+
 def parse_subsections(config: t.Dict) -> t.Dict:
     """Interprets [section.subsection] as nested dictionaries in `.cfg` files."""
     copy = cp.deepcopy(config)
@@ -441,13 +471,19 @@ def require(condition: bool, message: str, error_type: t.Type[Exception] = Value
         if not isinstance(selection[key], list):
             selection[key] = [selection[key]]
 
-    return Config.from_dict(config)
+    config = Config.from_dict(config)
+
+    _validate_target_path(config)
+
+    return config
 
 
-def prepare_target_name(config: Config) -> str:
+def prepare_target_name(partition_config: Config) -> str:
     """Returns name of target location."""
-    partition_dict = OrderedDict((key, typecast(key, config.selection[key][0])) for key in config.partition_keys)
-    target = config.target_path.format(*partition_dict.values(), **partition_dict)
+    partition_dict = OrderedDict(
+        (key, typecast(key, partition_config.selection[key][0]))
+        for key in partition_config.partition_keys)
+    target = partition_config.target_path.format(*partition_dict.values(), **partition_dict)
 
     return target
 

weather_dl/download_pipeline/partition.py

@@ -11,14 +11,14 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-import copy as cp
 import dataclasses
 import itertools
 import logging
 import typing as t
 
 import apache_beam as beam
 
+from .config import prepare_partitions
 from .manifest import Manifest
 from .parsers import prepare_target_name
 from .config import Config
@@ -84,33 +84,6 @@ def loop_through_subsections(it: Config) -> Partition:
         )
 
 
-def _create_partition_config(option: t.Tuple, config: Config) -> Config:
-    """Create a config for a single partition option.
-
-    Output a config dictionary, overriding the range of values for
-    each key with the partition instance in 'selection'.
-    Continuing the example from prepare_partitions, the selection section
-    would be:
-      { 'foo': ..., 'year': ['2020'], 'month': ['01'], ... }
-      { 'foo': ..., 'year': ['2020'], 'month': ['02'], ... }
-      { 'foo': ..., 'year': ['2020'], 'month': ['03'], ... }
-
-    Args:
-        option: A single item in the range of partition_keys.
-        config: The download config, including the parameters and selection sections.
-
-    Returns:
-        A configuration with that selects a single download partition.
-    """
-    copy = cp.deepcopy(config.selection)
-    out = cp.deepcopy(config)
-    for idx, key in enumerate(config.partition_keys):
-        copy[key] = [option[idx]]
-
-    out.selection = copy
-    return out
-
-
 def skip_partition(config: Config, store: Store) -> bool:
     """Return true if partition should be skipped."""
 
@@ -125,22 +98,6 @@ def skip_partition(config: Config, store: Store) -> bool:
     return False
 
 
-def prepare_partitions(config: Config) -> t.Iterator[Config]:
-    """Iterate over client parameters, partitioning over `partition_keys`.
-
-    This produces a Cartesian-Cross over the range of keys.
-
-    For example, if the keys were 'year' and 'month', it would produce
-    an iterable like:
-        ( ('2020', '01'), ('2020', '02'), ('2020', '03'), ...)
-
-    Returns:
-        An iterator of `Config`s.
-    """
-    for option in itertools.product(*[config.selection[key] for key in config.partition_keys]):
-        yield _create_partition_config(option, config)
-
-
 def new_downloads_only(candidate: Config, store: t.Optional[Store] = None) -> bool:
     """Predicate function to skip already downloaded partitions."""
     if store is None:

weather_dl/setup.py

@@ -14,17 +14,33 @@
 
 from setuptools import setup, find_packages
 
+beam_gcp_requirements = [
+    "google-cloud-bigquery==2.34.4",
+    "google-cloud-bigquery-storage==2.14.1",
+    "google-cloud-bigtable==1.7.2",
+    "google-cloud-core==1.7.3",
+    "google-cloud-datastore==1.15.5",
+    "google-cloud-dlp==3.8.0",
+    "google-cloud-language==1.3.2",
+    "google-cloud-pubsub==2.13.4",
+    "google-cloud-pubsublite==1.4.2",
+    "google-cloud-recommendations-ai==0.2.0",
+    "google-cloud-spanner==1.19.3",
+    "google-cloud-videointelligence==1.16.3",
+    "google-cloud-vision==1.0.2",
+    "apache-beam[gcp]==2.40.0",
+]
+
 base_requirements = [
     "cdsapi",
     "ecmwf-api-client",
-    "apache-beam[gcp]",
     "numpy>=1.19.1",
     "pandas",
     "xarray",
     "requests>=2.24.0",
     "firebase-admin>=5.0.0",
-    "google-cloud-firestore",
     "urllib3==1.26.5",
+    "google-cloud-firestore==2.6.0",
 ]
 
 setup(
@@ -35,5 +51,5 @@
     author_email='anthromets-ecmwf@google.com',
     url='https://weather-tools.readthedocs.io/en/latest/weather_dl/',
     description='A tool to download weather data.',
-    install_requires=base_requirements,
+    install_requires=beam_gcp_requirements + base_requirements,
 )

weather_mv/setup.py

@@ -26,13 +26,28 @@
 
 from setuptools import setup, find_packages, Command
 
+beam_gcp_requirements = [
+    "google-cloud-bigquery==2.34.4",
+    "google-cloud-bigquery-storage==2.14.1",
+    "google-cloud-bigtable==1.7.2",
+    "google-cloud-core==1.7.3",
+    "google-cloud-datastore==1.15.5",
+    "google-cloud-dlp==3.8.0",
+    "google-cloud-language==1.3.2",
+    "google-cloud-pubsub==2.13.4",
+    "google-cloud-pubsublite==1.4.2",
+    "google-cloud-recommendations-ai==0.2.0",
+    "google-cloud-spanner==1.19.3",
+    "google-cloud-videointelligence==1.16.3",
+    "google-cloud-vision==1.0.2",
+    "apache-beam[gcp]==2.40.0",
+]
+
 base_requirements = [
-    "apache-beam[gcp]",
     "dataclasses",
     "numpy",
     "pandas",
     "xarray",
-    "google-cloud-storage==2.2.1",
     "cfgrib",
     "netcdf4",
     "geojson",
@@ -133,7 +148,7 @@ def run(self):
     version='0.2.5',
     url='https://weather-tools.readthedocs.io/en/latest/weather_mv/',
     description='A tool to load weather data into BigQuery.',
-    install_requires=base_requirements,
+    install_requires=beam_gcp_requirements + base_requirements,
     cmdclass={
         # Command class instantiated and run during pip install scenarios.
         'build': build,