(destination_config: DestinationTestConfiguration)
| 40 | ids=lambda x: x.name, |
| 41 | ) |
| 42 | def test_simple_incremental(destination_config: DestinationTestConfiguration) -> None: |
| 43 | pipeline = destination_config.setup_pipeline("test_simple_incremental", dev_mode=True) |
| 44 | |
| 45 | pipeline.run( |
| 46 | [{"a": i, "b": i + 1} for i in range(10)], |
| 47 | table_name="example_table", |
| 48 | **destination_config.run_kwargs, |
| 49 | ) |
| 50 | dataset = pipeline.dataset() |
| 51 | |
| 52 | example_table_columns = dataset.schema.tables["example_table"]["columns"] |
| 53 | |
| 54 | @dlt.resource() |
| 55 | def copied_table( |
| 56 | cursor: dlt.sources.incremental[int] = dlt.sources.incremental( |
| 57 | "a", initial_value=2, end_value=100, on_cursor_value_missing="exclude" |
| 58 | ), |
| 59 | ) -> Any: |
| 60 | rel = dataset["example_table"].incremental(cursor) |
| 61 | yield dlt.mark.with_hints( |
| 62 | rel, |
| 63 | hints=make_hints(columns=example_table_columns), |
| 64 | ) |
| 65 | |
| 66 | info = pipeline.run( |
| 67 | [copied_table()], |
| 68 | loader_file_format="model", |
| 69 | table_format=destination_config.run_kwargs["table_format"], |
| 70 | ) |
| 71 | assert_load_info(info) |
| 72 | assert load_table_counts(pipeline, "copied_table") == {"copied_table": 8} |
| 73 | |
| 74 | |
| 75 | @pytest.mark.parametrize( |
nothing calls this directly
no test coverage detected