from pathlib import Path
import os
from datetime import datetime
import dlt
from dlt.sources.rest_api import (
    rest_api_resources,
    RESTAPIConfig,
    ClientConfig,
    EndpointResource,
    Endpoint,
)
from dlt.sources.rest_api.typing import IncrementalConfig

DEST_PATH = Path.home() / "example"
os.environ["DESTINATION__FILESYSTEM__BUCKET_URL"] = f"file:///{DEST_PATH}"
os.environ["NORMALIZE__DATA_WRITER__DISABLE_COMPRESSION"] = "true"


@dlt.source
def gemini_btc():
    config = RESTAPIConfig(
        client=ClientConfig(base_url="https://api.gemini.com/v1"),
        resources=[
            EndpointResource(
                name="btcusd_trades",
                write_disposition="append",
                endpoint=Endpoint(
                    path="trades/btcusd",
                    incremental=IncrementalConfig(
                        allow_external_schedulers=True,
                        cursor_path="timestamp",
                        initial_value=datetime(2024, 12, 21).timestamp(),
                        start_param="timestamp",
                    ),
                ),
            ),
        ],
    )
    return rest_api_resources(config)


if __name__ == "__main__":
    pipeline = dlt.pipeline(
        pipeline_name="gemini_btc",
        dataset_name="gemini_btc",
        destination="filesystem",
    )
    pipeline.run(gemini_btc())