Pharmacy Stock Expiry Alerts with Kestra and DuckDB

id: pharmacy-stock-expiry-alerts
namespace: company.team
description: |
  Check a pharmacy stock file every week for medicine batches that are already
  expired, expire soon, or have stock records that cannot be trusted (negative
  quantities, unreadable expiry dates, the same batch listed twice), and report
  the money at risk before the stock has to be written off.

inputs:
  - id: stock_file
    type: FILE
    required: false
    description: CSV with the columns branch, sku, product_name, batch_no, expiry,
      quantity and unit_cost. The expiry can be a full date (2027-03-31) or the
      month and year printed on the strip (03/2027, read as the last day of that
      month). Leave empty to run against the built-in sample stock.
  - id: critical_days
    type: INT
    defaults: 30
    description: Batches expiring within this many days are critical. They usually
      need to go back to the supplier or be sold first right now.
  - id: watch_days
    type: INT
    defaults: 90
    description: Batches expiring within this many days (but after critical_days)
      are put on the watch list so they are picked first when dispensing.
  - id: notify_slack
    type: BOOL
    defaults: false
    description: Post the findings to Slack when at least one batch is critical or
      has a data error. Requires the SLACK_WEBHOOK_URL secret.

triggers:
  - id: weekly_expiry_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the stock every Monday at 08:00 IST, before the week's
      purchase order goes to the distributor. Shipped disabled. Enable it once
      the flow reads your real stock export instead of the sample.
    cron: "0 8 * * MON"
    timezone: Asia/Kolkata
    disabled: true

tasks:
  - id: sample_stock
    type: io.kestra.plugin.core.storage.Write
    description: A small two-branch stock file with one example of every problem the
      check catches. Expiry dates are computed relative to today so the demo
      always shows the same mix. Only used when no stock_file is provided.
    extension: .csv
    content: |
      branch,sku,product_name,batch_no,expiry,quantity,unit_cost
      Main Road,PCM500,Paracetamol 500mg strip of 15,PB2401,{{ now() | dateAdd(400, 'DAYS') | date('yyyy-MM-dd') }},120,18.50
      Main Road,AMX250,Amoxicillin 250mg capsules strip of 10,AX1187,{{ now() | dateAdd(-12, 'DAYS') | date('yyyy-MM-dd') }},40,62.00
      Main Road,ORS21,ORS sachet 21g,OR3302,{{ now() | dateAdd(18, 'DAYS') | date('yyyy-MM-dd') }},200,21.00
      Main Road,INS100,Insulin glargine 100IU/ml pen,IG0912,{{ now() | dateAdd(55, 'DAYS') | date('yyyy-MM-dd') }},6,780.00
      Main Road,MTF500,Metformin 500mg strip of 10,MF2207,{{ now() | dateAdd(-40, 'DAYS') | date('yyyy-MM-dd') }},0,25.00
      Main Road,CGH100,Cough syrup 100ml,CS0098,31-13-2026,35,95.00
      Lake View,CTZ10,Cetirizine 10mg strip of 10,CT7781,{{ now() | dateAdd(160, 'DAYS') | date('MM/yyyy') }},80,22.00
      Lake View,AZT500,Azithromycin 500mg strip of 3,AZ5520,{{ now() | dateAdd(25, 'DAYS') | date('yyyy-MM-dd') }},30,112.00
      Lake View,VTD60K,Vitamin D3 60000IU capsules strip of 4,VD1104,{{ now() | dateAdd(200, 'DAYS') | date('yyyy-MM-dd') }},-4,46.00
      Lake View,PAN40,Pantoprazole 40mg strip of 15,PN4410,{{ now() | dateAdd(300, 'DAYS') | date('yyyy-MM-dd') }},50,105.00
      Lake View,PAN40,Pantoprazole 40mg strip of 15,PN4410,{{ now() | dateAdd(120, 'DAYS') | date('yyyy-MM-dd') }},20,105.00
      Lake View,ASP75,Aspirin 75mg strip of 14,AS3319,{{ now() | dateAdd(80, 'DAYS') | date('yyyy-MM-dd') }},60,8.00

  - id: check_stock
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Parse every batch, work out the days left until expiry and the
      value of the stock, and classify each batch as CRITICAL, DATA_ERROR, WATCH
      or OK in DuckDB. The full per-batch report is exported to CSV and the last
      query returns the summary.
    inputFiles:
      stock.csv: "{{ inputs.stock_file ?? outputs.sample_stock.uri }}"
    outputFiles:
      - expiry_report.csv
    fetchType: FETCH_ONE
    sql: |
      CREATE TABLE stock AS
      SELECT
        trim(branch) AS branch,
        trim(sku) AS sku,
        trim(product_name) AS product_name,
        trim(batch_no) AS batch_no,
        trim(coalesce(expiry, '')) AS expiry_raw,
        quantity AS quantity_raw,
        try_cast(quantity AS INTEGER) AS quantity,
        try_cast(unit_cost AS DECIMAL(12, 2)) AS unit_cost
      FROM read_csv('{{ workingDir }}/stock.csv', header = true, all_varchar = true);

      CREATE TABLE parsed AS
      SELECT
        *,
        coalesce(
          try_strptime(expiry_raw, '%Y-%m-%d')::DATE,
          last_day(try_strptime('01/' || expiry_raw, '%d/%m/%Y')::DATE)
        ) AS expiry_date,
        count(*) OVER (PARTITION BY branch, sku, batch_no) AS times_listed
      FROM stock;

      CREATE TABLE report AS
      SELECT
        branch,
        sku,
        product_name,
        batch_no,
        expiry_raw,
        expiry_date,
        days_left,
        quantity,
        unit_cost,
        stock_value,
        status,
        CASE status
          WHEN 'DATA_ERROR' THEN array_to_string(data_errors, ', ')
          WHEN 'CRITICAL' THEN CASE
            WHEN days_left < 0 THEN 'Expired ' || -days_left || ' day(s) ago with ' || quantity || ' unit(s) still in stock: quarantine and remove from sale'
            ELSE 'Expires in ' || days_left || ' day(s) with ' || quantity || ' unit(s) in stock: return to the supplier or sell first now'
          END
          WHEN 'WATCH' THEN 'Expires in ' || days_left || ' day(s): pick this batch first when dispensing'
          WHEN 'OK' THEN CASE WHEN days_left < 0 THEN 'Expired but no stock left' ELSE '' END
        END AS action
      FROM (
        SELECT
          *,
          expiry_date - current_date AS days_left,
          round(coalesce(quantity, 0) * coalesce(unit_cost, 0), 2) AS stock_value,
          list_filter([
            CASE WHEN expiry_date IS NULL THEN 'Expiry "' || expiry_raw || '" is not a date (use YYYY-MM-DD or MM/YYYY)' END,
            CASE WHEN quantity IS NULL THEN 'Quantity "' || coalesce(quantity_raw, '') || '" is not a whole number' END,
            CASE WHEN quantity < 0 THEN 'Negative quantity ' || quantity || ' (stock count or billing error)' END,
            CASE WHEN unit_cost IS NULL THEN 'Unit cost is missing or not a number' END,
            CASE WHEN times_listed > 1 THEN 'Batch listed ' || times_listed || ' times in this branch (merge the rows or fix the batch number)' END
          ], lambda x: x IS NOT NULL) AS data_errors,
          CASE
            WHEN len(data_errors) > 0 THEN 'DATA_ERROR'
            WHEN quantity > 0 AND expiry_date - current_date <= {{ inputs.critical_days }} THEN 'CRITICAL'
            WHEN quantity > 0 AND expiry_date - current_date <= {{ inputs.watch_days }} THEN 'WATCH'
            ELSE 'OK'
          END AS status
        FROM parsed
      )
      ORDER BY
        CASE status WHEN 'CRITICAL' THEN 0 WHEN 'DATA_ERROR' THEN 1 WHEN 'WATCH' THEN 2 ELSE 3 END,
        days_left NULLS FIRST,
        branch,
        sku;

      COPY report TO '{{ outputFiles["expiry_report.csv"] }}' (HEADER, DELIMITER ',');

      SELECT
        count(*) AS total_batches,
        count(*) FILTER (WHERE status = 'OK') AS ok_count,
        count(*) FILTER (WHERE status = 'WATCH') AS watch_count,
        count(*) FILTER (WHERE status = 'CRITICAL') AS critical_count,
        count(*) FILTER (WHERE status = 'DATA_ERROR') AS data_error_count,
        count(*) FILTER (WHERE status = 'CRITICAL' AND days_left < 0) AS expired_count,
        coalesce(sum(stock_value) FILTER (WHERE status = 'CRITICAL'), 0)::DOUBLE AS value_at_risk,
        coalesce(sum(stock_value) FILTER (WHERE status = 'WATCH'), 0)::DOUBLE AS value_on_watch,
        coalesce(string_agg(branch || ' / ' || product_name || ' batch ' || batch_no || ': ' || action, chr(10)) FILTER (WHERE status = 'CRITICAL'), '') AS critical_details,
        coalesce(string_agg(branch || ' / ' || product_name || ' batch ' || batch_no || ': ' || action, chr(10)) FILTER (WHERE status = 'DATA_ERROR'), '') AS data_error_details,
        coalesce(string_agg(branch || ' / ' || product_name || ' batch ' || batch_no || ': ' || action, chr(10)) FILTER (WHERE status = 'WATCH'), '') AS watch_details
      FROM report;

  - id: route_findings
    type: io.kestra.plugin.core.flow.If
    description: Branch on the result. Critical batches and data errors need someone
      to act this week, so they are logged as warnings and optionally sent to
      Slack. A clean run is logged so the execution history shows when the stock
      was last checked.
    condition: "{{ outputs.check_stock.outputs[0].row.critical_count +
      outputs.check_stock.outputs[0].row.data_error_count > 0 }}"
    then:
      - id: log_findings
        type: io.kestra.plugin.core.log.Log
        level: WARN
        description: Write every critical batch, data error and watch-list batch to the
          execution logs with the action to take.
        message: |
          Expiry check: {{ outputs.check_stock.outputs[0].row.critical_count }} critical ({{ outputs.check_stock.outputs[0].row.expired_count }} already expired), {{ outputs.check_stock.outputs[0].row.data_error_count }} data error(s), {{ outputs.check_stock.outputs[0].row.watch_count }} on the watch list, {{ outputs.check_stock.outputs[0].row.ok_count }} OK out of {{ outputs.check_stock.outputs[0].row.total_batches }} batches.
          Stock value at risk: Rs {{ outputs.check_stock.outputs[0].row.value_at_risk | numberFormat('#,##0.00') }}
          Critical:
          {{ outputs.check_stock.outputs[0].row.critical_details }}
          Data errors:
          {{ outputs.check_stock.outputs[0].row.data_error_details }}
          Watch list:
          {{ outputs.check_stock.outputs[0].row.watch_details }}

      - id: slack_enabled
        type: io.kestra.plugin.core.flow.If
        description: Only call Slack when the notify_slack input is true, so the flow
          runs out of the box without any secret.
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: alert_pharmacy_team
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Tell the pharmacy team which batches to pull, return or sell first,
              and which stock records to fix.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              :pill: Expiry check: {{ outputs.check_stock.outputs[0].row.critical_count }} critical batch(es), {{ outputs.check_stock.outputs[0].row.data_error_count }} data error(s), Rs {{ outputs.check_stock.outputs[0].row.value_at_risk | numberFormat('#,##0.00') }} of stock at risk.
              {{ outputs.check_stock.outputs[0].row.critical_details }}
              {{ outputs.check_stock.outputs[0].row.data_error_details }}
              Full report: execution {{ execution.id }} in flow {{ flow.namespace }}.{{ flow.id }}
    else:
      - id: log_clean
        type: io.kestra.plugin.core.log.Log
        description: Record a clean check, including how many batches are on the watch list.
        message: "Expiry check passed: {{
          outputs.check_stock.outputs[0].row.total_batches }} batches checked,
          nothing expired or due within {{ inputs.critical_days }} days, {{
          outputs.check_stock.outputs[0].row.watch_count }} batch(es) on the
          watch list."

errors:
  - id: log_check_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    description: Make a failed check loud. If the file is missing or malformed, no
      batch was checked, and a silent failure looks like a shelf with no expired
      medicine.
    message: "Expiry check FAILED in {{ flow.namespace }}.{{ flow.id }} (execution
      {{ execution.id }}). No batches were checked. Make sure stock_file is a
      CSV with the columns branch, sku, product_name, batch_no, expiry, quantity
      and unit_cost."

outputs:
  - id: expiry_report
    type: FILE
    description: Per-batch CSV report with the parsed expiry date, days left, stock
      value, status (CRITICAL, DATA_ERROR, WATCH or OK) and the action to take.
    value: "{{ outputs.check_stock.outputFiles['expiry_report.csv'] }}"
  - id: expiry_summary
    type: JSON
    description: 'Counts, value at risk and details, e.g. {"total_batches": 12,
      "critical_count": 3, "data_error_count": 4, "value_at_risk": 10040.0,
      ...}.'
    value: "{{ outputs.check_stock.outputs[0].row | toJson }}"

Expired medicine on the shelf is a patient safety problem, and a near-expiry batch that nobody noticed in time is money lost. Most suppliers only take back stock before it expires, so the earlier a batch is spotted, the more options you have. This blueprint checks the whole stock file on a schedule and tells the pharmacy team exactly which batches to pull, which to return or sell first, and which stock records are wrong. Every check runs locally in DuckDB: no external service, and no stock data leaving your Kestra instance.

How it works

sample_stock (io.kestra.plugin.core.storage.Write) writes a 12-batch stock file for two branches that contains one example of every problem. Its expiry dates are computed relative to today, so the demo always shows the same mix. It is used only when the stock_file input is empty.
check_stock (io.kestra.plugin.jdbc.duckdb.Queries) loads the CSV through inputFiles and, for every batch:
- reads the expiry as a full date (2027-03-31) or as the month and year printed on the strip (03/2027, read as the last day of that month);
- computes the days left and the stock value (quantity x unit_cost);
- marks the batch CRITICAL when it still has stock and is expired or expires within critical_days;
- marks it WATCH when it expires within watch_days;
- marks it DATA_ERROR when the expiry is not a date, the quantity is negative or not a whole number, the unit cost is missing, or the same batch is listed twice in one branch.
Expired batches with zero stock are reported as OK, because there is nothing left to pull. The full per-batch result is written to expiry_report.csv with outputFiles, and the final SELECT returns the summary row through fetchType: FETCH_ONE, available as outputs.check_stock.outputs[0].row.
route_findings (io.kestra.plugin.core.flow.If) logs every critical batch, data error and watch-list batch when anything needs action, and a nested If posts the findings to Slack when notify_slack is true. A clean check is logged as well.
The errors block logs a failed check loudly, because a check that silently does not run looks the same as a shelf with nothing expired.
The weekly_expiry_check Schedule trigger (Monday 08:00 IST, shipped disabled) runs the check before the week's purchase order goes out.

What you get

outputs.expiry_report: CSV with branch, sku, product_name, batch_no, expiry_raw, expiry_date, days_left, quantity, unit_cost, stock_value, status and action, sorted with the most urgent batches first.
outputs.expiry_summary: JSON with total_batches, ok_count, watch_count, critical_count, data_error_count, expired_count, value_at_risk, value_on_watch and the details for each group.
Plain-language actions such as "Expired 12 day(s) ago with 40 unit(s) still in stock: quarantine and remove from sale" or "Expires in 18 day(s) with 200 unit(s) in stock: return to the supplier or sell first now".
With the sample data: 3 critical batches (one already expired), 4 data errors (an unreadable expiry date, a negative quantity, and a batch listed twice), 2 batches on the watch list and 3 OK, including a strip-style MM/YYYY expiry and an expired batch with no stock left.

Who it's for

Pharmacy owners and chains who export stock from their billing or inventory software and want a weekly expiry review without doing it by hand.
Hospital and clinic stores that need to keep expired medicine off the shelf and return near-expiry stock in time.
Data teams that load inventory data into a warehouse and want a quality gate on batch and expiry fields.

Why orchestrate this with Kestra

A spreadsheet filter works once, when someone remembers to do it. Kestra runs the same check every week, keeps the report from every run in the execution history, branches on the result, and alerts the people who act on it. Because the input is just a file, the same flow can be triggered when a new stock export lands in S3, SFTP or Google Drive, or run once per branch with a ForEach.

Prerequisites

Nothing for the demo: the sample stock file is built in.
For your own data: a CSV export with the columns branch, sku, product_name, batch_no, expiry, quantity and unit_cost.

Inputs

stock_file (FILE, optional): your stock CSV. When empty, the built-in sample is used.
critical_days (INT, default 30): batches expiring within this many days are critical.
watch_days (INT, default 90): batches expiring within this many days go on the watch list.
notify_slack (BOOL, default false): post the findings to Slack when at least one batch is critical or has a data error.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook URL. Only needed when notify_slack is true.

Quick start

Save the flow and run it with the default inputs.
Open the logs of log_findings and download expiry_report from the Outputs tab to see the result for the sample stock.
Run it again and upload your own stock export as stock_file.
Set critical_days to match your supplier's return window.
Optional: add the SLACK_WEBHOOK_URL secret and set notify_slack to true.
Replace the input with your real source and set disabled: false on weekly_expiry_check.

Common pitfalls

Expiry dates in DD/MM/YYYY or DD-MM-YYYY format are reported as data errors rather than guessed, because 03/04/2027 could mean two different dates. Convert them to YYYY-MM-DD in the export.
Spreadsheets sometimes turn 03/2027 into a full date or a number. Keep the expiry column as text when exporting.
Days left are counted from today in the Kestra server's time zone. A batch that expires today has 0 days left and is critical.
A batch listed twice is usually two purchase entries for the same batch. Merge them in your billing software so the quantity is counted once.

How to extend

Read the stock straight from your inventory database (Postgres, MySQL, SQL Server) instead of a file, and write report back to a table.
Trigger the check when a new export lands, with the S3, SFTP or Google Drive triggers.
Email the report to each branch manager, or open a return request with your distributor for every critical batch.
Add a reorder check on the same file: flag fast-moving SKUs whose total stock is below a minimum level.

Catch Expired and Near-Expiry Medicine Batches Before They Become a Write-Off

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Inputs

Secrets

Quick start

Common pitfalls

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Catch Expired and Near-Expiry Medicine Batches Before They Become a Write-Off

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Inputs

Secrets

Quick start

Common pitfalls

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Business Blueprints

Send a daily Odoo sales digest to Slack

Create Odoo CRM leads from an inbound webhook

Okta Inactive User Deprovisioning Guard

Export Odoo invoices to an S3 warehouse staging bucket

Odoo Scheduled Action Overdue Sentinel

Pennylane Trial Balance Integrity Gate

More Data Blueprints

Real-Time Kafka Record Processing with Per-Record Executions

Kafka Consumer Lag Sentinel

Block a breaking Avro schema before a Kafka producer deploys

Kafka Schema Registry Compatibility and Mode Drift Guard

Kafka Producer Compression Ratio Sentinel

Kafka Schema Registry Orphaned Subject Cleanup Guard

New to Kestra?