Build a Quality Alert Pipeline

Build a Docker pipeline that alerts Discord about PostgreSQL quality spikes.

Introduction

30 Second Summary

A production defect becomes expensive when nobody notices that one station keeps failing. Repeated warnings create a different problem because operators start tuning them out.

In this project, you will build a local manufacturing quality monitor inside a Docker Compose stack. It turns recent PostgreSQL defect events into one deduplicated Discord alert when Station 5 crosses the quality threshold.

What You'll Build

Your finished monitor turns a Station 5 quality spike into one focused Discord alert while repeated checks stay silent about the unchanged anomaly.

By the end of this project, you'll have:

  • A one-command quality monitor with Station 5 visibly crossing the three-defect threshold in its logs.
  • A Discord quality alert that names the station, defect count, and 60-minute lookback window.
  • Persistent duplicate suppression that remembers an unchanged anomaly after the inspector container restarts.
  • Secret Mission: Track incident state and send a Discord recovery message when a station falls below the threshold.

Are there any prerequisites?

Your Linux device needs Docker Engine, Docker Compose, and Visual Studio Code ready to use. You should be comfortable with Python, pandas, SQL, and Docker plus have permission to create a webhook in a Discord server.

Before We Start

This checkpoint defines a local manufacturing quality monitor that flags stations with at least three High-severity defects in a rolling 60-minute UTC window. It suppresses duplicate Discord alerts so operators are not spammed.

Prepare the Linux Workspace

The monitor will soon connect production data to a Discord alert. A scattered workspace makes that pipeline harder to build.

A dedicated Visual Studio Code workspace keeps every project file in one place. Its integrated terminal also runs commands from the correct folder.

The local .env file eventually holds a sensitive webhook URL. Your .gitignore will exclude it from version control. The .env.example file will show the required configuration with a harmless placeholder.

In this step, get ready to:
  • Open the manufacturing-quality-alert workspace with its integrated terminal.
  • Protect the webhook configuration with .gitignore and .env.example.
  • Confirm that Docker Compose responds from the project terminal.
Open the project workspace

Visual Studio Code treats an open folder as a workspace. The integrated terminal starts inside that folder by default.

You may see a trust prompt

Visual Studio Code may ask whether you trust the folder after opening it. You created this empty folder yourself, so you can trust its contents for this project.

  • Open your Linux application search.
  • Type Visual Studio Code into the search field.
  • Select Visual Studio Code from the results.
  • Click File in the menu bar.
  • Select Open Folder....
  • Select New Folder in the folder dialog.
  • Enter manufacturing-quality-alert as the folder name.
  • Select Select Folder.
  • Confirm that you trust the empty folder if the trust prompt appears.

Good start. You can now see manufacturing-quality-alert at the top of the Explorer sidebar.

  • Click Terminal in the menu bar.
  • Select New Terminal.

The terminal panel opens at the bottom of Visual Studio Code. That setup win keeps every project command rooted in manufacturing-quality-alert.

Protect the webhook configuration

The real webhook URL belongs only in .env. An ignore file blocks that local secret from entering version control.

  • Select New File... in the Explorer sidebar.
  • Enter .gitignore as the file name.
  • Exclude the local secret file plus generated Python files by pasting these lines into .gitignore:
.env
__pycache__/
*.pyc

What does this file protect?

  • .env excludes the local file that will contain the webhook URL.
  • __pycache__/ excludes generated Python cache directories.
  • *.pyc excludes compiled Python files.
  • Press Ctrl+S on Linux to save .gitignore.
  • Confirm that the editor shows all three ignore patterns on separate lines.

Is the ignore file missing?

Check that the file name starts with a dot. Confirm that Visual Studio Code saved it inside manufacturing-quality-alert.

Help me check my ignore file.

A safe configuration template tells collaborators which value the project needs. It contains a placeholder instead of a working credential.

  • Select New File... in the Explorer sidebar.
  • Enter .env.example as the file name.
  • Document the required webhook setting by pasting this line into .env.example:
WEBHOOK_URL=your-discord-webhook-url-here

What does this template show?

WEBHOOK_URL names the environment variable that the inspector will read later. The placeholder documents the expected value without exposing a real Discord webhook URL.

  • Press Ctrl+S on Linux to save .env.example.
  • Confirm that .env.example appears beside .gitignore in the Explorer sidebar.

Does the template contain a real URL?

Replace any working webhook URL with your-discord-webhook-url-here. A real credential belongs only in the ignored .env file created later.

Help me protect my webhook configuration.

✔️ Awesome, I've got everything!

Both workspace files now match the safe project baseline.

ⓧ I'd like to double check the full code

.env
__pycache__/
*.pyc

This complete .gitignore keeps the local secret file plus generated Python cache files out of version control.

WEBHOOK_URL=your-discord-webhook-url-here

This complete template records the required setting without storing a real webhook URL.

Verify the workspace

The future stack depends on the Docker Compose command. This final check confirms that your existing installation responds inside the project workspace.

Before you run the check, what response do you expect from Docker Compose?

  • Confirm that Docker Compose responds by running this command:
docker compose version

What does this command check?

This asks Docker Compose to print its installed version information. A successful response proves that the Compose command is available from the integrated terminal.

You'll see Docker Compose version information in the terminal. That response confirms the container orchestration tool is ready for the next step.

Docker Compose not responding?

Confirm that the Docker installation available before this project includes Docker Compose. Restart Visual Studio Code after correcting the installation so its terminal receives the updated environment.

Help me diagnose my Docker Compose check.

  • Select .gitignore in the Explorer sidebar.
  • Confirm that .env appears as the first ignore pattern.
  • Select .env.example in the Explorer sidebar.
  • Confirm that the file contains only the safe WEBHOOK_URL placeholder.

Your workspace is ready. Docker Compose responds successfully while the future Discord credential has a protected place to live.

Next, you'll turn this empty workspace into a running database and a one-shot inspector that reveals the Station 5 quality spike.

Detect the Station 5 Quality Spike

Your protected workspace is ready in Visual Studio Code. Now you can turn it into a local quality monitor with Docker Compose.

The detector first needs to prove that its PostgreSQL query reads the correct UTC window. A pandas summary then makes the Station 5 spike visible before alerts add another moving part.

In this step, get ready to:
  • Pin the inspector image and Python dependencies.
  • Seed recent manufacturing defects and build the one-shot detector.
  • Validate the Compose configuration and reveal the Station 5 spike.
Pin the inspector runtime

A pinned dependency file gives every inspector build the same package versions. The Docker Image then supplies the matching Python runtime.

  • In Visual Studio Code's file sidebar, select the new-file icon.
  • Enter requirements.txt as the file name.
  • Press Enter.

You'll see requirements.txt listed beside the files from the previous step.

  • Add the pinned packages to requirements.txt by pasting this code:
pandas==3.0.6
SQLAlchemy==2.1.4
psycopg[binary]==3.3.6
schedule==1.2.2

What does this file do?

  • pandas transforms the database rows into a grouped defect summary.
  • SQLAlchemy creates the database engine and executes the parameterized query.
  • Psycopg connects SQLAlchemy to PostgreSQL.
  • schedule supplies the recurring job loop used later in the project.
  • Save requirements.txt.

Good progress. The editor now shows four exact package pins for the inspector image.

Package lines look different?

  • Remove any spaces surrounding each == version separator.
  • Check that psycopg[binary] retains its square brackets.

Ask for help with the dependency file if the four lines still differ: Help me compare my requirements.txt file with the required package pins.

The image recipe installs those dependencies before copying the detector. This ordering lets Docker reuse the dependency layer when only main.py changes.

  • In Visual Studio Code's file sidebar, select the new-file icon.
  • Enter Dockerfile as the file name.
  • Press Enter.

You'll see an empty Dockerfile ready for the inspector image recipe.

  • Define the inspector image in Dockerfile by pasting this code:
FROM python:3.12.15-slim-bookworm

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
CMD ["python", "main.py"]

How does the image come together?

  • The base image supplies Python 3.12.15 in a compact Debian environment.
  • The dependency layer installs every pin from requirements.txt.
  • The final command starts main.py when the inspector container runs.
  • Save Dockerfile.

The inspector now has a reproducible image recipe. You'll see its seven instructions in the editor.

Dockerfile formatting look wrong?

  • Place every Dockerfile instruction on its own line.
  • Keep the final command's double quotes and square brackets exactly as shown.

Use this if you want help checking the recipe: Help me find the formatting issue in my Dockerfile.

Build the one-shot detector

The database seed uses timestamps relative to its first startup. That keeps the important Station 5 defects inside the rolling 60-minute window.

  • In Visual Studio Code's file sidebar, select the new-file icon.
  • Enter init.sql as the file name.
  • Press Enter.

You'll see init.sql in the workspace file list.

  • Define and seed the defect table in init.sql by pasting this code:
CREATE TABLE defect_logs (
    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    occurred_at TIMESTAMPTZ NOT NULL,
    production_station TEXT NOT NULL,
    severity TEXT NOT NULL,
    defect_type TEXT NOT NULL
);

INSERT INTO defect_logs (
    occurred_at,
    production_station,
    severity,
    defect_type
) VALUES
    (CURRENT_TIMESTAMP - INTERVAL '10 minutes', 'Station 5', 'High', 'Surface crack'),
    (CURRENT_TIMESTAMP - INTERVAL '20 minutes', 'Station 5', 'High', 'Seal failure'),
    (CURRENT_TIMESTAMP - INTERVAL '35 minutes', 'Station 5', 'High', 'Dimension drift'),
    (CURRENT_TIMESTAMP - INTERVAL '5 minutes', 'Station 5', 'Medium', 'Label skew'),
    (CURRENT_TIMESTAMP - INTERVAL '15 minutes', 'Station 2', 'High', 'Paint chip'),
    (CURRENT_TIMESTAMP - INTERVAL '40 minutes', 'Station 2', 'High', 'Loose fitting'),
    (CURRENT_TIMESTAMP - INTERVAL '90 minutes', 'Station 8', 'High', 'Old sensor alarm');

How does the seed expose the spike?

  • The identity column assigns event IDs in insertion order.
  • Three recent Station 5 rows carry High severity.
  • The recent Station 5 Medium row tests the severity filter.
  • The Station 8 row sits outside the 60-minute window.
  • Save init.sql.

The seed file now contains one table definition and seven timestamped defect events.

Seed rows not aligned?

  • Keep a comma after every seed row except the final Station 8 row.
  • Keep the closing semicolon after the final row.

Use this if the SQL structure is hard to compare: Help me check the syntax and row separators in init.sql.

The inspector has two data responsibilities. It loads only recent events from PostgreSQL. It then groups the High-severity rows by production station.

  • In Visual Studio Code's file sidebar, select the new-file icon.
  • Enter main.py as the file name.
  • Press Enter.

You'll see an empty main.py ready for the detector.

  • Add the inspector configuration to main.py by pasting this code:
import logging
import os
from datetime import datetime, timedelta, timezone

import pandas as pd
from sqlalchemy import create_engine, text

DATABASE_URL = os.environ["DATABASE_URL"]
DEFECT_THRESHOLD = int(os.getenv("DEFECT_THRESHOLD", "3"))
LOOKBACK_MINUTES = int(os.getenv("LOOKBACK_MINUTES", "60"))

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("quality-inspector")
engine = create_engine(DATABASE_URL, pool_pre_ping=True)

What does this setup control?

  • The environment supplies the database address and detector thresholds.
  • The logger gives the container a visible result for each one-shot run.
  • Connection pre-ping checks pooled connections before the inspector uses them.
  • Save main.py.

The top of main.py now shows the imports, three environment settings, logger, and database engine.

Top-level setup misaligned?

  • Keep every import and configuration line flush with the left edge.
  • Check that each environment variable name retains its uppercase spelling.

Use this if the setup still looks different: Help me compare the top-level configuration in main.py.

The rolling query calculates its boundary in UTC. PostgreSQL applies that boundary before pandas receives any rows.

  • Add the recent-defect loader at the end of main.py by pasting this code:
def load_recent_defects():
    window_start = datetime.now(timezone.utc) - timedelta(minutes=LOOKBACK_MINUTES)
    statement = text("""
        SELECT id, occurred_at, production_station, severity, defect_type
        FROM defect_logs
        WHERE occurred_at >= :window_start
        ORDER BY occurred_at
    """)
    with engine.connect() as connection:
        defects = pd.read_sql_query(statement, connection, params={"window_start": window_start})
    if not defects.empty:
        defects["occurred_at"] = pd.to_datetime(defects["occurred_at"], utc=True)
    return defects

How does the rolling query work?

  • The window starts 60 minutes before the current UTC time.
  • The parameterized query limits the database read to recent events.
  • The timestamp conversion makes the returned values explicitly UTC-aware.
  • Save main.py.

You'll see load_recent_defects() beneath the engine setup with its complete query and UTC conversion.

Query block indentation shifted?

  • Keep the query text inside all three opening and closing double quotes.
  • Align the final return defects with the first line inside the function.

Use this if Python reports an indentation issue during the build: Help me fix the indentation in load_recent_defects().

The summary keeps station name, defect count, and latest event ID together. Empty checks preserve the same columns when no matching rows exist.

  • Add the High-severity summary function at the end of main.py by pasting this code:
def summarize_high_severity(defects):
    columns = ["production_station", "defect_count", "latest_event_id"]
    if defects.empty:
        return pd.DataFrame(columns=columns)
    high_severity = defects.loc[defects["severity"].eq("High")]
    if high_severity.empty:
        return pd.DataFrame(columns=columns)
    return high_severity.groupby("production_station", as_index=False).agg(
        defect_count=("id", "size"),
        latest_event_id=("id", "max"),
    )

How does pandas find the spike?

  • The equality filter keeps only rows whose severity is High.
  • The grouping operation creates one result row per production station.
  • The named aggregations count defect IDs and retain the largest event ID.
  • Save main.py.

You'll see summarize_high_severity() return the same three columns for populated and empty results.

Aggregation block look uneven?

  • Keep both named aggregations inside the final agg() call.
  • Keep the closing parenthesis aligned with the return statement.

Use this if the grouped summary raises a syntax error: Help me check summarize_high_severity().

The entry point applies the threshold after the grouped summary is ready. Its log converts each triggered station into a readable record.

  • Complete the one-shot detector at the end of main.py by pasting this code:
def main():
    defects = load_recent_defects()
    summary = summarize_high_severity(defects)
    triggered = summary.loc[summary["defect_count"] >= DEFECT_THRESHOLD]
    logger.info("%s", triggered.to_dict(orient="records"))
    engine.dispose()


if __name__ == "__main__":
    main()

What does the entry point prove?

  • The threshold filter retains stations with at least three High-severity defects.
  • The log exposes each triggered station as a record.
  • The final disposal releases the database connection pool after the one-shot check.
  • Save main.py.

The one-shot detector is complete. Its entry point now loads, summarizes, filters, logs, and closes the database engine.

Entry point not at the bottom?

  • Place the if __name__ guard after every function definition.
  • Keep main() indented beneath the guard.

Use this if the detector does not call its entry point: Help me check the bottom of main.py.

Validate and run the stack

The Compose file joins the database and inspector into one local stack. Its health dependency holds the inspector until PostgreSQL accepts connections.

  • In Visual Studio Code's file sidebar, select the new-file icon.
  • Enter compose.yaml as the file name.
  • Press Enter.

You'll see an empty compose.yaml ready for both services.

  • Define the two-service stack in compose.yaml by pasting this code:
services:
  db:
    image: postgres:18.6-alpine3.24
    environment:
      POSTGRES_USER: quality_app
      POSTGRES_PASSWORD: quality_dev_password
      POSTGRES_DB: quality
    volumes:
      - pg_data:/var/lib/postgresql
      - ./init.sql:/docker-entrypoint-initdb.d/001-init.sql:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5
      start_period: 10s

  inspector:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql+psycopg://quality_app:quality_dev_password@db:5432/quality
      DEFECT_THRESHOLD: "3"
      LOOKBACK_MINUTES: "60"

volumes:
  pg_data:

How does Compose coordinate the services?

  • The database service uses the pinned PostgreSQL image.
  • The named volume preserves database files across container restarts.
  • The initialization mount supplies init.sql during the database's first startup.
  • The health condition delays the inspector until the database reports readiness.
  • Save compose.yaml.

Your workspace now contains every file required for the one-shot quality detector.

YAML indentation look uneven?

  • Use spaces for every indentation level in compose.yaml.
  • Align db and inspector beneath services.
  • Align pg_data beneath the final volumes key.

Use this if the structure still looks wrong: Help me check the indentation in compose.yaml.

✔️ Awesome, I've got everything!

Your seven project files now match the complete one-shot detector.

ⓧ I'd like to double check the full code

  • Compare each saved file with the complete references below.
WEBHOOK_URL=your-discord-webhook-url-here
.env
__pycache__/
*.pyc
FROM python:3.12.15-slim-bookworm

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
CMD ["python", "main.py"]
services:
  db:
    image: postgres:18.6-alpine3.24
    environment:
      POSTGRES_USER: quality_app
      POSTGRES_PASSWORD: quality_dev_password
      POSTGRES_DB: quality
    volumes:
      - pg_data:/var/lib/postgresql
      - ./init.sql:/docker-entrypoint-initdb.d/001-init.sql:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5
      start_period: 10s

  inspector:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql+psycopg://quality_app:quality_dev_password@db:5432/quality
      DEFECT_THRESHOLD: "3"
      LOOKBACK_MINUTES: "60"

volumes:
  pg_data:
CREATE TABLE defect_logs (
    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    occurred_at TIMESTAMPTZ NOT NULL,
    production_station TEXT NOT NULL,
    severity TEXT NOT NULL,
    defect_type TEXT NOT NULL
);

INSERT INTO defect_logs (
    occurred_at,
    production_station,
    severity,
    defect_type
) VALUES
    (CURRENT_TIMESTAMP - INTERVAL '10 minutes', 'Station 5', 'High', 'Surface crack'),
    (CURRENT_TIMESTAMP - INTERVAL '20 minutes', 'Station 5', 'High', 'Seal failure'),
    (CURRENT_TIMESTAMP - INTERVAL '35 minutes', 'Station 5', 'High', 'Dimension drift'),
    (CURRENT_TIMESTAMP - INTERVAL '5 minutes', 'Station 5', 'Medium', 'Label skew'),
    (CURRENT_TIMESTAMP - INTERVAL '15 minutes', 'Station 2', 'High', 'Paint chip'),
    (CURRENT_TIMESTAMP - INTERVAL '40 minutes', 'Station 2', 'High', 'Loose fitting'),
    (CURRENT_TIMESTAMP - INTERVAL '90 minutes', 'Station 8', 'High', 'Old sensor alarm');
import logging
import os
from datetime import datetime, timedelta, timezone

import pandas as pd
from sqlalchemy import create_engine, text

DATABASE_URL = os.environ["DATABASE_URL"]
DEFECT_THRESHOLD = int(os.getenv("DEFECT_THRESHOLD", "3"))
LOOKBACK_MINUTES = int(os.getenv("LOOKBACK_MINUTES", "60"))

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("quality-inspector")
engine = create_engine(DATABASE_URL, pool_pre_ping=True)


def load_recent_defects():
    window_start = datetime.now(timezone.utc) - timedelta(minutes=LOOKBACK_MINUTES)
    statement = text("""
        SELECT id, occurred_at, production_station, severity, defect_type
        FROM defect_logs
        WHERE occurred_at >= :window_start
        ORDER BY occurred_at
    """)
    with engine.connect() as connection:
        defects = pd.read_sql_query(statement, connection, params={"window_start": window_start})
    if not defects.empty:
        defects["occurred_at"] = pd.to_datetime(defects["occurred_at"], utc=True)
    return defects


def summarize_high_severity(defects):
    columns = ["production_station", "defect_count", "latest_event_id"]
    if defects.empty:
        return pd.DataFrame(columns=columns)
    high_severity = defects.loc[defects["severity"].eq("High")]
    if high_severity.empty:
        return pd.DataFrame(columns=columns)
    return high_severity.groupby("production_station", as_index=False).agg(
        defect_count=("id", "size"),
        latest_event_id=("id", "max"),
    )


def main():
    defects = load_recent_defects()
    summary = summarize_high_severity(defects)
    triggered = summary.loc[summary["defect_count"] >= DEFECT_THRESHOLD]
    logger.info("%s", triggered.to_dict(orient="records"))
    engine.dispose()


if __name__ == "__main__":
    main()
pandas==3.0.6
SQLAlchemy==2.1.4
psycopg[binary]==3.3.6
schedule==1.2.2

Compose can now check the complete service model before it creates anything.

  • Validate compose.yaml by running this command in the integrated terminal:
docker compose config

What does this command check?

Compose parses the YAML and resolves the service configuration. A successful result prints the normalized configuration without a validation error.

You'll see the database service, inspector service, and named volume in the normalized output.

Configuration validation failed?

  • Read the reported line number in the terminal.
  • Compare that line's indentation with the complete compose.yaml reference.
  • Run the validation command again after saving the correction.

Use this if the validation message is unclear: Help me interpret my Docker Compose configuration error.

The first build can take a few minutes while Docker downloads the images and installs the pinned Python packages. Quiet stretches during that first build are expected.

  • Build and start the stack in detached mode by running this command:
docker compose up --build -d

What happens during startup?

  • Docker builds the inspector image from Dockerfile.
  • PostgreSQL creates the quality database inside pg_data.
  • The first database startup executes init.sql.
  • The inspector starts after the database health check succeeds.

You'll see the services build, create, and start. The inspector then performs its single quality check.

Stack did not start?

  • Check the build output for a misspelled package pin in requirements.txt.
  • Check the build output for a missing main.py or requirements.txt file.
  • Run the Compose configuration check again after saving your correction.

Use this if the build output does not reveal the cause: Help me troubleshoot why my Compose stack did not start.

Before you inspect the logs, which station do you expect to cross the three-defect threshold?

  • Reveal the inspector's one-shot result by running this command:
docker compose logs inspector

What proves the detector works?

The log record contains the station name, grouped defect count, and latest event ID. These values prove that the rolling query and pandas aggregation agree.

You'll see Station 5 with defect_count equal to 3 and latest_event_id equal to 3.

That's the detector proven: Station 5 is the only station that reaches the threshold inside the latest 60-minute window.

Station 5 missing from the log?

  • Check that the three recent Station 5 rows retain High severity in init.sql.
  • Check that LOOKBACK_MINUTES remains 60 in compose.yaml.
  • Check that DEFECT_THRESHOLD remains 3 in compose.yaml.

Use this if the inspector logs show another result: Help me trace why Station 5 is missing from my inspector output.

Your detector can now turn seeded production events into a visible quality signal. Next, you'll deliver that signal to a Discord channel and expose the duplicate-alert problem.

Send the Alert and Expose the Spam Problem

Your one-shot inspector already turns recent PostgreSQL events into a grouped pandas result. However, that result currently stays inside the container logs.

A quality threshold becomes operational when it reaches the person responsible for containment. You will connect the detector to Discord Incoming Webhooks. Then you will test how the sender behaves across repeated runs.

In this step, get ready to:
  • Create a Discord incoming webhook for your target channel.
  • Post a Station 5 quality alert using Python's standard-library HTTP client.
  • Recreate the inspector twice to expose its repeated-alert behavior.
Create the Discord webhook

An incoming webhook is a channel-specific URL that accepts an HTTP request and publishes its message. Discord handles this without a bot or a persistent connection.

  • Return to the Discord server that will receive your quality alerts.
  • Open Server Settings from the server menu.
  • Select Integrations.
  • Click Create Webhook.
  • Select the target text channel for manufacturing alerts.

Protect the webhook URL

This next copy deserves care because the webhook URL can post messages into your channel. Keep it inside the ignored .env file.

Do not paste the URL into screenshots or commit it to version control.

  • Click Copy Webhook URL.
  • Switch back to the manufacturing-quality-alert workspace in Visual Studio Code.
  • Create a file named .env inside the manufacturing-quality-alert workspace using the editor's file controls.
  • Add the webhook setting by copying this line into .env:
WEBHOOK_URL=your-discord-webhook-url-here

What does this setting do?

The WEBHOOK_URL name gives Docker Compose a value to inject into the inspector container. The placeholder keeps the expected format visible without exposing a real credential.

  • Replace your-discord-webhook-url-here with the webhook URL you copied.
  • Save .env.

Your webhook credential is now stored locally. The existing .gitignore rule keeps .env outside version control.

Build the webhook sender

The sender needs three pieces. It needs the webhook URL from the environment. It also needs a JSON payload plus an HTTP request that reports delivery failures.

  • In main.py, replace the import and configuration section at the top with this code:
import json
import logging
import os
from datetime import datetime, timedelta, timezone
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

import pandas as pd
from sqlalchemy import create_engine, text

DATABASE_URL = os.environ["DATABASE_URL"]
WEBHOOK_URL = os.environ["WEBHOOK_URL"]
DEFECT_THRESHOLD = int(os.getenv("DEFECT_THRESHOLD", "3"))
LOOKBACK_MINUTES = int(os.getenv("LOOKBACK_MINUTES", "60"))

What does this code add?

  • The json module converts the alert into the request body Discord expects.
  • The Request and urlopen imports build and send the request.
  • The HTTPError and URLError imports let the inspector log rejected requests or connection failures.
  • The WEBHOOK_URL assignment reads the credential supplied by the container environment.
  • In compose.yaml, replace the current inspector service block with this configuration:
  inspector:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql+psycopg://quality_app:quality_dev_password@db:5432/quality
      WEBHOOK_URL: ${WEBHOOK_URL:?Set WEBHOOK_URL in .env}
      DEFECT_THRESHOLD: "3"
      LOOKBACK_MINUTES: "60"

How does the webhook reach the container?

Docker Compose reads WEBHOOK_URL from the local .env file. The interpolation rule stops startup when the value is missing or empty.

The inspector receives the URL at runtime. The credential stays out of compose.yaml.

  • Save main.py.
  • Save compose.yaml.
  • Rebuild the inspector and print its logs by running these commands:
docker compose up --build --force-recreate -d inspector
docker compose logs inspector

What do these commands check?

The first command rebuilds the image and recreates the one-shot inspector. The second command prints the inspector output after it exits.

This run confirms that the new imports and environment setting load successfully before the sender changes the alert flow.

You should still see Station 5 with a defect count of 3 and a latest event ID of 3 in the logs.

Inspector failing during startup?

Confirm that .env is inside the manufacturing-quality-alert workspace. Check that the file contains a non-empty WEBHOOK_URL value.

Check the indentation under the environment key in compose.yaml.

Ask for help with the environment setup.

✔️ Awesome, I've got everything!

Your Compose configuration now injects the webhook URL into the inspector without storing the credential in the YAML file.

ⓧ I'd like to double check the full code

services:
  db:
    image: postgres:18.6-alpine3.24
    environment:
      POSTGRES_USER: quality_app
      POSTGRES_PASSWORD: quality_dev_password
      POSTGRES_DB: quality
    volumes:
      - pg_data:/var/lib/postgresql
      - ./init.sql:/docker-entrypoint-initdb.d/001-init.sql:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5
      start_period: 10s
  inspector:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql+psycopg://quality_app:quality_dev_password@db:5432/quality
      WEBHOOK_URL: ${WEBHOOK_URL:?Set WEBHOOK_URL in .env}
      DEFECT_THRESHOLD: "3"
      LOOKBACK_MINUTES: "60"
volumes:
  pg_data:

What should you compare?

Check the service indentation and the four inspector environment values. Confirm that the webhook line uses the local environment value instead of a real URL.

  • In main.py, add this function above the existing main() function:
def send_webhook(message):
    payload = json.dumps({"content": message}).encode("utf-8")
    request = Request(
        WEBHOOK_URL,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=10) as response:
            logger.info("Webhook delivered with HTTP %s", response.status)
            return 200 <= response.status < 300
    except HTTPError as error:
        logger.error("Webhook HTTP error: %s", error.code)
    except URLError as error:
        logger.error("Webhook connection error: %s", error.reason)
    return False

How does the sender work?

  • The payload wraps the alert inside Discord's content field before encoding it as UTF-8 bytes.
  • The request identifies the body as application/json and sends it with the POST method.
  • The 10 second timeout prevents the one-shot inspector from waiting forever on a stalled connection.
  • A successful response returns True. HTTP or connection failures are logged before the function returns False.

Why use Python's standard library?

This sender only needs one JSON POST request. Python's standard library covers that request without adding another package to requirements.txt.

  • Save main.py.
  • Confirm the sender function loads without breaking the detector by running these commands:
docker compose up --build --force-recreate -d inspector
docker compose logs inspector

What does this run prove?

The rebuilt image imports the HTTP modules and loads the new function successfully. The existing Station 5 result confirms that the detector still reaches the end of its one-shot run.

You should see the same Station 5 grouped result in the inspector logs.

Seeing a Python error?

Check that send_webhook() starts at the left edge of main.py. Confirm that every line inside the function uses four spaces of indentation.

Ask for help with the sender function.

  • In main.py, replace the existing main() function and its surrounding execution block with this alert flow:
def inspect_quality():
    defects = load_recent_defects()
    summary = summarize_high_severity(defects)
    triggered = summary.loc[summary["defect_count"] >= DEFECT_THRESHOLD]
    if triggered.empty:
        logger.info("No station currently meets the alert threshold.")
        return
    for alert in triggered.to_dict(orient="records"):
        station = alert["production_station"]
        defect_count = int(alert["defect_count"])
        message = (
            f"QUALITY ALERT: {station} recorded {defect_count} "
            f"High-severity defects in the last {LOOKBACK_MINUTES} minutes."
        )
        send_webhook(message)


def main():
    inspect_quality()
    engine.dispose()


if __name__ == "__main__":
    main()

How does the detector trigger the alert?

  • The threshold filter keeps stations whose grouped defect count is at least DEFECT_THRESHOLD.
  • Each triggered row supplies the station name and defect count for the alert text.
  • The message includes the LOOKBACK_MINUTES value so the recipient knows which time window produced the count.
  • The one-shot main() function runs the inspection before disposing of the database engine.
  • Save main.py.

✔️ Awesome, I've got everything!

Your inspector now turns each triggered station row into a Discord quality alert.

ⓧ I'd like to double check the full code

import json
import logging
import os
from datetime import datetime, timedelta, timezone
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

import pandas as pd
from sqlalchemy import create_engine, text

DATABASE_URL = os.environ["DATABASE_URL"]
WEBHOOK_URL = os.environ["WEBHOOK_URL"]
DEFECT_THRESHOLD = int(os.getenv("DEFECT_THRESHOLD", "3"))
LOOKBACK_MINUTES = int(os.getenv("LOOKBACK_MINUTES", "60"))

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("quality-inspector")
engine = create_engine(DATABASE_URL, pool_pre_ping=True)


def load_recent_defects():
    window_start = datetime.now(timezone.utc) - timedelta(minutes=LOOKBACK_MINUTES)
    statement = text(
        """
        SELECT id, occurred_at, production_station, severity, defect_type
        FROM defect_logs
        WHERE occurred_at >= :window_start
        ORDER BY occurred_at
        """
    )
    with engine.connect() as connection:
        defects = pd.read_sql_query(statement, connection, params={"window_start": window_start})
    if not defects.empty:
        defects["occurred_at"] = pd.to_datetime(defects["occurred_at"], utc=True)
    return defects


def summarize_high_severity(defects):
    columns = ["production_station", "defect_count", "latest_event_id"]
    if defects.empty:
        return pd.DataFrame(columns=columns)
    high_severity = defects.loc[defects["severity"].eq("High")]
    if high_severity.empty:
        return pd.DataFrame(columns=columns)
    return high_severity.groupby("production_station", as_index=False).agg(
        defect_count=("id", "size"),
        latest_event_id=("id", "max"),
    )


def send_webhook(message):
    payload = json.dumps({"content": message}).encode("utf-8")
    request = Request(
        WEBHOOK_URL,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=10) as response:
            logger.info("Webhook delivered with HTTP %s", response.status)
            return 200 <= response.status < 300
    except HTTPError as error:
        logger.error("Webhook HTTP error: %s", error.code)
    except URLError as error:
        logger.error("Webhook connection error: %s", error.reason)
    return False


def inspect_quality():
    defects = load_recent_defects()
    summary = summarize_high_severity(defects)
    triggered = summary.loc[summary["defect_count"] >= DEFECT_THRESHOLD]
    if triggered.empty:
        logger.info("No station currently meets the alert threshold.")
        return
    for alert in triggered.to_dict(orient="records"):
        station = alert["production_station"]
        defect_count = int(alert["defect_count"])
        message = (
            f"QUALITY ALERT: {station} recorded {defect_count} "
            f"High-severity defects in the last {LOOKBACK_MINUTES} minutes."
        )
        send_webhook(message)


def main():
    inspect_quality()
    engine.dispose()


if __name__ == "__main__":
    main()

What should you compare?

Check the imports and environment values first. Then compare the existing query helpers before checking send_webhook() and inspect_quality().

Recreate the inspector twice

The sender now has everything it needs to reach Discord. The first recreation tests the complete path from PostgreSQL data to a channel message.

  • Build and recreate the inspector by running this command:
docker compose up --build --force-recreate -d inspector

What does this command do?

The command rebuilds the inspector image before replacing its container. Detached mode lets the one-shot process run without occupying the terminal.

  • Wait a few seconds for the one-shot inspector to finish.
  • Return to the target Discord channel.

You should see QUALITY ALERT: Station 5 recorded 3 High-severity defects in the last 60 minutes. The database result has now reached a person who can act on it.

No Discord message?

Check the inspector logs for a webhook HTTP error or a connection error. Confirm that the webhook still targets the channel you selected.

If the URL was copied incorrectly, replace its value in .env without sharing the credential.

Ask for help diagnosing the missing alert.

Before you run the same inspector again, do you expect Discord to receive a new message from unchanged data? Your prediction is about to be tested.

  • Recreate the unchanged inspector by running the same command again:
docker compose up --build --force-recreate -d inspector

What does the second run repeat?

The recreated container queries the same recent defects and calculates the same Station 5 threshold result. It calls the webhook sender again because the one-shot process has no record of its previous delivery.

  • Wait a few seconds for the second one-shot run to finish.
  • Return to the same Discord channel.

You should now see two equivalent Station 5 quality alerts. The second message is the intended shortfall in this step.

The duplicate is the result

Each recreated container starts with fresh process memory. The sender sees a valid threshold result and posts it without knowing that an earlier container already sent the same anomaly.

Repeated alerts make genuine changes harder to notice. The pipeline now needs durable coordination across container recreations.

You have exposed the exact failure that persistent state needs to solve. Next, you will store anomaly keys in PostgreSQL so unchanged defect data produces one alert instead of repeated noise.

Persist Alert Deduplication

Your Discord alert now reaches the target channel. Recreating the inspector also proved that the same Station 5 anomaly can trigger another message.

A recreated Docker container starts with fresh process memory. PostgreSQL gives the inspector a durable ledger that survives those recreations.

In this step, get ready to:
  • Create a persistent table for delivered alerts.
  • Check anomaly keys with parameterized queries.
  • Record successful alerts and suppress unchanged duplicates.
Create the persistent alert table

The alert_events table stores one row for each delivered anomaly. Its primary key prevents the same anomaly key from becoming two database records.

  • Return to main.py in Visual Studio Code.
  • Find the line that creates engine with create_engine().
  • Add ensure_alert_table() below that line by pasting this code:
def ensure_alert_table():
    statement = text(
        """
        CREATE TABLE IF NOT EXISTS alert_events (
            anomaly_key TEXT PRIMARY KEY,
            production_station TEXT NOT NULL,
            defect_count INTEGER NOT NULL,
            sent_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    with engine.begin() as connection:
        connection.execute(statement)

What Does This Code Do?

  • The table keeps the anomaly key with the station name and defect count.
  • The sent_at column records when PostgreSQL accepted the row.
  • The IF NOT EXISTS clause makes the setup safe to run during every inspection.
  • The engine.begin() context commits the table creation when the block succeeds.
  • Save main.py.
  • Select View from the Visual Studio Code menu.
  • Select Problems from the menu.

You should see no Python syntax problems for ensure_alert_table().

Seeing a Syntax Problem?

Check that the triple quotes surround the complete SQL statement. Confirm that the closing parenthesis for text() appears before the with block.

Ask for help checking the table function.

Check and record anomaly keys

A durable table becomes useful when the inspector can ask whether a key already exists. SQLAlchemy binds the key as a query parameter before PostgreSQL executes the statement.

  • Find summarize_high_severity() in main.py.
  • Add alert_was_sent() below that function by pasting this code:
def alert_was_sent(anomaly_key):
    statement = text(
        """
        SELECT anomaly_key
        FROM alert_events
        WHERE anomaly_key = :anomaly_key
        """
    )
    with engine.connect() as connection:
        result = connection.execute(statement, {"anomaly_key": anomaly_key})
        return result.scalar_one_or_none() is not None

How Does the Lookup Work?

  • The query searches alert_events for one matching anomaly_key.
  • The :anomaly_key placeholder keeps the SQL separate from the supplied value.
  • The function returns True when PostgreSQL finds a matching row.
  • Save main.py.
  • Check the Problems panel again.

You should see no Python syntax problems for alert_was_sent().

Is the Lookup Flagged?

Check the indentation under the connection block. Confirm that the parameter dictionary uses anomaly_key as both its key and supplied value.

Ask for help checking the parameterized query.

A lookup blocks repeated delivery. A separate write records the key only after the webhook confirms success.

  • Find alert_was_sent() in main.py.
  • Add record_alert() below that function by pasting this code:
def record_alert(anomaly_key, station, defect_count):
    statement = text(
        """
        INSERT INTO alert_events (anomaly_key, production_station, defect_count)
        VALUES (:anomaly_key, :production_station, :defect_count)
        """
    )
    with engine.begin() as connection:
        connection.execute(
            statement,
            {
                "anomaly_key": anomaly_key,
                "production_station": station,
                "defect_count": defect_count,
            },
        )

Why Record After Delivery?

  • The insert stores the key with the station and count that produced the alert.
  • The parameter dictionary supplies each value without building SQL from string fragments.
  • The transaction commits the row when the insert succeeds.

Calling this function after a successful webhook prevents a failed delivery from being marked as sent.

  • Save main.py.
  • Check the Problems panel once more.

You should see no Python syntax problems for record_alert().

Is the Insert Flagged?

Confirm that every SQL placeholder has a matching dictionary key. Check that connection.execute() receives the statement before the dictionary.

Ask for help checking the insert function.

Suppress the unchanged Station 5 alert

Station 5 has three recent High-severity events. Its latest matching event has ID 3.

Combining the station with that latest event ID produces Station 5:3. The same data now produces the same persistent key after every inspector recreation.

  • Find the existing inspect_quality() function in main.py.
  • Replace the entire function with this version:
def inspect_quality():
    ensure_alert_table()
    defects = load_recent_defects()
    summary = summarize_high_severity(defects)
    triggered = summary.loc[summary["defect_count"] >= DEFECT_THRESHOLD]
    if triggered.empty:
        logger.info("No station currently meets the alert threshold.")
        return
    for alert in triggered.to_dict(orient="records"):
        station = alert["production_station"]
        defect_count = int(alert["defect_count"])
        latest_event_id = int(alert["latest_event_id"])
        anomaly_key = f"{station}:{latest_event_id}"
        if alert_was_sent(anomaly_key):
            logger.info("Duplicate alert suppressed for %s", station)
            continue
        message = (
            f"QUALITY ALERT: {station} recorded {defect_count} "
            f"High-severity defects in the last {LOOKBACK_MINUTES} minutes."
        )
        if send_webhook(message):
            record_alert(anomaly_key, station, defect_count)
            logger.info("Alert recorded for %s", station)

How Does Suppression Work?

  • The function creates alert_events before querying it.
  • The latest event ID makes each unchanged station result produce a stable key.
  • A stored key causes the loop to log the suppression message before continuing.
  • A new key reaches Discord before record_alert() stores it.
  • Save main.py.

✔️ Awesome, I've got everything!

Your saved main.py now creates the alert ledger before checking each Station 5 anomaly.

ⓧ I'd like to double check the full code

import json
import logging
import os
from datetime import datetime, timedelta, timezone
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

import pandas as pd
from sqlalchemy import create_engine, text

DATABASE_URL = os.environ["DATABASE_URL"]
WEBHOOK_URL = os.environ["WEBHOOK_URL"]
DEFECT_THRESHOLD = int(os.getenv("DEFECT_THRESHOLD", "3"))
LOOKBACK_MINUTES = int(os.getenv("LOOKBACK_MINUTES", "60"))

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("quality-inspector")
engine = create_engine(DATABASE_URL, pool_pre_ping=True)


def ensure_alert_table():
    statement = text(
        """
        CREATE TABLE IF NOT EXISTS alert_events (
            anomaly_key TEXT PRIMARY KEY,
            production_station TEXT NOT NULL,
            defect_count INTEGER NOT NULL,
            sent_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    with engine.begin() as connection:
        connection.execute(statement)


def load_recent_defects():
    window_start = datetime.now(timezone.utc) - timedelta(minutes=LOOKBACK_MINUTES)
    statement = text(
        """
        SELECT id, occurred_at, production_station, severity, defect_type
        FROM defect_logs
        WHERE occurred_at >= :window_start
        ORDER BY occurred_at
        """
    )
    with engine.connect() as connection:
        defects = pd.read_sql_query(statement, connection, params={"window_start": window_start})
    if not defects.empty:
        defects["occurred_at"] = pd.to_datetime(defects["occurred_at"], utc=True)
    return defects


def summarize_high_severity(defects):
    columns = ["production_station", "defect_count", "latest_event_id"]
    if defects.empty:
        return pd.DataFrame(columns=columns)
    high_severity = defects.loc[defects["severity"].eq("High")]
    if high_severity.empty:
        return pd.DataFrame(columns=columns)
    return high_severity.groupby("production_station", as_index=False).agg(
        defect_count=("id", "size"),
        latest_event_id=("id", "max"),
    )


def alert_was_sent(anomaly_key):
    statement = text(
        """
        SELECT anomaly_key
        FROM alert_events
        WHERE anomaly_key = :anomaly_key
        """
    )
    with engine.connect() as connection:
        result = connection.execute(statement, {"anomaly_key": anomaly_key})
        return result.scalar_one_or_none() is not None


def record_alert(anomaly_key, station, defect_count):
    statement = text(
        """
        INSERT INTO alert_events (anomaly_key, production_station, defect_count)
        VALUES (:anomaly_key, :production_station, :defect_count)
        """
    )
    with engine.begin() as connection:
        connection.execute(
            statement,
            {
                "anomaly_key": anomaly_key,
                "production_station": station,
                "defect_count": defect_count,
            },
        )


def send_webhook(message):
    payload = json.dumps({"content": message}).encode("utf-8")
    request = Request(
        WEBHOOK_URL,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=10) as response:
            logger.info("Webhook delivered with HTTP %s", response.status)
            return 200 <= response.status < 300
    except HTTPError as error:
        logger.error("Webhook HTTP error: %s", error.code)
    except URLError as error:
        logger.error("Webhook connection error: %s", error.reason)
    return False


def inspect_quality():
    ensure_alert_table()
    defects = load_recent_defects()
    summary = summarize_high_severity(defects)
    triggered = summary.loc[summary["defect_count"] >= DEFECT_THRESHOLD]
    if triggered.empty:
        logger.info("No station currently meets the alert threshold.")
        return
    for alert in triggered.to_dict(orient="records"):
        station = alert["production_station"]
        defect_count = int(alert["defect_count"])
        latest_event_id = int(alert["latest_event_id"])
        anomaly_key = f"{station}:{latest_event_id}"
        if alert_was_sent(anomaly_key):
            logger.info("Duplicate alert suppressed for %s", station)
            continue
        message = (
            f"QUALITY ALERT: {station} recorded {defect_count} "
            f"High-severity defects in the last {LOOKBACK_MINUTES} minutes."
        )
        if send_webhook(message):
            record_alert(anomaly_key, station, defect_count)
            logger.info("Alert recorded for %s", station)


def run_check():
    logger.info("Starting quality check")
    try:
        inspect_quality()
    except Exception:
        logger.exception("Quality check failed")


def main():
    run_check()


if __name__ == "__main__":
    main()

How to Use This Reference

Compare each function name with your saved file. Check the order of the alert lookup before the webhook call.

  • Rebuild the inspector image by running this command:
docker compose up --build --force-recreate -d inspector

What Does This Command Do?

Compose rebuilds the inspector image from your saved main.py. It recreates the one-shot inspector against the existing database volume.

Before you inspect the first run, predict whether PostgreSQL already knows about Station 5:3.

  • Inspect the first recreated inspector by running this command:
docker compose logs inspector

What Should You See First?

The first clean run sends one Station 5 alert. The log then shows Alert recorded for Station 5.

That run creates alert_events and stores Station 5:3 in the persistent PostgreSQL volume.

Did the First Run Fail?

A webhook error means the delivery was not recorded. Confirm that the ignored .env file still contains your current Discord webhook URL.

Ask for help diagnosing the first deduplicated run.

Before you recreate the inspector again, predict how the stored Station 5:3 key changes the next run.

  • Recreate the inspector a second time by running the same command:
docker compose up --build --force-recreate -d inspector

What Happens During Recreation?

Compose replaces the inspector container. The pg_data volume keeps the alert row available to the replacement process.

  • Inspect the second recreated inspector by running this command:
docker compose logs inspector

What Should You See Now?

The log contains Duplicate alert suppressed for Station 5. No webhook delivery log follows for that unchanged anomaly.

  • Return to the target Discord channel.
  • Compare its newest message count with the count after the first run.

You should see no additional Station 5 alert from the second run. PostgreSQL now remembers Station 5:3 across inspector recreations.

Still Receiving a Duplicate?

Confirm that record_alert() runs inside the successful send_webhook() condition. Check that alert_was_sent() runs before the message is created.

Ask for help tracing the duplicate key.

Your alert pipeline now remembers successful delivery beyond the life of one container. Next, you will keep the inspector running so it can repeat this protected check automatically.

Run the Inspector Every Five Minutes

Your PostgreSQL-backed deduplication now prevents repeated alerts for unchanged defect data. The inspector still exits after one check.

A one-shot script still depends on you to rerun it. The schedule library keeps checks running inside the Docker container.

In this step, get ready to:
  • Load the five-minute interval from the container environment.
  • Run one quality check immediately before registering repeated checks.
  • Watch a second check suppress the unchanged Station 5 anomaly.
Load the scheduler settings

The container already supplies the three operating values. The opening section of main.py converts them into numbers with defaults of 3, 60, and 5.

  • In the Visual Studio Code editor, replace the opening import and environment section of main.py with this block:
import json
import logging
import os
import time
from datetime import datetime, timedelta, timezone
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

import pandas as pd
import schedule
from sqlalchemy import create_engine, text

DATABASE_URL = os.environ["DATABASE_URL"]
WEBHOOK_URL = os.environ["WEBHOOK_URL"]
DEFECT_THRESHOLD = int(os.getenv("DEFECT_THRESHOLD", "3"))
LOOKBACK_MINUTES = int(os.getenv("LOOKBACK_MINUTES", "60"))
RUN_INTERVAL_MINUTES = int(os.getenv("RUN_INTERVAL_MINUTES", "5"))

What does this code do?

  • The time module pauses the polling loop between checks.
  • The schedule library tracks when the next quality check is due.
  • The RUN_INTERVAL_MINUTES setting defaults to five minutes.
  • The existing threshold defaults to three defects.
  • The existing lookback defaults to 60 minutes.
  • Save main.py.

Before you rebuild, do you think the inspector stays active after its first quality check?

  • Rebuild the current inspector by running this command:
docker compose up --build --force-recreate -d inspector

What does this command do?

Compose rebuilds the inspector image with your updated main.py. It then recreates the inspector container in detached mode.

  • Inspect the one-shot run by running this command:
docker compose logs inspector

What should I see?

You will see one Starting quality check entry. You will also see Duplicate alert suppressed for Station 5.

The output ends without a scheduled-check message. This is the one-shot limitation the scheduling loop fixes.

Did the rebuilt inspector fail?

  • Confirm that schedule==1.2.2 remains in requirements.txt.
  • Compare the opening section of main.py with the code block above.

Help me diagnose why the inspector failed after adding the scheduler imports.

Keep the inspector running

A scheduled job needs a process that remains active. The main() function runs an immediate check before entering a polling loop.

  • At the bottom of main.py, replace the one-shot entry point with this block:
def main():
    run_check()
    schedule.every(RUN_INTERVAL_MINUTES).minutes.do(run_check)
    logger.info("Scheduled quality checks every %s minute(s)", RUN_INTERVAL_MINUTES)
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    finally:
        engine.dispose()


if __name__ == "__main__":
    main()

How does the scheduling loop work?

  • The first run_check() invocation checks the database as soon as the container starts.
  • The schedule.every() call registers the repeating five-minute job.
  • The while True loop keeps the container process active.
  • The schedule.run_pending() call checks for due jobs once per second.
  • The finally block releases the database engine when the process stops.
  • Save main.py.

✔️ Awesome, I've got everything!

Your scheduling loop is ready. Confirm that main.py is saved.

ⓧ I'd like to double check the full code

Compare your complete main.py with this reference.

import json
import logging
import os
import time
from datetime import datetime, timedelta, timezone
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

import pandas as pd
import schedule
from sqlalchemy import create_engine, text

DATABASE_URL = os.environ["DATABASE_URL"]
WEBHOOK_URL = os.environ["WEBHOOK_URL"]
DEFECT_THRESHOLD = int(os.getenv("DEFECT_THRESHOLD", "3"))
LOOKBACK_MINUTES = int(os.getenv("LOOKBACK_MINUTES", "60"))
RUN_INTERVAL_MINUTES = int(os.getenv("RUN_INTERVAL_MINUTES", "5"))

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("quality-inspector")
engine = create_engine(DATABASE_URL, pool_pre_ping=True)


def ensure_alert_table():
    statement = text("""
        CREATE TABLE IF NOT EXISTS alert_events (
            anomaly_key TEXT PRIMARY KEY,
            production_station TEXT NOT NULL,
            defect_count INTEGER NOT NULL,
            sent_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
    """)
    with engine.begin() as connection:
        connection.execute(statement)


def load_recent_defects():
    window_start = datetime.now(timezone.utc) - timedelta(minutes=LOOKBACK_MINUTES)
    statement = text("""
        SELECT id, occurred_at, production_station, severity, defect_type
        FROM defect_logs WHERE occurred_at >= :window_start ORDER BY occurred_at
    """)
    with engine.connect() as connection:
        defects = pd.read_sql_query(statement, connection, params={"window_start": window_start})
    if not defects.empty:
        defects["occurred_at"] = pd.to_datetime(defects["occurred_at"], utc=True)
    return defects


def summarize_high_severity(defects):
    columns = ["production_station", "defect_count", "latest_event_id"]
    if defects.empty:
        return pd.DataFrame(columns=columns)
    high_severity = defects.loc[defects["severity"].eq("High")]
    if high_severity.empty:
        return pd.DataFrame(columns=columns)
    return high_severity.groupby("production_station", as_index=False).agg(defect_count=("id", "size"), latest_event_id=("id", "max"))


def alert_was_sent(anomaly_key):
    with engine.connect() as connection:
        return connection.execute(text("SELECT anomaly_key FROM alert_events WHERE anomaly_key = :anomaly_key"), {"anomaly_key": anomaly_key}).scalar_one_or_none() is not None


def record_alert(anomaly_key, station, defect_count):
    with engine.begin() as connection:
        connection.execute(text("INSERT INTO alert_events (anomaly_key, production_station, defect_count) VALUES (:anomaly_key, :production_station, :defect_count)"), {"anomaly_key": anomaly_key, "production_station": station, "defect_count": defect_count})


def send_webhook(message):
    request = Request(WEBHOOK_URL, data=json.dumps({"content": message}).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST")
    try:
        with urlopen(request, timeout=10) as response:
            logger.info("Webhook delivered with HTTP %s", response.status)
            return 200 <= response.status < 300
    except HTTPError as error:
        logger.error("Webhook HTTP error: %s", error.code)
    except URLError as error:
        logger.error("Webhook connection error: %s", error.reason)
    return False


def inspect_quality():
    ensure_alert_table()
    summary = summarize_high_severity(load_recent_defects())
    for alert in summary.loc[summary["defect_count"] >= DEFECT_THRESHOLD].to_dict(orient="records"):
        station, defect_count, latest_event_id = alert["production_station"], int(alert["defect_count"]), int(alert["latest_event_id"])
        anomaly_key = f"{station}:{latest_event_id}"
        if alert_was_sent(anomaly_key):
            logger.info("Duplicate alert suppressed for %s", station)
            continue
        if send_webhook(f"QUALITY ALERT: {station} recorded {defect_count} High-severity defects in the last {LOOKBACK_MINUTES} minutes."):
            record_alert(anomaly_key, station, defect_count)
            logger.info("Alert recorded for %s", station)


def run_check():
    logger.info("Starting quality check")
    try:
        inspect_quality()
    except Exception:
        logger.exception("Quality check failed")


def main():
    run_check()
    schedule.every(RUN_INTERVAL_MINUTES).minutes.do(run_check)
    logger.info("Scheduled quality checks every %s minute(s)", RUN_INTERVAL_MINUTES)
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    finally:
        engine.dispose()


if __name__ == "__main__":
    main()
Rebuild and watch the second check

This verification takes a little over five minutes because the log stream stays open until the second check is due.

Before you start the container, what do you expect the immediate check and the five-minute check to do with the existing Station 5 anomaly?

  • Rebuild the long-running inspector and follow its logs by running this command:
docker compose up --build --force-recreate -d inspector && docker compose logs -f inspector

What should I see?

The immediate run logs Starting quality check. The inspector then logs Scheduled quality checks every 5 minute(s).

Five minutes later, you will see another quality check. It logs Duplicate alert suppressed for Station 5.

Your Discord channel receives no additional copy. PostgreSQL still holds the anomaly key that both checks find.

The scheduler does not replay checks missed while the container is stopped. A restart registers a fresh schedule after the immediate startup check.

  • Press Ctrl+C after the second check appears to stop following the logs.

Still seeing only one check?

  • Confirm that main() contains the scheduler registration.
  • Confirm that the while True loop calls schedule.run_pending().
  • Keep the log stream open for the full five-minute interval.
  • Rebuild the inspector again if you saved main.py after the previous build.

Help me find why my scheduled inspector only runs once.

Your inspector now stays alive for scheduled checks. PostgreSQL keeps the unchanged Station 5 alert quiet.

Secret mission

Send Recovery Notices

Your inspector can open a quality incident without repeating the same alert. In this mission, you will persist each station's incident state and notify operators when the station recovers.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep your Docker Compose stack running. You can also pause it for another session or delete its local resources.

Your PostgreSQL database and scheduled inspector run entirely on your Linux device. This project has no metered cloud resources or ongoing cloud costs.

Resources you used:

  • The db and inspector containers plus their Compose network.
  • The pg_data volume containing defect_logs and station_alert_state.
  • The manufacturing-quality-alert folder containing the project files and local .env file.
  • The Discord incoming webhook connected to your alert channel.

Keep everything running

No action is needed. Choose this option if you are still monitoring the scheduled quality checks or testing recovery notices.

  • Switch back to the integrated terminal in Visual Studio Code.
  • Follow the live inspector output by running this command:
docker compose logs -f inspector

What Does This Command Do?

This command streams log entries from the inspector container. Each scheduled quality check appears as it reads the latest database state.

You will see the inspector continue checking for quality alerts and recoveries every five minutes.

  • Keep .env excluded through .gitignore so the webhook URL stays out of version control.

Pause - I'll come back to this later

Shut down the running processes to free local resources while keeping your files and database state. This option preserves the incident history for your next session.

  • Switch back to the integrated terminal in Visual Studio Code.
  • Stop the Compose services by running this command:
docker compose stop

What Does This Command Do?

This command stops the db and inspector containers. The pg_data volume remains available.

You will see Compose stop both services. Scheduled checks remain paused while the containers are stopped.

  • Keep the manufacturing-quality-alert folder to preserve your code and local configuration.
  • Keep the Discord webhook to reuse the same alert channel when you resume.

Delete - I don't want to use this again

Removing these resources permanently deletes your saved defect history and station incident state. The commands below target only the Compose resources and folder created for this project.

  • Switch back to the integrated terminal in Visual Studio Code.
  • Remove the containers and database volume by running this command:
docker compose down -v

What Does This Command Remove?

This command removes the db and inspector containers. It also removes their Compose network.

The -v option deletes pg_data. Your seeded defects and station incident state are permanently removed.

You will see Compose remove the project containers, network, and named volume.

  • Return to Discord.
  • Open Server Settings for the server that received your quality notifications.
  • Select Integrations.
  • Select the incoming webhook used by this project.
  • Use the webhook's delete action.
  • Confirm the project webhook no longer appears under Integrations.

Deleting the local folder also removes the .env file containing the old webhook URL. Make sure there is nothing in the folder that you want to keep.

  • Return to the integrated terminal in Visual Studio Code.
  • Remove the manufacturing-quality-alert folder by running these commands:
cd ..
rm -rf manufacturing-quality-alert

What Do These Commands Remove?

The first command moves your terminal to the folder above manufacturing-quality-alert. The second command permanently removes the project folder.

This removes every project file. It also removes the local .env file.

The terminal returns to the parent folder. The deleted workspace files no longer appear in the Visual Studio Code Explorer.

Nice Work!

Nice Work!

You did it! Your containerized quality monitor now turns recent PostgreSQL defect events into scheduled Discord alerts and recovery notices.

You've learned how to:

  • Built a rolling quality detector that uses a SQL time-window query to load recent defects. pandas filters High-severity events before grouping them by production station.
  • Delivered actionable Discord alerts containing the station name. Each message also reports the defect count and lookback window.
  • Ran a health-aware Docker Compose stack that schedules quality checks every five minutes. PostgreSQL-backed state suppresses unchanged alerts after inspector restarts.
  • Secret Mission: Upgraded deduplication into a station incident state machine that sends one QUALITY ALERT when an incident opens. It sends one QUALITY RECOVERY when the station recovers.

Ready to quiz yourself?