dltLearning/rest_api_pipeline.py at main · kelazarus/dltLearning · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
from typing import Any, Optional

import dlt
from dlt.sources.credentials import ConnectionStringCredentials
from dlt.common.pendulum import pendulum
from dlt.sources.rest_api import (
    RESTAPIConfig,
    check_connection,
    rest_api_resources,
    rest_api_source,
)


def load_test_json() -> None:
    pipeline = dlt.pipeline(
        pipeline_name="rest_json_sample",
        destination='mssql',
        dataset_name="rjs",
    )

    test_json_source = rest_api_source(
        {
            "client": {
                "base_url": "https://samples.json-format.com/employees/json/",
                # If you leave out the paginator, it will be inferred from the API:
                # "paginator": "json_link",
            },
            "resource_defaults": {
                "endpoint": {
                    "params": {
                        "limit": 1000,
                    },
                },
            },
            "resources": [
                "employees_100KB.json",
            ],
        }
    )

    def check_network_and_authentication() -> None:
        (can_connect, error_msg) = check_connection(
            test_json_source,
            # "not_existing_endpoint",
        )
        if not can_connect:
            pass  # do something with the error message

    check_network_and_authentication()

    load_info = pipeline.run(test_json_source)
    print(load_info)  # noqa: T201


if __name__ == "__main__":
    connection_string = "mssql+pyodbc://localhost\\sqlexpress/dlt2?trusted_connection=yes&driver=ODBC Driver 17 for SQL Server"
    credentials = ConnectionStringCredentials(connection_string)
    # load_github()
    load_test_json()