pipeline_id: ingestion_api_spku_air_quality
description: >-
  REST API Ingestion for SPKU Realtime Air Quality Data -> MinIO S3 Bronze ->
  DuckDB Iceberg Silver -> GE Quality Suite -> dbt Trino Gold Mart
schedule: 0 4 * * *
mode: flexible
tasks:
  - task_id: bronze_spku_api_source_task
    type: source_api
    label: Bronze SPKU API Source
    dependencies: []
    config:
      source_type: api
      connection_id: conn_spku_api_service
      api_endpoint: https://api.spku.dlh.id/v1/stasiun/spku/data-realtime
      http_method: GET
      target:
        bucket: dlh-bronze
        format: json
  - task_id: bronze_spku_minio_storage_task
    type: minio_storage
    label: MinIO Bronze Storage
    dependencies:
      - bronze_spku_api_source_task
    config:
      bucket: dlh-bronze
      format: json
  - task_id: silver_spku_duckdb_iceberg_task
    type: duckdb_transform
    label: DuckDB Silver Iceberg Transform
    dependencies:
      - bronze_spku_minio_storage_task
    config:
      source_parquet_path: s3://dlh-bronze/spku_realtime.parquet
      target_table: dlh_lakehouse.silver.spku_realtime
      primary_keys:
        - stasiun_id
        - waktu
      select_models: stg_spku_realtime
  - task_id: quality_spku_ge_suite_task
    type: ge_quality
    label: GE Quality (spku_api_suite)
    dependencies:
      - silver_spku_duckdb_iceberg_task
    config:
      expectation_suite: spku_api_suite
  - task_id: gold_spku_dbt_mart_task
    type: dbt_transform
    label: dbt Gold Mart (fct_spku_daily_mart)
    dependencies:
      - quality_spku_ge_suite_task
    config:
      dbt_layer: gold
      dbt_model: fct_spku_daily_mart