diff --git a/dev-project/docker-compose.yml b/dev-project/docker-compose.yml index 6803f05a3..b958f2e2d 100644 --- a/dev-project/docker-compose.yml +++ b/dev-project/docker-compose.yml @@ -121,6 +121,7 @@ services: --tlsMode requireTLS --tlsAllowConnectionsWithoutCertificates --tlsCertificateKeyFile /etc/ssl/mongodb.pem + --tlsAllowInvalidHostnames --tlsCAFile /etc/ssl/rootCA.pem --quiet networks: diff --git a/dev-project/mongo/create-pipelinewise-user.sh b/dev-project/mongo/create-pipelinewise-user.sh index ae2fc477b..ac3307721 100755 --- a/dev-project/mongo/create-pipelinewise-user.sh +++ b/dev-project/mongo/create-pipelinewise-user.sh @@ -1,6 +1,6 @@ echo 'CREATE MONGODB PIPELINEWISE USER' -mongo --tls --tlsAllowInvalidCertificates -u $MONGO_INITDB_ROOT_USERNAME -p $MONGO_INITDB_ROOT_PASSWORD --authenticationDatabase admin admin < +.. attention:: + + There is an option to chose tables for re-sync which has a specific replication method by ``--replication_method_only `` + $ pipelinewise sync_tables --target --tap --replication_method_only log_based diff --git a/pipelinewise/cli/__init__.py b/pipelinewise/cli/__init__.py index fa1bea09b..943a0db7c 100644 --- a/pipelinewise/cli/__init__.py +++ b/pipelinewise/cli/__init__.py @@ -253,6 +253,8 @@ def main(): parser.add_argument('--force', default=False, required=False, help='Force sync_tables for full sync', action='store_true' ) + parser.add_argument('--replication_method_only', default='*', type=str, + help='Sync only tables which their replication method is as entered value') args = parser.parse_args() diff --git a/pipelinewise/cli/pipelinewise.py b/pipelinewise/cli/pipelinewise.py index 4696c1330..91a82b7c0 100644 --- a/pipelinewise/cli/pipelinewise.py +++ b/pipelinewise/cli/pipelinewise.py @@ -1397,7 +1397,9 @@ def do_sync_tables(self, fastsync_stream_ids=None): else: tables_to_sync = self.args.tables - selected_tables = self._get_sync_tables_setting_from_selection_file(tables_to_sync) + selected_tables = self._get_sync_tables_setting_from_selection_file( + tables_to_sync, self.args.replication_method_only) + processes_list = [] if selected_tables['partial_sync']: self._reset_state_file_for_partial_sync(selected_tables) @@ -2098,7 +2100,8 @@ def _clean_tables_from_bookmarks_in_state_file(state_file_to_clean: str, tables: def _get_fixed_name_of_table(stream_id): return stream_id.replace('-', '.', 1) - def _get_sync_tables_setting_from_selection_file(self, tables): + def _get_sync_tables_setting_from_selection_file(self, tables, replication_method_only='*'): + replication_method = replication_method_only.upper() selection = utils.load_json(self.tap['files']['selection']) selection = selection.get('selection') all_tables = {'full_sync': [], 'partial_sync': {}} @@ -2107,10 +2110,11 @@ def _get_sync_tables_setting_from_selection_file(self, tables): for table in selection: table_name = self._get_fixed_name_of_table(table['tap_stream_id']) if tables_list is None or table_name in tables_list: - if table.get('sync_start_from'): - all_tables['partial_sync'][table_name] = table['sync_start_from'] - else: - all_tables['full_sync'].append(table_name) + if replication_method in ['*', table.get('replication_method')]: + if table.get('sync_start_from'): + all_tables['partial_sync'][table_name] = table['sync_start_from'] + else: + all_tables['full_sync'].append(table_name) return all_tables def __check_if_table_is_selected(self, table_in_properties): diff --git a/scripts/publish_docs.sh b/scripts/publish_docs.sh index b51d3bb04..aa58025bd 100755 --- a/scripts/publish_docs.sh +++ b/scripts/publish_docs.sh @@ -33,7 +33,7 @@ if [ -z "$GH_TOKEN" ]; then fi # Install dependencies in a virtual env -python3.8 -m venv ~/venv-doc +python3 -m venv ~/venv-doc . ~/venv-doc/bin/activate pip install --upgrade pip pip install sphinx sphinx-rtd-theme diff --git a/tests/units/cli/cli_args.py b/tests/units/cli/cli_args.py index c8204ea12..795f5b3bb 100644 --- a/tests/units/cli/cli_args.py +++ b/tests/units/cli/cli_args.py @@ -21,7 +21,8 @@ def __init__( extra_log=False, debug=False, profiler=False, - force=False + force=False, + replication_method_only='*' ): self.target = target self.tap = tap @@ -36,6 +37,7 @@ def __init__( self.debug = debug self.profiler = profiler self.force = force + self.replication_method_only = replication_method_only # "log" Getters and setters @property diff --git a/tests/units/cli/test_cli.py b/tests/units/cli/test_cli.py index 4f4a259d7..366df11c7 100644 --- a/tests/units/cli/test_cli.py +++ b/tests/units/cli/test_cli.py @@ -1077,3 +1077,33 @@ def test_get_sync_tables_setting_from_selection_file_if_tables_parameter_is_none 'partial_sync': {'baz': 'PARTIAL', 'par': 'PARTIAL_par'} } assert actual_selected_tables == expected_selected_tables + + def test_get_sync_tables_if_using_replication_method_only(self): + """Test if the method for getting list of tables for syncing returns only tables with selected + replication method""" + tables = 'foo,foo_bar,foo_bar_baz,bar_baz,bar,par,par_foo' + with patch('pipelinewise.cli.pipelinewise.utils.load_json') as mocked_load_json: + self.pipelinewise.tap = { + 'files': { + 'selection': 'foo.json' + } + } + mocked_load_json.return_value = { + 'selection': [ + {'tap_stream_id': 'foo', 'replication_method': 'FULL_TABLE'}, + {'tap_stream_id': 'foo_bar', 'replication_method': 'LOG_BASED'}, + {'tap_stream_id': 'foo_bar_baz', 'replication_method': 'INCREMENTAL'}, + {'tap_stream_id': 'bar_baz', 'replication_method': 'FULL_TABLE'}, + {'tap_stream_id': 'bar', 'replication_method': 'LOG_BASED'}, + {'tap_stream_id': 'par', 'sync_start_from': 'PARTIAL_par', 'replication_method': 'INCREMENTAL'}, + {'tap_stream_id': 'par_foo', 'sync_start_from': 'PARTIAL_par_foo', + 'replication_method': 'LOG_BASED'} + ] + } + actual_selected_tables = self.pipelinewise._get_sync_tables_setting_from_selection_file( + tables, replication_method_only='log_based') + expected_selected_tables = { + 'full_sync': ['foo_bar', 'bar'], + 'partial_sync': {'par_foo': 'PARTIAL_par_foo'} + } + assert actual_selected_tables == expected_selected_tables