from dagster import asset, repository
from dagster_airbyte import (
    AirbyteManagedElementReconciler,
    airbyte_resource,
    AirbyteConnection,
    AirbyteSyncMode,
    load_assets_from_connections,
)
from dagster_airbyte.managed.generated.sources import PokeapiSource
from dagster_airbyte.managed.generated.destinations import BigqueryDestination
airbyte_instance = airbyte_resource.configured(
    {
        "host": HOST,
        "port": "8000",
        "username": "airbyte",
        "password": "password",
    }
)
pokeapi_source = PokeapiSource(
    name="pokeapi_to_bq",
    pokemon_name="snorlax",
)
bigquery_destination = BigqueryDestination(
    name="bigquery",
    project_id="winter-cab-380906",
    dataset_location="asia-southeast2",
    dataset_id="airbyte_testing",
    loading_method=BigqueryDestination.StandardInserts(),
    credentials_json='credentials.json', # dari mana?
)
stargazer_connection = AirbyteConnection(
    name="fetch_stargazer",
    source=pokeapi_source,
    destination=bigquery_destination,
    stream_config={"stargazers": AirbyteSyncMode.incremental_append_dedup()},
    normalize_data=True,
)
airbyte_reconciler = AirbyteManagedElementReconciler(
    airbyte=airbyte_instance,
    connections=[stargazer_connection],
)
# load airbyte connection from above pythonic definitions
airbyte_assets = load_assets_from_connections(
    airbyte=airbyte_instance,
    connections=[stargazer_connection],
    key_prefix=["bigquery"],
)