Introduction to Data Modelling

select
  count(*) filter (where facility_id is null)          as null_facility_ids,
  count(*) - count(distinct shipment_id)               as duplicate_shipment_ids,
  count(*) filter (where quantity !~ '^[0-9]+$')       as non_numeric_quantities
from raw.vaccine_shipments;

-- models/staging/stg_vaccine__shipments.sql
with ranked as (
  select
    shipment_id,
    facility_id                                        as health_facility_id,
    cast(replace(quantity, ' doses', '') as integer)   as quantity_doses,
    received_at,
    row_number() over (
      partition by shipment_id
      order by received_at desc
    ) as row_num
  from {{ source('raw', 'vaccine_shipments') }}
)
 
select shipment_id, health_facility_id, quantity_doses, received_at
from ranked
where row_num = 1

-- models/intermediate/int_vaccine_stock_ledger.sql
with shipments as (
  select * from {{ ref('stg_vaccine__shipments') }}
),
 
doses as (
  select * from {{ ref('stg_immunization__doses') }}
),
 
facilities as (
  select * from {{ ref('stg_facility__master') }}
),
 
stock as (
  select * from {{ ref('stg_facility__stock_counts') }}
)
 
select
  s.health_facility_id,
  f.district,
  s.received_at::date as ledger_date,
  s.quantity_doses    as received,
  d.doses_administered,
  d.doses_wasted,
  -- one shared rule, applied once
  k.opening_balance + s.quantity_doses
    - d.doses_administered - d.doses_wasted as closing_balance
from shipments s
join doses d
  on d.health_facility_id = s.health_facility_id
 and d.dose_date = s.received_at::date
join stock k
  on k.health_facility_id = s.health_facility_id
 and k.count_date = s.received_at::date
join facilities f
  on f.health_facility_id = s.health_facility_id

-- models/marts/fct_vaccine_stock_movements.sql
with ledger as (
  select * from {{ ref('int_vaccine_stock_ledger') }}
),
 
-- unpivot the daily ledger into one row per stock transaction
movements as (
  select health_facility_id, ledger_date,
         'receipt' as transaction_type, received as quantity_doses
  from ledger
  union all
  select health_facility_id, ledger_date,
         'administered', doses_administered
  from ledger
  union all
  select health_facility_id, ledger_date,
         'adjustment', doses_wasted
  from ledger
)
 
select
  movements.health_facility_id,
  facility.province,
  facility.district,
  facility.facility_type,
  movements.ledger_date,
  movements.transaction_type,
  movements.quantity_doses
from movements
join {{ ref('dim_health_facility') }} as facility
  on facility.health_facility_id = movements.health_facility_id

# models/marts/schema.yml
models:
  - name: fct_vaccine_stock_movements
    columns:
      - name: health_facility_id
        description: Facility receiving or issuing the stock movement.
        tests:
          - not_null
          - relationships:
              to: ref('dim_health_facility')
              field: health_facility_id
      - name: transaction_type
        description: Kind of stock movement recorded.
        tests:
          - accepted_values:
              values: [issue, receipt, administered, adjustment]
 
  - name: dim_health_facility
    columns:
      - name: health_facility_id
        description: Unique identifier for the health facility.
        tests:
          - not_null
          - unique