Data Modeling: Star Schema, Snowflake and Data Vault

import shutil
import sqlite3
import urllib.request
from pathlib import Path

url = "https://github.com/lerocha/chinook-database/releases/download/v1.4.5/Chinook_Sqlite.sqlite"
if not Path("chinook.sqlite").exists():
    urllib.request.urlretrieve(url, "chinook.sqlite")
shutil.copy("chinook.sqlite", "work.sqlite")
con = sqlite3.connect("work.sqlite")

con.executescript("""
-- star: one fact table, flat dimensions
CREATE TABLE star_fact_sales AS
  SELECT il.InvoiceLineId, i.CustomerId AS customer_key, il.TrackId AS track_key, date(i.InvoiceDate) AS date_key,
         il.Quantity AS quantity, il.UnitPrice * il.Quantity AS amount
  FROM InvoiceLine il JOIN Invoice i USING (InvoiceId);
CREATE TABLE star_dim_track AS
  SELECT t.TrackId AS track_key, t.Name AS track, al.Title AS album, ar.Name AS artist, g.Name AS genre
  FROM Track t JOIN Album al USING (AlbumId) JOIN Artist ar USING (ArtistId) JOIN Genre g USING (GenreId);
CREATE TABLE star_dim_customer AS SELECT CustomerId AS customer_key, Country AS country FROM Customer;

-- snowflake: the same fact table, but the track dimension's hierarchy is normalised
CREATE TABLE snow_dim_track AS SELECT TrackId AS track_key, Name AS track, AlbumId AS album_key, GenreId AS genre_key FROM Track;
CREATE TABLE snow_dim_album AS SELECT AlbumId AS album_key, Title AS album, ArtistId AS artist_key FROM Album;
CREATE TABLE snow_dim_artist AS SELECT ArtistId AS artist_key, Name AS artist FROM Artist;
CREATE TABLE snow_dim_genre AS SELECT GenreId AS genre_key, Name AS genre FROM Genre;

-- data vault: hubs hold business keys, links hold relationships, satellites hold descriptive data
CREATE TABLE hub_customer AS SELECT CustomerId AS customer_bk, '2013-12-31' AS load_date, 'chinook' AS source FROM Customer;
CREATE TABLE hub_track    AS SELECT TrackId AS track_bk, '2013-12-31' AS load_date, 'chinook' AS source FROM Track;
CREATE TABLE hub_genre    AS SELECT GenreId AS genre_bk, '2013-12-31' AS load_date, 'chinook' AS source FROM Genre;
CREATE TABLE hub_invoice  AS SELECT InvoiceId AS invoice_bk, '2013-12-31' AS load_date, 'chinook' AS source FROM Invoice;
CREATE TABLE sat_customer AS SELECT CustomerId AS customer_bk, Country AS country, '2013-12-31' AS load_date FROM Customer;
CREATE TABLE sat_genre    AS SELECT GenreId AS genre_bk, Name AS genre, '2013-12-31' AS load_date FROM Genre;
CREATE TABLE sat_invoice  AS SELECT InvoiceId AS invoice_bk, date(InvoiceDate) AS invoice_date, '2013-12-31' AS load_date FROM Invoice;
CREATE TABLE link_track_genre      AS SELECT TrackId AS track_bk, GenreId AS genre_bk FROM Track;
CREATE TABLE link_invoice_customer AS SELECT InvoiceId AS invoice_bk, CustomerId AS customer_bk FROM Invoice;
CREATE TABLE link_invoice_track    AS SELECT InvoiceLineId AS line_bk, InvoiceId AS invoice_bk, TrackId AS track_bk FROM InvoiceLine;
CREATE TABLE sat_invoice_track     AS SELECT InvoiceLineId AS line_bk, Quantity AS quantity, UnitPrice * Quantity AS amount FROM InvoiceLine;
""")

SELECT = "SELECT genre, country, ROUND(SUM(amount), 2) AS revenue FROM "
TAIL = " GROUP BY genre, country ORDER BY revenue DESC, genre, country LIMIT 3"
queries = {
    "star": SELECT + """star_fact_sales f JOIN star_dim_track t USING (track_key) JOIN star_dim_customer c USING (customer_key)""" + TAIL,
    "snowflake": SELECT + """star_fact_sales f JOIN snow_dim_track t USING (track_key) JOIN snow_dim_genre g USING (genre_key)
                             JOIN star_dim_customer c USING (customer_key)""" + TAIL,
    "data vault": SELECT + """link_invoice_track l JOIN sat_invoice_track s USING (line_bk)
                              JOIN link_track_genre tg USING (track_bk) JOIN sat_genre g USING (genre_bk)
                              JOIN link_invoice_customer ic USING (invoice_bk) JOIN sat_customer c USING (customer_bk)""" + TAIL,
}
tables = [r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type = 'table'")]
count = lambda prefixes: sum(t.startswith(prefixes) for t in tables)
print("tables: star", count("star_"), "| snowflake", count(("star_fact", "star_dim_customer", "snow_")),
      "| data vault", count(("hub_", "link_", "sat_")))
results = {name: con.execute(sql).fetchall() for name, sql in queries.items()}
for name, sql in queries.items():
    print(f"{name:<11} joins in the query: {sql.count('JOIN')}")
print("top rows:", results["star"])
print("all three agree:", results["star"] == results["snowflake"] == results["data vault"])
con.close()

tables: star 3 | snowflake 6 | data vault 11
star        joins in the query: 2
snowflake   joins in the query: 3
data vault  joins in the query: 5
top rows: [('Rock', 'USA', 155.43), ('Rock', 'Canada', 105.93), ('Latin', 'USA', 90.09)]
all three agree: True