Build a Register Data Quality Dashboard

Build a local dashboard that finds quality issues in public register data.

Introduction

30 Second Summary

En rapport kan vise riktig antall rader uten å være til å stole på. Ett manglende navn kan fortsatt sende oppfølgingen i feil retning.

I dette prosjektet skal du bygge RegisterVakt som et lokalt dashboard for å gjøre feil i registerdata synlige for en dataeier. Appen henter åpne data fra Enhetsregisteret for å teste en simulert mottakerkopi.

What You'll Build

Du åpner RegisterVakt i nettleseren for å se et godkjent radtall ved siden av fem feilede kvalitetskontroller med konkrete rader for oppfølging.

Når prosjektet er ferdig, har du:

  • Et live kildeutvalg der du kan vise 20 underenheter med deres tilknyttede hovedenheter.
  • En radtallskontroll som viser PASS mens mottakerkopien inneholder fem kontrollerte avvik.
  • Et prioriteringsbilde for dataeieren der fire nøkkeltall peker videre til konkrete karantenefunn per kvalitetsdimensjon.
  • Secret Mission: Sammenlign kildeleveransen med mottakerkopien for å flagge kvalitetsregresjon per dimensjon.

Hvilke forkunnskaper trenger jeg?

Du bør kunne grunnleggende Python samt SQL.

Du trenger internettilgang uten å opprette noen konto.

Before We Start

Før du begynner det praktiske arbeidet, skal du forankre hvem RegisterVakt er til for og hvilket problem dashboardet løser. Like radtall kan fortsatt skjule manglende verdier, dubletter, ugyldige identifikatorer, brutte referanser og utdaterte rader.

Set Up the Local Analytics Workspace

RegisterVakt needs a workspace that behaves predictably on your machine. The same setup lets another person reproduce your analysis.

Visual Studio Code gives the project one visible workspace. A supported Python release provides the runtime for every later step.

A virtual environment keeps project packages separate from system packages. Pinned DuckDB releases make the database behavior repeatable. Pinned Streamlit releases do the same for the dashboard.

In this step, get ready to:
  • Confirm Visual Studio Code is available on Linux.
  • Create the local registervakt workspace with an isolated Python environment.
  • Prove the setup with Streamlit Hello after installing the pinned dependencies.
Install Visual Studio Code

Visual Studio Code provides the editor used for every project file. Its command-line interface also lets you open the correct folder directly from your terminal.

  • Check whether Visual Studio Code is already available by running this command:
code --version

What Does This Command Check?

The command prints the installed Visual Studio Code version. It also reports the build identifier and system architecture.

✔️ I see a version number

Visual Studio Code is already available through your terminal. You can continue to the project workspace.

ⓧ Command not found

Visual Studio Code needs to be installed before the project folder can open in the editor. Choose the package format supported by your Linux system.

A system-package installation may ask for your Linux account password. The value stays on your machine.

Debian or Ubuntu

  • Visit the official Visual Studio Code Linux setup page.
  • Download the official .deb package.
  • Move your terminal into the folder containing the downloaded package.
  • Replace <file> with the downloaded package filename without the .deb extension.
  • Install Visual Studio Code by running this command:
sudo apt install ./<file>.deb
What Does This Command Do?

The package manager installs the downloaded Visual Studio Code package. It also makes the code command available in your terminal.

  • Verify the installation by running this command:
code --version
What Should You See?

The command should print the installed version. It should also show the build identifier and system architecture.

Fedora or RHEL

  • Visit the official Visual Studio Code Linux setup page.
  • Download the official .rpm package.
  • Move your terminal into the folder containing the downloaded package.
  • Replace <file> with the downloaded package filename including the .rpm extension.
  • Install Visual Studio Code by running this command:
sudo dnf install <file>.rpm
What Does This Command Do?

The package manager installs the downloaded Visual Studio Code package. It also registers the code command for terminal use.

  • Verify the installation by running this command:
code --version
What Should You See?

The command should print the installed version. It should also show the build identifier and system architecture.

Snap

This option applies when Snap is already available on your Linux system.

  • Install the official Visual Studio Code Snap package by running this command:
sudo snap install --classic code
What Does This Command Do?

Snap installs Visual Studio Code with classic confinement. The installation also exposes the code command.

  • Verify the installation by running this command:
code --version
What Should You See?

The command should print the installed version. It should also show the build identifier and system architecture.

Visual Studio Code Still Unavailable?

Start a fresh terminal session after the installation completes. This reloads the commands available through your shell.

Confirm that the package matches your Linux distribution before retrying the installation.

Ask for targeted help with diagnosing the Visual Studio Code installation.

Create the isolated project workspace

The project folder gives every later file one predictable location. Placing the virtual environment inside it keeps the interpreter tied to RegisterVakt.

  • Move to your Desktop by running this command:
cd ~/Desktop

What Does This Command Do?

The command moves your terminal session to the Desktop. Your new project folder will be easy to find from the Linux file manager.

  • Create the registervakt folder by running these commands:
mkdir registervakt
cd registervakt
code .

What Do These Commands Do?

  • The first command creates the registervakt folder on your Desktop.
  • The second command moves the terminal session into that folder.
  • The final command opens the current folder in Visual Studio Code.

Visual Studio Code should open with registervakt shown as the current folder. You now have a stable home for every project artifact.

The Project Folder Did Not Open?

Confirm that the terminal prompt is inside registervakt. Check that the Visual Studio Code version command worked earlier.

Ask for help with opening the project folder from Linux.

Streamlit supports Python releases from 3.10 through 3.14. Checking the runtime now prevents an unsupported interpreter from entering the virtual environment.

  • Switch back to the terminal from earlier.
  • Check the available Python version by running this command:
python --version

What Does This Command Check?

The version flag prints the Python release selected by the python command. RegisterVakt requires a release from 3.10 through 3.14.

✔️ I see version 3.10 through 3.14

Your Python runtime is inside the supported range. Continue with the virtual environment commands below.

ⓧ I see a version outside the range

The displayed Python release cannot support this pinned workspace. Install a release from 3.10 through 3.14 before continuing.

  • Visit the official Python downloads page.
  • Follow the installation instructions for your Linux distribution.
  • Return to the terminal after installing a supported release.
  • Repeat the version check by running this command:
python --version

What Should You Confirm?

The reported release must be between 3.10 and 3.14 before you create the environment.

ⓧ Command not found

Your terminal cannot currently select Python through the required command. Install a supported Python release before continuing.

  • Visit the official Python downloads page.
  • Follow the installation instructions for your Linux distribution.
  • Return to the terminal after the installation finishes.
  • Repeat the version check by running this command:
python --version

What Should You Confirm?

The command should report a Python release from 3.10 through 3.14.

  • Create the project-local environment by running this command:
python -m venv .venv

What Does This Command Do?

Python creates an isolated interpreter inside .venv. Packages installed there remain separate from your system Python installation.

  • Activate the new environment by running this command:
source .venv/bin/activate

What Does Activation Change?

Activation makes the interpreter inside .venv the default for this terminal session. Package installations now stay inside the RegisterVakt workspace.

  • Verify the active interpreter by running this command:
python --version

What Should You See?

The command should still report the supported Python release you checked earlier. Your terminal prompt should also show the virtual environment name.

That is the isolation in place. RegisterVakt now has its own Python runtime context.

The Environment Did Not Activate?

Confirm that your terminal is inside the registervakt folder. Check that the .venv folder appears in the Visual Studio Code file sidebar.

Ask for targeted help with activating the Linux virtual environment.

Install the pinned analytics packages

The project needs exact package releases so later SQL controls match the tested implementation. The requirements file records those releases in one reproducible place.

  • Create .gitignore inside the registervakt folder using the Visual Studio Code file sidebar.
  • Paste the following content into .gitignore:
.venv/
__pycache__/
data/

What Does This File Protect?

The file keeps the virtual environment out of future Git changes. It also excludes Python cache files and the local database directory.

  • Save .gitignore.
  • Confirm that .gitignore appears inside registervakt in the file sidebar.

Cannot Find .gitignore?

Confirm that the filename begins with a full stop. Check that the file sits directly inside registervakt.

Ask for help with creating the Git ignore file.

  • Create requirements.txt inside the registervakt folder using the Visual Studio Code file sidebar.
  • Paste the following pinned dependencies into requirements.txt:
duckdb==1.5.6
streamlit==1.65.0

Why Pin These Versions?

The first line fixes DuckDB at 1.5.6. The second line fixes Streamlit at 1.65.0.

Exact pins keep the project on the releases used by this walkthrough.

  • Save requirements.txt.
  • Confirm that requirements.txt appears beside .gitignore in the file sidebar.

The Requirements File Looks Different?

Confirm that the filename ends with .txt. Check that each dependency occupies its own line.

Ask for help with checking the dependency file.

✔️ Awesome, I've got everything!

Great. Double-check that both files are saved inside registervakt.

ⓧ I'd like to double check the full code

Compare each saved file with the complete versions below.

.venv/
__pycache__/
data/

duckdb==1.5.6
streamlit==1.65.0

The first package installation can take about a minute while pip downloads both tools. A quiet pause during the download is normal.

  • Install the project dependencies by running this command in the activated environment:
pip install -r requirements.txt

What Does This Command Do?

The command reads both package pins from requirements.txt. It installs those releases inside the active .venv environment.

  • Wait for the terminal to return to its prompt.
  • Confirm that the installation finishes without package errors.

The Package Installation Failed?

Confirm that the terminal prompt shows the active virtual environment. Check your internet connection before retrying the installation.

Verify that your Python release remains between 3.10 and 3.14.

Ask for targeted help with diagnosing the dependency installation.

Before you run the final check, what do you expect the browser to show?

  • Start the official Streamlit test app by running this command:
python -m streamlit hello

What Does This Command Prove?

Python launches Streamlit from the active virtual environment. The Hello app confirms that the pinned dashboard package can start a local web application.

Your browser should display the Streamlit Hello app. This proves the isolated analytics workspace can serve an interactive page.

The Hello App Did Not Open?

Keep the terminal process running while you check the browser. Use the local address printed by Streamlit if the browser does not open automatically.

Confirm that the command runs from the activated .venv environment.

Ask for help with starting the Streamlit Hello app.

  • Return to the terminal running Streamlit.
  • Press Ctrl+C to stop the Hello app.

Your reproducible analytics workspace is ready. Next, you will pull a live register sample into your first RegisterVakt page.

Pull a Live Register Sample

Arbeidsområdet ditt i Visual Studio Code er klart. Nå skal Python hente registerdata som Streamlit kan vise i nettleseren.

En kvalitetsmåling trenger en tydelig kilde som referanse. Denne kilden lar deg senere måle om en mottakerkopi har blitt dårligere.

In this step, get ready to:
  • Hent et offentlig utvalg av underenheter fra Enhetsregisterets API.
  • Normaliser underenhetene med deres tilknyttede hovedenheter.
  • Lagre kilderadene i DuckDB.
Connect to the register API

Enhetsregisterets API leverer åpne virksomhetsdata uten konto. Standardresponsen for underenheter inneholder 20 rader.

Hvordan fungerer kildeforespørselen?

Responsen bruker JSON til å representere registerfeltene. fetch_json() sender forespørselen til https://data.brreg.no/enhetsregisteret/api og gjør responsen om til Python-data.

Et eget kildepunkt gjør det tydelig hvor registerdataene kommer fra. Det gir analysen et sporbarhetsanker.

  • Velg ikonet for en ny fil i filtretreet i Visual Studio Code.
  • Skriv app.py som filnavn.
  • Legg inn API-oppsettet ved å lime inn denne koden:
import json
import os
from urllib.request import Request, urlopen

import duckdb
import streamlit as st

API_ROOT = "https://data.brreg.no/enhetsregisteret/api"
DB_PATH = "data/register_quality.duckdb"

st.set_page_config(page_title="RegisterVakt", layout="wide")


def fetch_json(url):
    request = Request(
        url,
        headers={
            "Accept": "application/json",
            "User-Agent": "RegisterVakt/1.0",
        },
    )
    with urlopen(request, timeout=30) as response:
        return json.load(response)


st.title("RegisterVakt")

Hva gjør denne koden?

  • Importene gir appen støtte for HTTP-forespørsler, JSON, lokale mapper, DuckDB og Streamlit.
  • API_ROOT samler den dokumenterte API-roten på ett sted.
  • DB_PATH peker til den lokale databasefilen som appen oppretter senere i steget.
  • fetch_json() sender en forespørsel med JSON som ønsket responsformat.
  • Lagre app.py.
  • Bytt tilbake til terminalen med det aktive .venv-miljøet.
  • Start den første appversjonen ved å kjøre denne kommandoen:
python -m streamlit run app.py

Hva gjør denne kommandoen?

Kommandoen starter Streamlit med app.py som appfil. Terminalen fortsetter å kjøre prosessen mens appen er tilgjengelig i nettleseren.

Nettleseren skal vise tittelen RegisterVakt. Det bekrefter at den nye appfilen kjører i det aktive miljøet.

Vises ikke RegisterVakt?

Kontroller at app.py er lagret i RegisterVakt-mappen som allerede er åpen. Bekreft at terminalen fortsatt bruker .venv.

Be om målrettet hjelp med å finne hvorfor Streamlit ikke starter app.py.

Appen trenger nå en byggfunksjon som faktisk henter standardresponsen. Tittelen blir først vist etter at forespørselen har lykkes.

  • Finn linjen med st.title("RegisterVakt").
  • Sett inn denne koden rett over tittellinjen:
def build_database():
    os.makedirs("data", exist_ok=True)

    subunit_payload = fetch_json(f"{API_ROOT}/underenheter")
    source_subunits = subunit_payload.get("_embedded", {}).get("underenheter", [])

    if len(source_subunits) < 5:
        raise RuntimeError("API-et returnerte for få underenheter til prosjektets kontroller.")


con = build_database()

Hva henter appen nå?

  • os.makedirs() oppretter data-mappen dersom den mangler.
  • subunit_payload holder hele standardresponsen fra endepunktet for underenheter.
  • source_subunits leser radene fra _embedded.underenheter.
  • Minimumskontrollen stopper analysen dersom responsen er for liten til de senere kvalitetskontrollene.
  • Lagre app.py.
  • Oppdater nettleserfanen med RegisterVakt.

Du skal fortsatt se RegisterVakt. Tittelen vises etter API-kallet, så dette bekrefter at standardresponsen kunne hentes.

Forsvinner tittelen etter API-kallet?

Kontroller at maskinen har internettilgang. Sjekk også at API_ROOT er kopiert nøyaktig fra kodeblokken.

Be om hjelp med å feilsøke forespørselen til Enhetsregisteret.

Normalize linked register rows

Hver underenhet peker til en hovedenhet gjennom overordnetEnhet. Normalisering gir begge radtypene stabile kolonnenavn før de lagres.

  • Finn avslutningen av fetch_json() i app.py.
  • Legg inn denne funksjonen under fetch_json():
def parent_rows(parent_payloads):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in parent_payloads
    ]

Hvordan normaliseres hovedenhetene?

  • parent_rows() gjør hver hovedenhet om til en dictionary med faste nøkler.
  • orgnr og name kommer fra de dokumenterte registerfeltene.
  • org_form henter koden under organisasjonsform.
  • age_hours starter på null for den ferske kilden.
  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.

Tittelen skal fortsatt lastes uten en Python-feil i terminalen. Det bekrefter at hovedenhetsnormaliseringen har gyldig syntaks.

Ser du en syntaksfeil?

Kontroller innrykket under return. Bekreft at alle klammeparenteser i dictionaryen har en matchende avslutning.

Be om hjelp med å kontrollere parent_rows-funksjonen.

  • Plasser denne funksjonen rett under parent_rows():
def subunit_rows(subunits):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "parent_orgnr": item["overordnetEnhet"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in subunits
    ]

Hvordan normaliseres underenhetene?

  • subunit_rows() bruker den samme formen som hovedenhetene.
  • parent_orgnr bevarer koblingen fra overordnetEnhet.
  • Den felles navngivingen gjør senere SQL-kontroller enklere å gjenbruke.
  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.

RegisterVakt-tittelen skal lastes uten en ny feil i terminalen. Begge normaliseringsfunksjonene kan nå kjøres av appen.

Feiler normaliseringen?

Kontroller at parent_orgnr henter overordnetEnhet. Sjekk at feltet står innenfor dictionaryen som funksjonen returnerer.

Be om hjelp med å kontrollere subunit_rows-funksjonen.

Standardresponsen kan inneholde flere underenheter som peker til samme hovedenhet. Et sett med organisasjonsnumre sørger for at hver hovedenhet bare hentes én gang.

  • Finn minimumskontrollen inne i build_database().
  • Legg inn denne koden rett under raise RuntimeError(...)-linjen:
    parent_orgnrs = sorted(
        {item["overordnetEnhet"] for item in source_subunits}
    )
    parent_payloads = [
        fetch_json(f"{API_ROOT}/enheter/{orgnr}") for orgnr in parent_orgnrs
    ]

    source_parents = parent_rows(parent_payloads)
    source_subunits = subunit_rows(source_subunits)

Hvordan kobles leveransene?

  • parent_orgnrs samler de unike verdiene fra overordnetEnhet.
  • parent_payloads henter én hovedenhetsrespons for hvert unike organisasjonsnummer.
  • source_parents og source_subunits blir de normaliserte kildesettene.
  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.

Du skal se RegisterVakt-tittelen etter at de koblede hovedenhetene er hentet. Terminalen skal fullføre forespørslene uten traceback.

Stopper en hovedenhetsforespørsel?

Kontroller at endepunktet bygges fra API_ROOT og hvert orgnr. Bekreft at list comprehension-blokken har riktig innrykk.

Be om hjelp med å feilsøke koblingen mellom underenheter og hovedenheter.

Store and display the source sample

DuckDB lagrer de normaliserte kildesettene i én lokal fil. Tabellene gjør registerutvalget tilgjengelig for gjenbrukbare SQL-kontroller.

Hvorfor DuckDB i dette prosjektet?

DuckDB kjører lokalt og lagrer databasen under data/register_quality.duckdb. Det holder oppsettet lite.

Synlig SQL-logikk gjør kvalitetsreglene enkle å forklare i et porteføljeprosjekt. Senere steg kan kontrollere de samme tabellene uten en ekstern database.

Tabellbyggeren bruker ulike skjemaer for hovedenheter og underenheter. Den første grenen håndterer underenhetenes foreldrekobling.

  • Plasser denne funksjonen rett under subunit_rows():
def create_tables(connection, table_name, rows, is_subunit):
    if is_subunit:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                parent_orgnr VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["parent_orgnr"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )

Hvordan lagres underenhetene?

  • is_subunit velger skjemaet med parent_orgnr.
  • CREATE OR REPLACE TABLE gjør kjøringen repeterbar ved å erstatte den eksisterende tabellen.
  • executemany() setter inn de normaliserte radene med posisjonelle parametere.
  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.

RegisterVakt skal fortsatt lastes uten en syntaksfeil. Underenhetsgrenen er nå klar til å motta normaliserte rader.

Feiler underenhetsskjemaet?

Kontroller at de fem kolonnene har samme rekkefølge som verdiene i innsettingslisten. Sjekk innrykket under if is_subunit.

Be om hjelp med å kontrollere underenhetsgrenen i create_tables.

Hovedenheter trenger fire kolonner fordi de ikke peker til en egen overordnet enhet i denne modellen.

  • Legg denne else-grenen rett etter underenhetsgrenen i create_tables():
    else:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )

Hvordan lagres hovedenhetene?

Else-grenen oppretter tabellen uten parent_orgnr. Innsettingslisten følger det samme firekolonneskjemaet.

  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.

Tittelen skal lastes uten en syntaksfeil. Tabellbyggeren støtter nå begge kildesettene.

Feiler hovedenhetsgrenen?

Kontroller at else står på samme innrykksnivå som if. Bekreft at fire plassholdere brukes i innsettingsspørringen.

Be om hjelp med å kontrollere hovedenhetsgrenen i create_tables.

  • Finn normaliseringen av source_subunits inne i build_database().
  • Legg inn denne databasekoden rett under normaliseringen:
    connection = duckdb.connect(database=DB_PATH)
    create_tables(connection, "source_parents", source_parents, False)
    create_tables(connection, "source_subunits", source_subunits, True)

    return connection

Hva lagres i DuckDB?

  • duckdb.connect() åpner den vedvarende databasen på DB_PATH.
  • source_parents lagrer én normalisert rad per hentet hovedenhet.
  • source_subunits lagrer standardutvalget av underenheter.
  • build_database() returnerer den åpne tilkoblingen slik at Streamlit kan spørre tabellene.
  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.
  • Utvid data-mappen i filtretreet i Visual Studio Code.

Du skal se register_quality.duckdb i data-mappen. Databasefilen bekrefter at kildetabellene er skrevet lokalt.

Mangler databasefilen?

Kontroller at return connection står inne i build_database(). Sjekk terminalen for en feil fra tabellopprettelsen.

Be om hjelp med å finne hvorfor DuckDB-filen ikke opprettes.

Streamlit trenger Python-rader for å vise DuckDB-resultater. En liten spørringsfunksjon gjør kolonnenavnene tilgjengelige sammen med verdiene.

  • Plasser denne funksjonen rett under fetch_json():
def query_rows(connection, query):
    result = connection.execute(query)
    columns = [column[0] for column in result.description]
    return [dict(zip(columns, row)) for row in result.fetchall()]

Hvordan blir SQL-resultater til tabellrader?

query_rows() kjører en spørring gjennom DuckDB-tilkoblingen. Den kombinerer kolonnenavnene med hver resultatverdi i en dictionary.

  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.

RegisterVakt skal fortsatt lastes uten traceback. Spørringsfunksjonen er nå klar til å hente kilderadene.

Feiler resultatkonverteringen?

Kontroller at result.description brukes etter at spørringen er kjørt. Sjekk at fetchall() står i returverdien.

Be om hjelp med å kontrollere query_rows-funksjonen.

  • Finn linjen con = build_database().
  • Legg inn disse kildespørringene rett under linjen:
source_count = con.execute("SELECT COUNT(*) FROM source_subunits").fetchone()[0]

sample_rows = query_rows(
    con,
    """
    SELECT orgnr, name, parent_orgnr, org_form
    FROM source_subunits
    ORDER BY orgnr
    """,
)

Hva henter spørringene?

  • source_count teller radene i source_subunits.
  • sample_rows henter de fire feltene som forklarer hver underenhet og foreldrekoblingen.
  • ORDER BY orgnr gir tabellen en stabil sortering.
  • Lagre app.py.
  • Oppdater RegisterVakt-fanen i nettleseren.

RegisterVakt skal lastes etter at begge spørringene er kjørt. Terminalen skal ikke vise en DuckDB-feil.

Finner ikke spørringen source_subunits?

Kontroller at create_tables() bruker tabellnavnet source_subunits. Bekreft at spørringene står etter kallet til build_database().

Be om hjelp med å feilsøke kildespørringene.

  • Finn linjen st.title("RegisterVakt").
  • Legg inn denne visningen rett under tittellinjen:
metric_source, metric_passed, metric_failed, metric_findings = st.columns(4)
metric_source.metric("Kilderader", source_count, border=True)

st.write("### Kildeutvalg")
st.dataframe(sample_rows, width="stretch", hide_index=True)

con.close()

Hva viser appen?

  • Målekortet viser antallet rader som ble lagret i source_subunits.
  • Datatabellen viser orgnr, name, parent_orgnr og org_form.
  • con.close() lukker DuckDB-tilkoblingen etter at visningsdataene er hentet.
  • Lagre app.py.

Vises ikke kildetabellen?

Kontroller at visningskoden står etter kildespørringene. Bekreft at con.close() er den siste linjen.

Be om hjelp med å feilsøke Streamlit-tabellen.

Sammenlign hele filen dersom du vil kontrollere plasseringen av hver funksjon.

✔️ Awesome, I've got everything!

Flott. Kontroller at app.py er lagret før du kjører sluttkontrollen.

ⓧ I'd like to double check the full code

import json
import os
from urllib.request import Request, urlopen

import duckdb
import streamlit as st

API_ROOT = "https://data.brreg.no/enhetsregisteret/api"
DB_PATH = "data/register_quality.duckdb"

st.set_page_config(page_title="RegisterVakt", layout="wide")


def fetch_json(url):
    request = Request(
        url,
        headers={
            "Accept": "application/json",
            "User-Agent": "RegisterVakt/1.0",
        },
    )
    with urlopen(request, timeout=30) as response:
        return json.load(response)


def query_rows(connection, query):
    result = connection.execute(query)
    columns = [column[0] for column in result.description]
    return [dict(zip(columns, row)) for row in result.fetchall()]


def parent_rows(parent_payloads):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in parent_payloads
    ]


def subunit_rows(subunits):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "parent_orgnr": item["overordnetEnhet"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in subunits
    ]


def create_tables(connection, table_name, rows, is_subunit):
    if is_subunit:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                parent_orgnr VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["parent_orgnr"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )
    else:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )


def build_database():
    os.makedirs("data", exist_ok=True)

    subunit_payload = fetch_json(f"{API_ROOT}/underenheter")
    source_subunits = subunit_payload.get("_embedded", {}).get("underenheter", [])

    if len(source_subunits) < 5:
        raise RuntimeError("API-et returnerte for få underenheter til prosjektets kontroller.")

    parent_orgnrs = sorted(
        {item["overordnetEnhet"] for item in source_subunits}
    )
    parent_payloads = [
        fetch_json(f"{API_ROOT}/enheter/{orgnr}") for orgnr in parent_orgnrs
    ]

    source_parents = parent_rows(parent_payloads)
    source_subunits = subunit_rows(source_subunits)

    connection = duckdb.connect(database=DB_PATH)
    create_tables(connection, "source_parents", source_parents, False)
    create_tables(connection, "source_subunits", source_subunits, True)

    return connection


con = build_database()

source_count = con.execute("SELECT COUNT(*) FROM source_subunits").fetchone()[0]

sample_rows = query_rows(
    con,
    """
    SELECT orgnr, name, parent_orgnr, org_form
    FROM source_subunits
    ORDER BY orgnr
    """,
)

st.title("RegisterVakt")

metric_source, metric_passed, metric_failed, metric_findings = st.columns(4)
metric_source.metric("Kilderader", source_count, border=True)

st.write("### Kildeutvalg")
st.dataframe(sample_rows, width="stretch", hide_index=True)

con.close()
  • Stopp Streamlit-prosessen i terminalen som kjører appen.

Før du kjører sluttkontrollen, tror du kildetabellen vil inneholde standardutvalget på 20 underenheter?

  • Start den fullførte kildevisningen ved å kjøre denne kommandoen:
python -m streamlit run app.py

Hva bekrefter sluttkontrollen?

Kommandoen bygger DuckDB-filen fra den levende API-responsen før Streamlit viser siden. En vellykket visning bekrefter hele kjeden fra kilde til lokal tabell.

Du skal se 20 i målekortet for kilderader. Tabellen skal vise kolonnene orgnr, name, parent_orgnr og org_form.

Godt jobbet. RegisterVakt har nå en levende og sporbar registerkilde som de neste kvalitetskontrollene kan måle mot.

Kildetabellene er klare. Neste steg lager en mottakerkopi med samme radtall og skjulte kvalitetsfeil.

Prove Why Row Counts Are Not Enough

Kildeutvalget ditt ligger nå i DuckDB. Det gir deg en tydelig referanse for å teste en nedstrøms mottakerkopi.

En mottakerkopi kan beholde samme radtall samtidig som innholdet blir dårligere. I dette steget lager du fem kontrollerte avvik som avslører svakheten ved en ren radtallskontroll.

In this step, get ready to:
  • Opprette mottakertabeller med samme antall rader som kildetabellene.
  • Injisere fem kontrollerte kvalitetsavvik i mottakerkopien.
  • Kjøre en naiv radtallskontroll som gir PASS til tross for avvikene.
Opprett en mottakerkopi med samme størrelse

Mottakerkopien simulerer data som har beveget seg fra kilderegisteret til et annet system. Separate Python-dictionaryer sørger for at du kan endre kopien uten å endre kildedataene.

  • Bytt tilbake til app.py i Visual Studio Code.
  • Finn linjen source_parents = parent_rows(parent_payloads) i build_database().
  • Marker fra denne linjen til linjen con = build_database().
  • Erstatt den markerte delen med denne utvidede databaseseksjonen:
    source_parents = parent_rows(parent_payloads)
    source_subunits = subunit_rows(source_subunits)

    consumer_parents = [dict(row) for row in source_parents]
    consumer_subunits = [dict(row) for row in source_subunits]

    for row in consumer_parents:
        row["age_hours"] = 4
    for row in consumer_subunits:
        row["age_hours"] = 4

    connection = duckdb.connect(database=DB_PATH)
    create_tables(connection, "source_parents", source_parents, False)
    create_tables(connection, "source_subunits", source_subunits, True)
    create_tables(connection, "consumer_parents", consumer_parents, False)
    create_tables(connection, "consumer_subunits", consumer_subunits, True)

    return connection

con = build_database()

source_count = con.execute("SELECT COUNT(*) FROM source_subunits").fetchone()[0]
consumer_count = con.execute("SELECT COUNT(*) FROM consumer_subunits").fetchone()[0]
naive_status = "PASS" if source_count == consumer_count else "FAIL"

Hva Gjør Denne Koden?

  • List comprehension lager en selvstendig dictionary for hver mottakerrad.
  • Løkkene setter mottakerens leveringsalder til fire timer.
  • De to nye tabellkallene lagrer mottakerkopien i den samme lokale databasen.
  • Radtallsspørringene måler størrelsen på kildeutvalget og mottakerkopien.
  • Uttrykket for naive_status vurderer bare om de to radtallene er like.

Den naive statusen trenger en synlig plass i Streamlit-appen. Bildeteksten viser begge radtallene slik at du kan kontrollere grunnlaget for resultatet.

  • Finn bildeteksten under st.title("RegisterVakt").
  • Lim inn denne nye bildeteksten rett under den eksisterende bildeteksten:
st.caption(
    f"Naiv radtallskontroll: {naive_status}. Kilde: {source_count} rader. "
    f"Mottakerkopi: {consumer_count} rader."
)

Hva Viser Bildeteksten?

Bildeteksten setter statusen ved siden av de to radtallene. Du kan dermed se nøyaktig hvorfor den naive kontrollen består.

  • Lagre app.py.
  • Start den oppdaterte appen fra den aktiverte terminalen ved å kjøre denne kommandoen:
python -m streamlit run app.py

Hva Gjør Denne Kommandoen?

Kommandoen starter app.py gjennom Streamlit-modulen i det aktiverte virtuelle miljøet. Appen bygger begge tabellsettene før siden vises.

Du skal se PASS sammen med identiske radtall for kilde og mottakerkopi. Du har nå en kontrollert grunnlinje for testen.

Ser Du Ikke Begge Radtallene?

Kontroller at source_count og consumer_count står etter con = build_database().

Sjekk at begge mottakertabellene opprettes før build_database() returnerer tilkoblingen.

Be om målrettet hjelp med å feilsøke mottakertabellene.

Injiser fem kontrollerte avvik

Kontrollerte avvik gjør svakheten målbar uten å legge feilene på Enhetsregisteret. Alle endringene skjer i den lokale mottakerkopien.

  • Finn løkken som setter age_hours til 4 for consumer_subunits.
  • Lim inn disse fem endringene rett under løkken:
    consumer_parents[0]["name"] = ""
    consumer_subunits[-1]["orgnr"] = consumer_subunits[0]["orgnr"]
    consumer_subunits[1]["orgnr"] = "12345"
    consumer_subunits[2]["parent_orgnr"] = "000000000"
    consumer_subunits[3]["age_hours"] = 72

Hvilke Avvik Injiseres?

  • Den første hovedenheten får et tomt navn.
  • Den siste underenheten får samme organisasjonsnummer som den første underenheten.
  • Den andre underenheten får det femsifrede organisasjonsnummeret 12345.
  • Den tredje underenheten peker til den ukjente hovedenheten 000000000.
  • Den fjerde underenheten får en leveringsalder på 72 timer.
  • Lagre app.py.

Før du kjører appen igjen, tror du radtallsstatusen oppdager de fem innholdsfeilene?

  • Kjør appen med den endrede mottakerkopien ved å bruke denne kommandoen:
python -m streamlit run app.py

Hva Tester Denne Kjøringen?

Appen bygger mottakertabellene på nytt med de fem avvikene. Den naive kontrollen sammenligner fortsatt bare antall rader.

Du ser fortsatt PASS. Kilden og mottakerkopien har fortsatt samme radtall.

Dette er prosjektets planlagte feiløyeblikk. Ett grønt resultat skjuler et manglende navn, en dublett, en ugyldig identifikator, en foreldreløs kobling og en utdatert rad.

Stopper Appen Etter Endringen?

Kontroller innrykket på de fem endringene. Alle linjene skal ligge inne i build_database() før databasetilkoblingen opprettes.

Sjekk at standardutvalget inneholder minst fem underenheter før radindeksene brukes.

Be om hjelp med å finne feilen i avviksinjiseringen.

Dobbeltsjekk den samlede appen

Den samlede filen skal nå bevare kildedataene samtidig som den endrer mottakerkopien. Bruk referansen under for å fange opp manglende linjer før den siste kontrollen.

✔️ Jeg har alle endringene

Flott. Lagre app.py før du kjører sluttkontrollen.

ⓧ Jeg vil dobbeltsjekke hele koden

import json
import os
from urllib.request import Request, urlopen

import duckdb
import streamlit as st

API_ROOT = "https://data.brreg.no/enhetsregisteret/api"
DB_PATH = "data/register_quality.duckdb"

st.set_page_config(page_title="RegisterVakt", layout="wide")


def fetch_json(url):
    request = Request(
        url,
        headers={
            "Accept": "application/json",
            "User-Agent": "RegisterVakt/1.0",
        },
    )
    with urlopen(request, timeout=30) as response:
        return json.load(response)


def query_rows(connection, query):
    result = connection.execute(query)
    columns = [column[0] for column in result.description]
    return [dict(zip(columns, row)) for row in result.fetchall()]


def parent_rows(parent_payloads):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in parent_payloads
    ]


def subunit_rows(subunits):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "parent_orgnr": item["overordnetEnhet"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in subunits
    ]


def create_tables(connection, table_name, rows, is_subunit):
    if is_subunit:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                parent_orgnr VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["parent_orgnr"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )
    else:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )


def build_database():
    os.makedirs("data", exist_ok=True)

    subunit_payload = fetch_json(f"{API_ROOT}/underenheter")
    source_subunits = subunit_payload.get("_embedded", {}).get("underenheter", [])

    if len(source_subunits) < 5:
        raise RuntimeError("API-et returnerte for få underenheter til prosjektets kontroller.")

    parent_orgnrs = sorted(
        {item["overordnetEnhet"] for item in source_subunits}
    )
    parent_payloads = [
        fetch_json(f"{API_ROOT}/enheter/{orgnr}") for orgnr in parent_orgnrs
    ]

    source_parents = parent_rows(parent_payloads)
    source_subunits = subunit_rows(source_subunits)

    consumer_parents = [dict(row) for row in source_parents]
    consumer_subunits = [dict(row) for row in source_subunits]

    for row in consumer_parents:
        row["age_hours"] = 4
    for row in consumer_subunits:
        row["age_hours"] = 4

    consumer_parents[0]["name"] = ""
    consumer_subunits[-1]["orgnr"] = consumer_subunits[0]["orgnr"]
    consumer_subunits[1]["orgnr"] = "12345"
    consumer_subunits[2]["parent_orgnr"] = "000000000"
    consumer_subunits[3]["age_hours"] = 72

    connection = duckdb.connect(database=DB_PATH)
    create_tables(connection, "source_parents", source_parents, False)
    create_tables(connection, "source_subunits", source_subunits, True)
    create_tables(connection, "consumer_parents", consumer_parents, False)
    create_tables(connection, "consumer_subunits", consumer_subunits, True)

    return connection


con = build_database()

source_count = con.execute("SELECT COUNT(*) FROM source_subunits").fetchone()[0]
consumer_count = con.execute("SELECT COUNT(*) FROM consumer_subunits").fetchone()[0]
naive_status = "PASS" if source_count == consumer_count else "FAIL"

sample_rows = query_rows(
    con,
    """
    SELECT orgnr, name, parent_orgnr, org_form
    FROM source_subunits
    ORDER BY orgnr
    """,
)

st.title("RegisterVakt")
st.caption(
    "Datakvalitetskontroll av en simulert mottakerkopi basert på åpne data fra Enhetsregisteret."
)
st.caption(
    f"Naiv radtallskontroll: {naive_status}. Kilde: {source_count} rader. "
    f"Mottakerkopi: {consumer_count} rader."
)

st.write("### Kildeutvalg")
st.dataframe(sample_rows, width="stretch", hide_index=True)

st.caption(
    "Datakilde: https://data.brreg.no/enhetsregisteret/api. "
    "Dataene følger Norsk lisens for offentlige data. "
    "Prosjektet bruker ikke data fra Skatteetaten."
)

con.close()

Slik Bruker Du Referansen

Sammenlign funksjonene og rekkefølgen i build_database() med din egen fil. Kontroller spesielt at mottakerradene endres før de skrives til DuckDB.

Før sluttkontrollen, hvilken status forventer du når de to tabellene fortsatt har samme antall rader?

  • Kjør den ferdige versjonen av dette steget med denne kommandoen:
python -m streamlit run app.py

Hva Bekrefter Sluttkontrollen?

Kjøringen bygger kilde- og mottakertabellene fra bunnen av. Bildeteksten bruker de ferdige tabellene til å beregne den naive statusen.

Nettleseren skal vise PASS for radtallskontrollen. Den skal også vise identiske radtall for kilde og mottakerkopi.

Du har bevist at identiske radtall kan skjule alvorlige innholdsfeil. Neste steg gjør de skjulte avvikene synlige med gjenbrukbare SQL-regler.

Build Reusable SQL Quality Controls

Your consumer copy now passes the row-count check despite containing five controlled defects. That result proves the copy needs checks that inspect individual quality dimensions.

In this step, you will express those checks as reusable SQL rules. You will store both summary results and traceable row-level findings in DuckDB.

In this step, get ready to:
  • Define five quality rules for the consumer tables.
  • Collect every row-level failure in a quarantine table.
  • Run the rules from the app and verify five failed controls.
Define the five quality rules

A quality rule turns an expectation into a measurable query. Each rule records its dimension, the number of failed rows, the number of checked rows, and a status.

Why Use Explicit SQL Rules?

Keeping the rules in SQL makes every quality decision visible in the project. A reviewer can inspect the exact conditions behind each result.

This approach also keeps the same rules available for future deliveries. Reuse makes changes in quality easier to compare.

The first table summarizes five controls as one row per rule. You will assemble its SQL statement in four adjacent sections before saving the file.

  • Return to the Visual Studio Code window from earlier.
  • Create quality_checks.sql beside app.py using the file sidebar's new-file control.
  • Add the completeness rule as the first section of quality_checks.sql by pasting this code:
CREATE OR REPLACE TABLE quality_results AS
WITH checks AS (
    SELECT
        'DQ-001' AS rule_id,
        'Fullstendighet' AS dimension,
        'Hovedenhetens navn skal være utfylt' AS rule_description,
        SUM(CASE WHEN name IS NULL OR trim(name) = '' THEN 1 ELSE 0 END) AS failed_rows,
        COUNT(*) AS total_rows
    FROM consumer_parents

What Does This Section Add?

  • The checks common table expression gives all five controls one shared result shape.
  • Rule DQ-001 counts parent rows whose name is null or blank.
  • The control records how many parent rows were checked.
  • Continue the same SQL statement by adding the duplicate rule directly below the previous section:

    UNION ALL

    SELECT
        'DQ-002',
        'Dublett',
        'Underenhetens organisasjonsnummer skal være unikt',
        (
            SELECT COUNT(*)
            FROM consumer_subunits
            WHERE orgnr IN (
                SELECT orgnr
                FROM consumer_subunits
                GROUP BY orgnr
                HAVING COUNT(*) > 1
            )
        ),
        (SELECT COUNT(*) FROM consumer_subunits)

How Are Duplicates Counted?

Rule DQ-002 finds organization numbers that occur more than once. It then counts every row belonging to those duplicate groups.

This makes the result match the number of affected records. The quarantine table can point to those same records later.

  • Add the accuracy rule and the consistency rule directly below the duplicate rule:

    UNION ALL

    SELECT
        'DQ-003',
        'Nøyaktighet',
        'Organisasjonsnummer skal bestå av nøyaktig ni siffer',
        SUM(CASE WHEN NOT regexp_full_match(orgnr, '^[0-9]{9}$') THEN 1 ELSE 0 END),
        COUNT(*)
    FROM consumer_subunits

    UNION ALL

    SELECT
        'DQ-004',
        'Konsistens',
        'Underenheten skal peke til en eksisterende hovedenhet',
        COUNT(*),
        (SELECT COUNT(*) FROM consumer_subunits)
    FROM consumer_subunits AS subunit
    LEFT JOIN consumer_parents AS parent
        ON subunit.parent_orgnr = parent.orgnr
    WHERE parent.orgnr IS NULL

What Do These Rules Measure?

  • Rule DQ-003 requires the complete organization number to match exactly nine numeric characters.
  • Rule DQ-004 joins each subunit to its referenced parent.
  • A missing parent match becomes a consistency failure.
  • Complete the statement by adding the freshness rule and the final result projection:

    UNION ALL

    SELECT
        'DQ-005',
        'Aktualitet',
        'Mottakerkopien skal være maksimalt 24 timer gammel',
        SUM(CASE WHEN age_hours > 24 THEN 1 ELSE 0 END),
        COUNT(*)
    FROM consumer_subunits
)
SELECT
    rule_id,
    dimension,
    rule_description,
    failed_rows,
    total_rows,
    CASE WHEN failed_rows = 0 THEN 'PASS' ELSE 'FAIL' END AS status
FROM checks;

How Is the Status Decided?

Rule DQ-005 counts rows older than the allowed 24-hour limit. The final projection assigns PASS only when a rule has zero failed rows.

Every nonzero failure count becomes FAIL. The status therefore follows one consistent decision rule.

  • Save quality_checks.sql.
  • Confirm that quality_checks.sql remains listed beside app.py in the file sidebar.
Collect row-level quarantine findings

A failed rule shows the size of a problem. A quarantine table preserves the records behind that number so a data owner can investigate each finding.

  • Add the table declaration and the first two quarantine rules at the bottom of quality_checks.sql by pasting this code:

CREATE OR REPLACE TABLE quarantined_records AS
SELECT
    'DQ-001' AS rule_id,
    'consumer_parents' AS dataset,
    orgnr,
    'Mangler navn' AS issue
FROM consumer_parents
WHERE name IS NULL OR trim(name) = ''

UNION ALL

SELECT
    'DQ-002',
    'consumer_subunits',
    orgnr,
    'Duplisert organisasjonsnummer'
FROM consumer_subunits
WHERE orgnr IN (
    SELECT orgnr
    FROM consumer_subunits
    GROUP BY orgnr
    HAVING COUNT(*) > 1
)

What Does the Quarantine Store?

Each finding carries its rule ID, dataset, organization number, and issue description. These columns connect the summary count to the affected row.

The first two queries capture missing parent names and every row in a duplicate group.

  • Continue the quarantine statement by adding the accuracy and consistency findings:

UNION ALL

SELECT
    'DQ-003',
    'consumer_subunits',
    orgnr,
    'Organisasjonsnummeret er ikke nisifret og numerisk'
FROM consumer_subunits
WHERE NOT regexp_full_match(orgnr, '^[0-9]{9}$')

UNION ALL

SELECT
    'DQ-004',
    'consumer_subunits',
    subunit.orgnr,
    'Overordnet hovedenhet finnes ikke i mottakerkopien'
FROM consumer_subunits AS subunit
LEFT JOIN consumer_parents AS parent
    ON subunit.parent_orgnr = parent.orgnr
WHERE parent.orgnr IS NULL

How Do These Findings Stay Traceable?

The accuracy query selects the exact organization numbers with an invalid format. The consistency query selects the subunits whose parent reference has no match.

Both queries use the same conditions as their summary rules. The total and the evidence stay aligned.

  • Finish the quarantine statement by adding the freshness findings:

UNION ALL

SELECT
    'DQ-005',
    'consumer_subunits',
    orgnr,
    'Leveransen er eldre enn 24 timer'
FROM consumer_subunits
WHERE age_hours > 24;

What Completes the Quarantine?

The final query adds every consumer subunit older than 24 hours. Its rule ID connects each stale row to the freshness result in quality_results.

  • Save the completed quality_checks.sql file.

✔️ Awesome, I've got everything!

Your SQL file now defines five summary controls and one row-level quarantine table. Keep the file saved beside app.py.

ⓧ I'd like to double check the full code

CREATE OR REPLACE TABLE quality_results AS
WITH checks AS (
    SELECT
        'DQ-001' AS rule_id,
        'Fullstendighet' AS dimension,
        'Hovedenhetens navn skal være utfylt' AS rule_description,
        SUM(CASE WHEN name IS NULL OR trim(name) = '' THEN 1 ELSE 0 END) AS failed_rows,
        COUNT(*) AS total_rows
    FROM consumer_parents

    UNION ALL

    SELECT
        'DQ-002',
        'Dublett',
        'Underenhetens organisasjonsnummer skal være unikt',
        (
            SELECT COUNT(*)
            FROM consumer_subunits
            WHERE orgnr IN (
                SELECT orgnr
                FROM consumer_subunits
                GROUP BY orgnr
                HAVING COUNT(*) > 1
            )
        ),
        (SELECT COUNT(*) FROM consumer_subunits)

    UNION ALL

    SELECT
        'DQ-003',
        'Nøyaktighet',
        'Organisasjonsnummer skal bestå av nøyaktig ni siffer',
        SUM(CASE WHEN NOT regexp_full_match(orgnr, '^[0-9]{9}$') THEN 1 ELSE 0 END),
        COUNT(*)
    FROM consumer_subunits

    UNION ALL

    SELECT
        'DQ-004',
        'Konsistens',
        'Underenheten skal peke til en eksisterende hovedenhet',
        COUNT(*),
        (SELECT COUNT(*) FROM consumer_subunits)
    FROM consumer_subunits AS subunit
    LEFT JOIN consumer_parents AS parent
        ON subunit.parent_orgnr = parent.orgnr
    WHERE parent.orgnr IS NULL

    UNION ALL

    SELECT
        'DQ-005',
        'Aktualitet',
        'Mottakerkopien skal være maksimalt 24 timer gammel',
        SUM(CASE WHEN age_hours > 24 THEN 1 ELSE 0 END),
        COUNT(*)
    FROM consumer_subunits
)
SELECT
    rule_id,
    dimension,
    rule_description,
    failed_rows,
    total_rows,
    CASE WHEN failed_rows = 0 THEN 'PASS' ELSE 'FAIL' END AS status
FROM checks;

CREATE OR REPLACE TABLE quarantined_records AS
SELECT
    'DQ-001' AS rule_id,
    'consumer_parents' AS dataset,
    orgnr,
    'Mangler navn' AS issue
FROM consumer_parents
WHERE name IS NULL OR trim(name) = ''

UNION ALL

SELECT
    'DQ-002',
    'consumer_subunits',
    orgnr,
    'Duplisert organisasjonsnummer'
FROM consumer_subunits
WHERE orgnr IN (
    SELECT orgnr
    FROM consumer_subunits
    GROUP BY orgnr
    HAVING COUNT(*) > 1
)

UNION ALL

SELECT
    'DQ-003',
    'consumer_subunits',
    orgnr,
    'Organisasjonsnummeret er ikke nisifret og numerisk'
FROM consumer_subunits
WHERE NOT regexp_full_match(orgnr, '^[0-9]{9}$')

UNION ALL

SELECT
    'DQ-004',
    'consumer_subunits',
    subunit.orgnr,
    'Overordnet hovedenhet finnes ikke i mottakerkopien'
FROM consumer_subunits AS subunit
LEFT JOIN consumer_parents AS parent
    ON subunit.parent_orgnr = parent.orgnr
WHERE parent.orgnr IS NULL

UNION ALL

SELECT
    'DQ-005',
    'consumer_subunits',
    orgnr,
    'Leveransen er eldre enn 24 timer'
FROM consumer_subunits
WHERE age_hours > 24;

Compare this reference with your saved file from top to bottom. The order keeps each summary rule aligned with its quarantine query.

Run the controls from the app

The database build already finishes by creating the four source and consumer tables. The next addition executes quality_checks.sql against that same connection.

  • In app.py, locate the two create_tables() calls for the consumer tables near the end of build_database().
  • Replace the ending of build_database() with this expanded section:
    create_tables(connection, "consumer_parents", consumer_parents, False)
    create_tables(connection, "consumer_subunits", consumer_subunits, True)

    with open("quality_checks.sql", encoding="utf-8") as sql_file:
        connection.execute(sql_file.read())

    return connection

What Does This Code Do?

  • The file is opened with UTF-8 encoding so its Norwegian rule descriptions load correctly.
  • The complete SQL file is executed after all four input tables exist.
  • The same database connection now contains the two quality output tables.
  • Find the line that assigns naive_status below con = build_database().
  • Add the summary and detailed result queries directly below that line:
summary = query_rows(
    con,
    """
    SELECT
        COUNT(*) AS total_rules,
        SUM(CASE WHEN status = 'PASS' THEN 1 ELSE 0 END) AS passed_rules,
        SUM(CASE WHEN status = 'FAIL' THEN 1 ELSE 0 END) AS failed_rules
    FROM quality_results
    """,
)[0]

finding_count = con.execute(
    "SELECT COUNT(*) FROM quarantined_records"
).fetchone()[0]

results = query_rows(
    con,
    """
    SELECT rule_id, dimension, rule_description, failed_rows, total_rows, status
    FROM quality_results
    ORDER BY rule_id
    """,
)

What Do These Queries Prepare?

  • The summary query counts all rules and groups their outcomes into passed and failed totals.
  • The finding query counts every row-level quarantine record.
  • The results query keeps the five rules ordered from DQ-001 through DQ-005.
  • Save app.py.
  • Return to the terminal with the active .venv.
  • Start the app with the new SQL execution path by running this command:
python -m streamlit run app.py

What Does This Check Prove?

The Streamlit process rebuilds the local database before rendering the page. A successful load proves the SQL file executed and the result queries found both quality tables.

The browser should load RegisterVakt with the naive PASS result still visible. The terminal should remain free of SQL file and missing-table errors.

Did the App Stop During the SQL Run?

Confirm that quality_checks.sql is saved beside app.py. Check that each UNION ALL section appears in the same order as the full-file reference.

Ask for targeted help with diagnosing the SQL execution error.

  • Return to app.py while the app process remains active.
  • Locate the results = query_rows( block you just added.
  • Add the dimension aggregation and quarantine query directly below that block:
failures_by_dimension = query_rows(
    con,
    """
    SELECT dimension, SUM(failed_rows) AS failed_rows
    FROM quality_results
    GROUP BY dimension
    ORDER BY dimension
    """,
)

quarantine = query_rows(
    con,
    """
    SELECT rule_id, dataset, orgnr, issue
    FROM quarantined_records
    ORDER BY rule_id, orgnr
    """,
)

What Do These Queries Add?

The first query groups failed rows by quality dimension. The second query retrieves every quarantined record in a predictable rule and organization-number order.

These datasets support both the detailed evidence in this step and the dashboard visualizations in the next step.

  • Find the existing source display that starts with st.write("### Kildeutvalg").
  • Replace that display section with the quality tables followed by the source sample:
st.write("### Kontrollresultater")
st.dataframe(results, width="stretch", hide_index=True)

st.write("### Karantene for oppfølging")
st.dataframe(quarantine, width="stretch", hide_index=True)

st.write("### Kildeutvalg")
st.dataframe(sample_rows, width="stretch", hide_index=True)

What Will the Page Expose?

The first table shows each rule's dimension, description, failure count, checked-row count, and status. The second table exposes the records behind those failures.

The source sample stays on the page as a reference for the live register delivery.

Before you check the refreshed page, how many rule rows do you expect to show FAIL?

  • Save app.py.
  • Refresh the running RegisterVakt page in your browser.

You should see five rows in the control table. Every row should have the status FAIL.

You should also see quarantine findings for the missing name, duplicate organization number, invalid organization number, unknown parent, and stale row. Each finding should carry the matching rule ID.

Missing a Failed Rule or Finding?

Compare the rule condition with its matching quarantine condition. Both places must use the same consumer table and the same failure test.

Check that the app executes quality_checks.sql after creating both consumer tables. An earlier execution point leaves the rules without their inputs.

Ask for help with tracing a missing quality result.

✔️ Awesome, I've got everything!

Your app now executes all five reusable rules and shows their row-level evidence. Save app.py before continuing.

ⓧ I'd like to double check the full code

import json
import os
from urllib.request import Request, urlopen

import duckdb
import streamlit as st

API_ROOT = "https://data.brreg.no/enhetsregisteret/api"
DB_PATH = "data/register_quality.duckdb"

st.set_page_config(page_title="RegisterVakt", layout="wide")


def fetch_json(url):
    request = Request(
        url,
        headers={
            "Accept": "application/json",
            "User-Agent": "RegisterVakt/1.0",
        },
    )
    with urlopen(request, timeout=30) as response:
        return json.load(response)


def query_rows(connection, query):
    result = connection.execute(query)
    columns = [column[0] for column in result.description]
    return [dict(zip(columns, row)) for row in result.fetchall()]


def parent_rows(parent_payloads):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in parent_payloads
    ]


def subunit_rows(subunits):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "parent_orgnr": item["overordnetEnhet"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in subunits
    ]


def create_tables(connection, table_name, rows, is_subunit):
    if is_subunit:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                parent_orgnr VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["parent_orgnr"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )
    else:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )


def build_database():
    os.makedirs("data", exist_ok=True)

    subunit_payload = fetch_json(f"{API_ROOT}/underenheter")
    source_subunits = subunit_payload.get("_embedded", {}).get("underenheter", [])

    if len(source_subunits) < 5:
        raise RuntimeError("API-et returnerte for få underenheter til prosjektets kontroller.")

    parent_orgnrs = sorted(
        {item["overordnetEnhet"] for item in source_subunits}
    )
    parent_payloads = [
        fetch_json(f"{API_ROOT}/enheter/{orgnr}") for orgnr in parent_orgnrs
    ]

    source_parents = parent_rows(parent_payloads)
    source_subunits = subunit_rows(source_subunits)

    consumer_parents = [dict(row) for row in source_parents]
    consumer_subunits = [dict(row) for row in source_subunits]

    for row in consumer_parents:
        row["age_hours"] = 4
    for row in consumer_subunits:
        row["age_hours"] = 4

    consumer_parents[0]["name"] = ""
    consumer_subunits[-1]["orgnr"] = consumer_subunits[0]["orgnr"]
    consumer_subunits[1]["orgnr"] = "12345"
    consumer_subunits[2]["parent_orgnr"] = "000000000"
    consumer_subunits[3]["age_hours"] = 72

    connection = duckdb.connect(database=DB_PATH)
    create_tables(connection, "source_parents", source_parents, False)
    create_tables(connection, "source_subunits", source_subunits, True)
    create_tables(connection, "consumer_parents", consumer_parents, False)
    create_tables(connection, "consumer_subunits", consumer_subunits, True)

    with open("quality_checks.sql", encoding="utf-8") as sql_file:
        connection.execute(sql_file.read())

    return connection


con = build_database()

source_count = con.execute("SELECT COUNT(*) FROM source_subunits").fetchone()[0]
consumer_count = con.execute("SELECT COUNT(*) FROM consumer_subunits").fetchone()[0]
naive_status = "PASS" if source_count == consumer_count else "FAIL"

summary = query_rows(
    con,
    """
    SELECT
        COUNT(*) AS total_rules,
        SUM(CASE WHEN status = 'PASS' THEN 1 ELSE 0 END) AS passed_rules,
        SUM(CASE WHEN status = 'FAIL' THEN 1 ELSE 0 END) AS failed_rules
    FROM quality_results
    """,
)[0]

finding_count = con.execute(
    "SELECT COUNT(*) FROM quarantined_records"
).fetchone()[0]

results = query_rows(
    con,
    """
    SELECT rule_id, dimension, rule_description, failed_rows, total_rows, status
    FROM quality_results
    ORDER BY rule_id
    """,
)

failures_by_dimension = query_rows(
    con,
    """
    SELECT dimension, SUM(failed_rows) AS failed_rows
    FROM quality_results
    GROUP BY dimension
    ORDER BY dimension
    """,
)

quarantine = query_rows(
    con,
    """
    SELECT rule_id, dataset, orgnr, issue
    FROM quarantined_records
    ORDER BY rule_id, orgnr
    """,
)

sample_rows = query_rows(
    con,
    """
    SELECT orgnr, name, parent_orgnr, org_form
    FROM source_subunits
    ORDER BY orgnr
    """,
)

st.title("RegisterVakt")
st.caption(
    "Datakvalitetskontroll av en simulert mottakerkopi basert på åpne data fra Enhetsregisteret."
)
st.caption(
    f"Naiv radtallskontroll: {naive_status}. Kilde: {source_count} rader. "
    f"Mottakerkopi: {consumer_count} rader."
)

st.write("### Kontrollresultater")
st.dataframe(results, width="stretch", hide_index=True)

st.write("### Karantene for oppfølging")
st.dataframe(quarantine, width="stretch", hide_index=True)

st.write("### Kildeutvalg")
st.dataframe(sample_rows, width="stretch", hide_index=True)

st.caption(
    "Datakilde: https://data.brreg.no/enhetsregisteret/api. "
    "Dataene følger Norsk lisens for offentlige data. "
    "Prosjektet bruker ikke data fra Skatteetaten."
)

con.close()

Use this reference to compare the cumulative app.py for this step. Keep the quality tables above the source sample on the page.

Your hidden defects now produce measurable failures and traceable evidence. Next, you will turn these results into a dashboard a data owner can use for prioritization.

Turn Findings into a Data-Owner Dashboard

De fem kontrollene viser nå konkrete avvik i DuckDB. Neste mål er å gjøre funnene forståelige for en dataeier.

Rå SQL-resultater gir liten verdi før omfanget blir synlig. Dashboardet skal vise hvilke dimensjoner som krever oppfølging.

In this step, get ready to:
  • Bygg fire målekort som oppsummerer kilderader og kontrollresultater.
  • Vis avvik per kvalitetsdimensjon og sporbare karantenefunn.
  • Dokumenter prosjektet i README.md før du lagrer det i Git.
Build the dashboard overview

En dataeier trenger en rask oppsummering før detaljene undersøkes. Streamlit kan plassere fire målekort på samme rad.

  • Gå tilbake til app.py i Visual Studio Code.
  • Finn linjen st.title("RegisterVakt").
  • Fjern den eksisterende visningskoden mellom denne linjen og con.close().
  • Behold con.close() nederst i filen.
  • Lim inn denne oversikten over con.close():
st.title("RegisterVakt")
st.caption(
    "Datakvalitetskontroll av en simulert mottakerkopi basert på åpne data fra Enhetsregisteret."
)
st.caption(
    f"Naiv radtallskontroll: {naive_status}. Kilde: {source_count} rader. "
    f"Mottakerkopi: {consumer_count} rader."
)

metric_source, metric_passed, metric_failed, metric_findings = st.columns(4)
metric_source.metric("Kilderader", source_count, border=True)
metric_passed.metric("Beståtte regler", summary["passed_rules"], border=True)
metric_failed.metric("Feilede regler", summary["failed_rules"], border=True)
metric_findings.metric("Avviksfunn", finding_count, border=True)

What Does This Code Do?

  • Den første bildeteksten beholder den naive radtallskontrollen som kontrast.
  • Variabelen metric_source holder kolonnen for antall kilderader.
  • De tre andre kolonnene viser beståtte regler, feilede regler og antall karantenefunn.
  • Lagre app.py.
  • Gå tilbake til terminalen med det aktiverte .venv-miljøet.
  • Start dashboardet med denne kommandoen:
python -m streamlit run app.py

What Does This Command Do?

Kommandoen starter appen fra app.py gjennom Streamlit-modulen i det aktive miljøet. Nettleseren viser den nyeste lagrede versjonen.

Du skal se PASS for radtallskontrollen. Under den skal du se målekortene Kilderader, Beståtte regler, Feilede regler og Avviksfunn.

Missing the Metric Cards?

Kontroller at målekortkoden står etter spørringene som lager summary og finding_count.

Sjekk at du beholdt con.close() nederst i filen.

Få målrettet hjelp med å finne hvorfor målekortene mangler.

Show dimensions and affected records

Målekortene viser omfanget på toppnivå. Et stolpediagram viser hvor feilene samler seg.

  • Finn spørringen som oppretter quarantine i app.py.
  • Legg inn denne spørringen rett under den avsluttende parentesen:
sample_rows = query_rows(
    con,
    """
    SELECT orgnr, name, parent_orgnr, org_form
    FROM source_subunits
    ORDER BY orgnr
    """,
)

What Does This Query Prepare?

Spørringen henter de normaliserte feltene fra source_subunits. Sorteringen gir kildetabellen en stabil rekkefølge.

  • Gå til con.close() nederst i app.py.
  • Erstatt denne linjen med detaljvisningen nedenfor:
st.write("### Avvik per kvalitetsdimensjon")
st.bar_chart(
    failures_by_dimension,
    x="dimension",
    y="failed_rows",
    x_label="Kvalitetsdimensjon",
    y_label="Antall avviksfunn",
)

st.write("### Kontrollresultater")
st.dataframe(results, width="stretch", hide_index=True)

st.write("### Karantene for oppfølging")
st.dataframe(quarantine, width="stretch", hide_index=True)

st.write("### Kildeutvalg")
st.dataframe(sample_rows, width="stretch", hide_index=True)

st.caption(
    "Datakilde: https://data.brreg.no/enhetsregisteret/api. "
    "Dataene følger Norsk lisens for offentlige data. "
    "Prosjektet bruker ikke data fra Skatteetaten."
)

con.close()

How Does the Detail Layer Help?

  • Stolpediagrammet grupperer failed_rows etter kvalitetsdimensjon.
  • Kontrolltabellen viser regelen, beskrivelsen, omfanget og statusen.
  • Karantenetabellen kobler hvert funn til en regel og en berørt organisasjon.
  • Kildeutvalget viser dataene som mottakerkopien bygger på.
  • Lagre app.py.
  • Start den oppdaterte appen med denne kommandoen:
python -m streamlit run app.py

What Does This Check Prove?

Kommandoen bygger databasen på nytt. Den kjører også de fem reglene før dashboardet tegnes.

Du skal se fem dimensjoner i stolpediagrammet. Kontrolltabellen skal vise fem rader med status FAIL.

Karantenevisningen skal inneholde regel-ID, datasett, organisasjonsnummer og forklaring. Nederst skal du se kildetabellen med orgnr, name, parent_orgnr og org_form.

Dashboard Details Are Missing?

Kontroller at sample_rows opprettes før den brukes av st.dataframe().

Sjekk at quality_checks.sql fortsatt kjøres inne i build_database().

Få hjelp med å feilsøke detaljvisningen.

✔️ Awesome, I've got everything!

Flott. Lagre app.py før du går videre til prosjektdokumentasjonen.

ⓧ I'd like to double check the full code

import json
import os
from urllib.request import Request, urlopen

import duckdb
import streamlit as st

API_ROOT = "https://data.brreg.no/enhetsregisteret/api"
DB_PATH = "data/register_quality.duckdb"

st.set_page_config(page_title="RegisterVakt", layout="wide")


def fetch_json(url):
    request = Request(
        url,
        headers={
            "Accept": "application/json",
            "User-Agent": "RegisterVakt/1.0",
        },
    )
    with urlopen(request, timeout=30) as response:
        return json.load(response)


def query_rows(connection, query):
    result = connection.execute(query)
    columns = [column[0] for column in result.description]
    return [dict(zip(columns, row)) for row in result.fetchall()]


def parent_rows(parent_payloads):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in parent_payloads
    ]


def subunit_rows(subunits):
    return [
        {
            "orgnr": item["organisasjonsnummer"],
            "name": item["navn"],
            "parent_orgnr": item["overordnetEnhet"],
            "org_form": item.get("organisasjonsform", {}).get("kode", ""),
            "age_hours": 0,
        }
        for item in subunits
    ]


def create_tables(connection, table_name, rows, is_subunit):
    if is_subunit:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                parent_orgnr VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["parent_orgnr"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )
    else:
        connection.execute(
            f"""
            CREATE OR REPLACE TABLE {table_name} (
                orgnr VARCHAR,
                name VARCHAR,
                org_form VARCHAR,
                age_hours INTEGER
            )
            """
        )
        connection.executemany(
            f"INSERT INTO {table_name} VALUES (?, ?, ?, ?)",
            [
                (
                    row["orgnr"],
                    row["name"],
                    row["org_form"],
                    row["age_hours"],
                )
                for row in rows
            ],
        )


def build_database():
    os.makedirs("data", exist_ok=True)

    subunit_payload = fetch_json(f"{API_ROOT}/underenheter")
    source_subunits = subunit_payload.get("_embedded", {}).get("underenheter", [])

    if len(source_subunits) < 5:
        raise RuntimeError("API-et returnerte for få underenheter til prosjektets kontroller.")

    parent_orgnrs = sorted(
        {item["overordnetEnhet"] for item in source_subunits}
    )
    parent_payloads = [
        fetch_json(f"{API_ROOT}/enheter/{orgnr}") for orgnr in parent_orgnrs
    ]

    source_parents = parent_rows(parent_payloads)
    source_subunits = subunit_rows(source_subunits)

    consumer_parents = [dict(row) for row in source_parents]
    consumer_subunits = [dict(row) for row in source_subunits]

    for row in consumer_parents:
        row["age_hours"] = 4
    for row in consumer_subunits:
        row["age_hours"] = 4

    consumer_parents[0]["name"] = ""
    consumer_subunits[-1]["orgnr"] = consumer_subunits[0]["orgnr"]
    consumer_subunits[1]["orgnr"] = "12345"
    consumer_subunits[2]["parent_orgnr"] = "000000000"
    consumer_subunits[3]["age_hours"] = 72

    connection = duckdb.connect(database=DB_PATH)
    create_tables(connection, "source_parents", source_parents, False)
    create_tables(connection, "source_subunits", source_subunits, True)
    create_tables(connection, "consumer_parents", consumer_parents, False)
    create_tables(connection, "consumer_subunits", consumer_subunits, True)

    with open("quality_checks.sql", encoding="utf-8") as sql_file:
        connection.execute(sql_file.read())

    return connection


con = build_database()

source_count = con.execute("SELECT COUNT(*) FROM source_subunits").fetchone()[0]
consumer_count = con.execute("SELECT COUNT(*) FROM consumer_subunits").fetchone()[0]
naive_status = "PASS" if source_count == consumer_count else "FAIL"

summary = query_rows(
    con,
    """
    SELECT
        COUNT(*) AS total_rules,
        SUM(CASE WHEN status = 'PASS' THEN 1 ELSE 0 END) AS passed_rules,
        SUM(CASE WHEN status = 'FAIL' THEN 1 ELSE 0 END) AS failed_rules
    FROM quality_results
    """,
)[0]

finding_count = con.execute(
    "SELECT COUNT(*) FROM quarantined_records"
).fetchone()[0]

results = query_rows(
    con,
    """
    SELECT rule_id, dimension, rule_description, failed_rows, total_rows, status
    FROM quality_results
    ORDER BY rule_id
    """,
)

failures_by_dimension = query_rows(
    con,
    """
    SELECT dimension, SUM(failed_rows) AS failed_rows
    FROM quality_results
    GROUP BY dimension
    ORDER BY dimension
    """,
)

quarantine = query_rows(
    con,
    """
    SELECT rule_id, dataset, orgnr, issue
    FROM quarantined_records
    ORDER BY rule_id, orgnr
    """,
)

sample_rows = query_rows(
    con,
    """
    SELECT orgnr, name, parent_orgnr, org_form
    FROM source_subunits
    ORDER BY orgnr
    """,
)

st.title("RegisterVakt")
st.caption(
    "Datakvalitetskontroll av en simulert mottakerkopi basert på åpne data fra Enhetsregisteret."
)
st.caption(
    f"Naiv radtallskontroll: {naive_status}. Kilde: {source_count} rader. "
    f"Mottakerkopi: {consumer_count} rader."
)

metric_source, metric_passed, metric_failed, metric_findings = st.columns(4)
metric_source.metric("Kilderader", source_count, border=True)
metric_passed.metric("Beståtte regler", summary["passed_rules"], border=True)
metric_failed.metric("Feilede regler", summary["failed_rules"], border=True)
metric_findings.metric("Avviksfunn", finding_count, border=True)

st.write("### Avvik per kvalitetsdimensjon")
st.bar_chart(
    failures_by_dimension,
    x="dimension",
    y="failed_rows",
    x_label="Kvalitetsdimensjon",
    y_label="Antall avviksfunn",
)

st.write("### Kontrollresultater")
st.dataframe(results, width="stretch", hide_index=True)

st.write("### Karantene for oppfølging")
st.dataframe(quarantine, width="stretch", hide_index=True)

st.write("### Kildeutvalg")
st.dataframe(sample_rows, width="stretch", hide_index=True)

st.caption(
    "Datakilde: https://data.brreg.no/enhetsregisteret/api. "
    "Dataene følger Norsk lisens for offentlige data. "
    "Prosjektet bruker ikke data fra Skatteetaten."
)

con.close()

Denne referansen viser hele den ferdige app.py-filen. Sammenlign funksjoner, spørringer og visningsrekkefølge med din fil.

Document and save the finished project

Et dashboard trenger dokumentasjon som forklarer datakilden og avgrensningen. README.md gjør prosjektet forståelig uten at leseren må åpne kildekoden først.

  • Velg ikonet for en ny fil i filpanelet i Visual Studio Code.
  • Opprett README.md ved siden av app.py.
  • Legg inn denne prosjektbeskrivelsen og kontrolloversikten:
# RegisterVakt

RegisterVakt er et lokalt porteføljeprosjekt for analyse av datakvalitet i registerlignende leveranser. Prosjektet henter åpne virksomhetsdata fra Brønnøysundregistrene, oppretter en simulert mottakerkopi, kjører SQL-kontroller og viser resultatene i Streamlit.

Prosjektet bruker ikke data fra Skatteetaten og inneholder ikke private personopplysninger.

## Kontroller

- manglende navn
- dupliserte organisasjonsnumre
- organisasjonsnumre som ikke består av ni siffer
- underenheter uten kjent hovedenhet
- leveranser som er eldre enn 24 timer

What Does This Section Explain?

Åpningen beskriver prosjektets formål og avgrensning. Kontrollisten forteller hvilke typer avvik dashboardet måler.

  • Lagre README.md.
  • Bekreft at README.md vises ved siden av de andre prosjektfilene i filpanelet.
  • Legg inn disse delene under kontrollisten:
## Kjør lokalt

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python -m streamlit run app.py
```

## Datakilde

- API: https://data.brreg.no/enhetsregisteret/api
- Dokumentasjon: https://data.brreg.no/enhetsregisteret/api/dokumentasjon/no/index.html
- Lisens: Norsk lisens for offentlige data

Why Include Run Instructions?

Kjøreseksjonen gjør miljøet reproduserbart. Datakildeseksjonen viser hvor registerdataene kommer fra.

  • Lagre README.md.
  • Kontroller at kodegjerdet inneholder fire kommandoer.
  • Legg inn dette CV-punktet nederst i filen:
## Forslag til CV-punkt

Utviklet et Python- og SQL-basert dashboard for datakvalitet i registerdata. Implementerte kontroller for fullstendighet, dubletter, identifikatorformat, referanseintegritet og aktualitet, med avvikskarantene og visualisering for dataeiere.

What Does the CV Point Capture?

CV-punktet oppsummerer den tekniske løsningen og de fem kvalitetsområdene. Det fremhever også karantenen som gjør funnene sporbare.

  • Lagre den ferdige README.md-filen.

✔️ Awesome, I've got everything!

Bra. Dokumentasjonen dekker datakilden, kontrollene, lokal kjøring og et kort CV-punkt.

ⓧ I'd like to double check the full code

# RegisterVakt

RegisterVakt er et lokalt porteføljeprosjekt for analyse av datakvalitet i registerlignende leveranser. Prosjektet henter åpne virksomhetsdata fra Brønnøysundregistrene, oppretter en simulert mottakerkopi, kjører SQL-kontroller og viser resultatene i Streamlit.

Prosjektet bruker ikke data fra Skatteetaten og inneholder ikke private personopplysninger.

## Kontroller

- manglende navn
- dupliserte organisasjonsnumre
- organisasjonsnumre som ikke består av ni siffer
- underenheter uten kjent hovedenhet
- leveranser som er eldre enn 24 timer

## Kjør lokalt

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python -m streamlit run app.py
```

## Datakilde

- API: https://data.brreg.no/enhetsregisteret/api
- Dokumentasjon: https://data.brreg.no/enhetsregisteret/api/dokumentasjon/no/index.html
- Lisens: Norsk lisens for offentlige data

## Forslag til CV-punkt

Utviklet et Python- og SQL-basert dashboard for datakvalitet i registerdata. Implementerte kontroller for fullstendighet, dubletter, identifikatorformat, referanseintegritet og aktualitet, med avvikskarantene og visualisering for dataeiere.

Denne referansen matcher den ferdige README.md-filen. Behold overskriftene og teksten i samme rekkefølge.

Et lokalt Git-commit lagrer den ferdige tilstanden som et sporbar øyeblikksbilde. Filene i data og det virtuelle miljøet forblir utelatt gjennom .gitignore.

  • Velg Source Control i aktivitetslinjen i Visual Studio Code.
  • Velg Initialize Repository.
  • Hold pekeren over Changes.
  • Velg Stage All Changes.
  • Skriv Complete RegisterVakt dashboard i feltet for commit-meldingen.
  • Velg Commit.
  • Utvid Source Control Graph.
  • Velg commiten Complete RegisterVakt dashboard for å kontrollere innholdet.

Commit-visningen skal inneholde .gitignore, requirements.txt, app.py, quality_checks.sql og README.md.

Cannot Create the Commit?

Kontroller at alle prosjektfilene står under Staged Changes før du velger Commit.

Hvis Git ber om en identitet, følger du meldingen i Visual Studio Code før du prøver igjen.

Få hjelp med å fullføre det lokale Git-commitet.

Før du kjører den siste kontrollen, tror du den naive radtallskontrollen vil være enig med de fem kvalitetsreglene?

  • Start den ferdige RegisterVakt-appen med denne kommandoen:
python -m streamlit run app.py

What Does the Final Run Verify?

Den siste kjøringen henter registerutvalget og gjenoppbygger mottakerkopien. Deretter oppdaterer den alle visningene fra DuckDB-resultatene.

Nettleseren skal vise PASS for den naive radtallskontrollen. De fire målekortene skal vises på samme rad.

Stolpediagrammet skal dekke fem dimensjoner. Under diagrammet skal du se kontrollresultatene, karantenefunnene og kildeutvalget.

Final Dashboard Check Failed?

Kontroller at terminalen fortsatt bruker .venv.

Sjekk at app.py og quality_checks.sql ligger i samme prosjektmappe.

Få hjelp med å feilsøke den ferdige RegisterVakt-appen.

Sterkt jobbet. RegisterVakt gjør nå avvikene synlige på en måte en dataeier kan bruke til prioritering.

Secret mission

Detect Quality Regression Between Deliveries

Sammenlign kilden og mottakerkopien med de samme fem kvalitetsreglene. Vis hvilke dimensjoner som har fått flere avvik i den nye leveransen.

Clean Up Your Resources

Clean Up Your Resources

Velg om du vil beholde ressursene i bruk, stoppe dashboardet midlertidig eller slette prosjektet. Alt kjører lokalt på maskinen din, så det finnes ingen løpende kostnader.

Ressurser du brukte:

  • Den aktive Streamlit-prosessen som viser dashboardet i nettleseren.
  • Det lokale Git-arbeidsområdet for RegisterVakt.
  • Det virtuelle Python-miljøet i .venv.
  • Den lokale DuckDB-databasen i data/register_quality.duckdb.

Keep everything running

Ingen handling er nødvendig. Velg dette hvis du vil fortsette å bruke RegisterVakt nå.

  • La Streamlit-prosessen kjøre mens du bruker dashboardet.
  • Behold hele RegisterVakt-arbeidsområdet.
  • Behold data/register_quality.duckdb for å ta vare på de lokale tabellene.

Dashboardet forblir tilgjengelig i nettleseren. Arbeidsområdet er klart for flere kvalitetsregler.

Pause - I'll come back to this later

Denne varianten frigjør ressursene som appen bruker mens den kjører. Alle prosjektfilene blir liggende på maskinen din.

  • Gå tilbake til terminalen som kjører Streamlit.
  • Trykk Ctrl+C for å stoppe dashboardet.
  • Behold RegisterVakt-arbeidsområdet på maskinen din.

Terminalprompten vises igjen når prosessen har stoppet. Nettleseren kan ikke lenger hente dashboardet fra den lokale appen.

Delete - I don't want to use this again

Denne varianten fjerner hele RegisterVakt-arbeidsområdet fra maskinen din. Slettingen kan ikke angres, så bruk den bare når du ikke trenger de lokale filene.

  • Gå tilbake til terminalen som kjører Streamlit.
  • Trykk Ctrl+C for å stoppe appprosessen.
  • Registrer navnet på den øverste prosjektmappen i Visual Studio Code som navnet på RegisterVakt-prosjektmappen.
  • Slett hele prosjektmappen fra terminalen ved å kjøre disse kommandoene:
cd ..
rm -rf "[[PROJECT_FOLDER="navnet på RegisterVakt-prosjektmappen"]]"

Hva Gjør Kommandoene?

  • Den første linjen flytter terminalen til mappen som inneholder RegisterVakt-arbeidsområdet.
  • Den andre linjen fjerner den valgte prosjektmappen med alle lokale prosjektressurser.

Terminalprompten vises igjen når slettingen er ferdig. RegisterVakt-arbeidsområdet finnes ikke lenger på maskinen.

Ble Ikke Mappen Slettet?

  • Kontroller at mappenavnet du registrerte samsvarer nøyaktig med navnet i Visual Studio Code.
  • Kontroller at terminalen står i mappen over RegisterVakt-arbeidsområdet før du prøver igjen.

Hjelp meg å kontrollere hvorfor RegisterVakt-mappen fortsatt finnes etter den lokale slettekommandoen.

Nice Work!

Nice Work!

Du klarte det! RegisterVakt er nå et kjørbart datakvalitetsdashboard som gjør registeravvik synlige for en dataeier.

Dette har du lært:

  • Bygge en sporbar registerleveranse fra åpne data hos Brønnøysundregistrene til normaliserte tabeller i DuckDB.
  • Avsløre falsk trygghet i radtall med fem gjenbrukbare SQL-kontroller. Isolere hvert avvik i en sporbar karantene.
  • Formidle prioriteringsklare avvik gjennom et Streamlit-dashboard. Vise omfanget gjennom nøkkeltall. Peke ut berørte rader i karantenetabellen.
  • Secret Mission: Måle kvalitetsregresjon fra kildeleveransen til mottakerkopien. Flagge forverrede dimensjoner med REGRESJON.

Klar for å teste deg selv? Se hva du husker.