From 348d9ef184e9890b7e55df262cdf0501261918d1 Mon Sep 17 00:00:00 2001 From: Artur Nowosielski Date: Thu, 13 Jan 2022 18:34:12 +0100 Subject: [PATCH] FST-250 bf build arguments validation --- bigflow/build/operate.py | 52 +++++++++++-------- bigflow/cli.py | 17 +----- bigflow/commons.py | 21 ++++++++ .../bf_simple_pytest/workflow.py | 6 +++ test/test_commons.py | 39 ++++++++++++++ 5 files changed, 97 insertions(+), 38 deletions(-) create mode 100644 test/bf-projects/bf_simple_pytest/bf_simple_pytest/workflow.py diff --git a/bigflow/build/operate.py b/bigflow/build/operate.py index e600deb1..20a3015b 100644 --- a/bigflow/build/operate.py +++ b/bigflow/build/operate.py @@ -114,17 +114,11 @@ def build_dags( start_time: str, workflow_id: typing.Optional[str] = None, ): - logger.info("Building airflow DAGs...") - clear_dags_leftovers(project_spec) - - image_version = bf_commons.build_docker_image_tag(project_spec.docker_repository, project_spec.version) - create_image_version_file(str(project_spec.project_dir), image_version) - - # TODO: Move common frunctions from bigflow.cli to bigflow.commons (or other shared module) - from bigflow.cli import _valid_datetime, walk_workflows - _valid_datetime(start_time) + # TODO: Move common functions from bigflow.cli to bigflow.commons (or other shared module) + from bigflow.cli import walk_workflows - cnt = 0 + logger.debug('Loading workflow(s)...') + workflows = [] for root_package in project_spec.packages: if "." in root_package: # leaf package @@ -133,18 +127,32 @@ def build_dags( for workflow in walk_workflows(project_spec.project_dir / root_package): if workflow_id is not None and workflow_id != workflow.workflow_id: continue - logger.info("Generating DAG file for %s", workflow.workflow_id) - cnt += 1 - bigflow.dagbuilder.generate_dag_file( - str(project_spec.project_dir), - image_version, - workflow, - start_time, - project_spec.version, - root_package, - ) + workflows.append((workflow, root_package)) + + if not workflows: + if not workflow_id: + raise Exception('No workflow found') + else: + raise Exception("Workflow '{}' not found".format(workflow_id)) + + logger.info("Building airflow DAGs...") + clear_dags_leftovers(project_spec) + + image_version = bf_commons.build_docker_image_tag(project_spec.docker_repository, project_spec.version) + create_image_version_file(str(project_spec.project_dir), image_version) + + for (workflow, package) in workflows: + logger.info("Generating DAG file for %s", workflow.workflow_id) + bigflow.dagbuilder.generate_dag_file( + str(project_spec.project_dir), + image_version, + workflow, + start_time, + project_spec.version, + package, + ) - logger.info("Geneated %d DAG files", cnt) + logger.info("Generated %d DAG files", len(workflows)) def _rmtree(p: Path): @@ -194,7 +202,7 @@ def build_project( workflow_id: typing.Optional[str] = None, ): logger.info("Build the project") + build_dags(project_spec, start_time, workflow_id=workflow_id) build_package(project_spec) build_image(project_spec) - build_dags(project_spec, start_time, workflow_id=workflow_id) logger.info("Project was built") diff --git a/bigflow/cli.py b/bigflow/cli.py index 03b5081f..9a2244e8 100644 --- a/bigflow/cli.py +++ b/bigflow/cli.py @@ -303,21 +303,6 @@ def _create_build_package_parser(subparsers): subparsers.add_parser('build-package', description='Builds .whl package from local sources.') -def _valid_datetime(dt): - if dt == 'NOW': - return - - try: - datetime.strptime(dt, "%Y-%m-%d %H:%M:%S") - return dt - except ValueError: - try: - datetime.strptime(dt, "%Y-%m-%d") - return dt - except ValueError: - raise ValueError("Not a valid date: '{0}'.".format(dt)) - - def _add_build_dags_parser_arguments(parser): parser.add_argument('-w', '--workflow', type=str, @@ -330,7 +315,7 @@ def _add_build_dags_parser_arguments(parser): 'For workflows triggered hourly -- datetime in format: Y-m-d H:M:S, for example 2020-01-01 00:00:00. ' 'For workflows triggered daily -- date in format: Y-m-d, for example 2020-01-01. ' 'If empty or set as NOW, current hour is used.', - type=_valid_datetime) + type=bf_commons.valid_datetime) def _create_build_dags_parser(subparsers): diff --git a/bigflow/commons.py b/bigflow/commons.py index 94d58f6b..034ede70 100644 --- a/bigflow/commons.py +++ b/bigflow/commons.py @@ -320,3 +320,24 @@ def as_timedelta(v: None | str | Number | timedelta) -> timedelta | None: return None else: return timedelta(seconds=float(v)) + + +def valid_datetime(dt: str) -> str: + """ + Validates provided datetime string. Raises ValueError for strings that are none of: + * 'NOW' + * valid '%Y-%m-%d %H:%M:%S' + * valid '%Y-%m-%d' + """ + if dt == 'NOW': + return dt + + try: + datetime.strptime(dt, "%Y-%m-%d %H:%M:%S") + except ValueError: + try: + datetime.strptime(dt, "%Y-%m-%d") + except ValueError: + raise ValueError("Not a valid date: '{0}'.".format(dt)) + + return dt diff --git a/test/bf-projects/bf_simple_pytest/bf_simple_pytest/workflow.py b/test/bf-projects/bf_simple_pytest/bf_simple_pytest/workflow.py new file mode 100644 index 00000000..c5b24342 --- /dev/null +++ b/test/bf-projects/bf_simple_pytest/bf_simple_pytest/workflow.py @@ -0,0 +1,6 @@ +import bigflow + +the_workflow = bigflow.Workflow( + workflow_id='the_workflow', + definition=[] +) diff --git a/test/test_commons.py b/test/test_commons.py index 3dbfb889..4b40c6d1 100644 --- a/test/test_commons.py +++ b/test/test_commons.py @@ -43,6 +43,45 @@ def test_should_raise_error_when_given_file_does_not_exists(self): # when decode_version_number_from_file_name(Path('/Users/image-0.1122123.0.tar')) + def test_valid_datetime_should_pass_for_NOW(self): + # when + valid_datetime('NOW') + + # then + # no error is raised + + def test_valid_datetime_should_pass_for_valid_YmdHMS(self): + # when + valid_datetime('2022-01-01 12:34:56') + + # then + # no error is raised + + def test_valid_datetime_should_raise_error_for_invalid_YmdHMS(self): + # then + with self.assertRaises(ValueError): + # when + valid_datetime('2022-01-01 34:56:78') + + def test_valid_datetime_should_pass_for_valid_Ymd(self): + # when + valid_datetime('2022-01-01') + + # then + # no error is raised + + def test_valid_datetime_should_raise_error_for_invalid_Ymd(self): + # then + with self.assertRaises(ValueError): + # when + valid_datetime('2022-23-45') + + def test_valid_datetime_should_raise_error_for_other_string(self): + # then + with self.assertRaises(ValueError): + # when + valid_datetime('foo bar baz') + def _touch_file(self, file_name: str, content: str = ''): workdir = Path(os.getcwd()) f = workdir / file_name