A complete eight-stage procedure built with Python and SQL Server. Each stage explains why it exists, how it works, presents the complete validated script, and closes with the result achieved.
Procedimiento completo de ocho etapas construido con Python y SQL Server. Cada etapa explica por qué existe, cómo funciona, presenta el script validado completo y cierra con el resultado alcanzado.
This implementation demonstrates a reusable path for transforming a large public CSV into an auditable analytical platform. The workflow preserves the raw source, profiles the real data, measures quality, creates a normalized layer, builds a dimensional warehouse, publishes business views, and generates the documentation required to implement the Power BI model.
Esta implementación demuestra una ruta reutilizable para transformar un CSV público de gran tamaño en una plataforma analítica auditable. El flujo preserva el origen raw, perfila los datos reales, mide la calidad, crea una capa normalizada, construye un warehouse dimensional, publica vistas de negocio y genera la documentación necesaria para implementar el modelo Power BI.
01_AKPE_SQL_Loader.py
The first stage preserves the original dataset before any business transformation. A raw staging table provides traceability, repeatability, and a stable source for every later module.
La primera etapa preserva el dataset original antes de cualquier transformación de negocio. Una tabla raw ofrece trazabilidad, repetibilidad y una fuente estable para todos los módulos posteriores.
Python creates the database and schemas, reads the CSV in 5,000-row chunks, inserts every source field as text, adds AKPE technical metadata, and records the execution in an audit table.
Python crea la base y los schemas, lee el CSV en bloques de 5,000 filas, inserta cada campo de origen como texto, añade metadata técnica de AKPE y registra la ejecución en una tabla de auditoría.
from __future__ import annotations
import csv
import sys
import time
import uuid
from pathlib import Path
import pandas as pd
import pyodbc
# ============================================================
# AKPE MODULE 01
# SQL SERVER CSV LOADER
# ============================================================
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
CSV_FILE = Path(r"C:\AKPE\data\Fire_Inspections.csv")
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
CHUNK_SIZE = 5_000
RAW_SCHEMA = "staging"
RAW_TABLE = "FireInspections_Raw"
AUDIT_SCHEMA = "audit"
AUDIT_TABLE = "LoadLog"
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def master_connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
"DATABASE=master;"
"Trusted_Connection=yes;"
)
def database_connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
)
def create_database_if_needed() -> None:
with pyodbc.connect(master_connection_string(), autocommit=True) as connection:
cursor = connection.cursor()
cursor.execute(
f"""
IF DB_ID(?) IS NULL
BEGIN
EXEC(N'CREATE DATABASE {quote_identifier(DATABASE_NAME)}');
END;
""",
DATABASE_NAME,
)
def create_schemas_and_audit_table(connection: pyodbc.Connection) -> None:
cursor = connection.cursor()
cursor.execute(
f"""
IF SCHEMA_ID(N'{RAW_SCHEMA}') IS NULL
EXEC(N'CREATE SCHEMA {quote_identifier(RAW_SCHEMA)}');
IF SCHEMA_ID(N'{AUDIT_SCHEMA}') IS NULL
EXEC(N'CREATE SCHEMA {quote_identifier(AUDIT_SCHEMA)}');
"""
)
cursor.execute(
f"""
IF OBJECT_ID(N'{AUDIT_SCHEMA}.{AUDIT_TABLE}', N'U') IS NULL
BEGIN
CREATE TABLE {quote_identifier(AUDIT_SCHEMA)}.{quote_identifier(AUDIT_TABLE)}
(
LoadLogId BIGINT IDENTITY(1,1) PRIMARY KEY,
BatchId UNIQUEIDENTIFIER NOT NULL,
SourceFile NVARCHAR(1000) NOT NULL,
TargetTable NVARCHAR(300) NOT NULL,
StartedAt DATETIME2(0) NOT NULL,
CompletedAt DATETIME2(0) NULL,
RowsLoaded BIGINT NULL,
Status NVARCHAR(30) NOT NULL,
ErrorMessage NVARCHAR(MAX) NULL
);
END;
"""
)
connection.commit()
def read_csv_headers(csv_file: Path) -> list[str]:
with csv_file.open("r", encoding="utf-8-sig", newline="") as handle:
reader = csv.reader(handle)
return next(reader)
def create_raw_table(
connection: pyodbc.Connection,
source_columns: list[str],
) -> None:
cursor = connection.cursor()
cursor.execute(
f"""
IF OBJECT_ID(
N'{RAW_SCHEMA}.{RAW_TABLE}',
N'U'
) IS NOT NULL
DROP TABLE {quote_identifier(RAW_SCHEMA)}.{quote_identifier(RAW_TABLE)};
"""
)
source_definition = ",\n".join(
f" {quote_identifier(column_name)} NVARCHAR(MAX) NULL"
for column_name in source_columns
)
create_sql = f"""
CREATE TABLE {quote_identifier(RAW_SCHEMA)}.{quote_identifier(RAW_TABLE)}
(
AKPE_RowId BIGINT IDENTITY(1,1) NOT NULL PRIMARY KEY,
{source_definition},
AKPE_BatchId UNIQUEIDENTIFIER NOT NULL,
AKPE_LoadedAt DATETIME2(0) NOT NULL
CONSTRAINT DF_{RAW_TABLE}_LoadedAt DEFAULT SYSUTCDATETIME()
);
"""
cursor.execute(create_sql)
connection.commit()
def start_audit(
connection: pyodbc.Connection,
batch_id: uuid.UUID,
) -> int:
cursor = connection.cursor()
cursor.execute(
f"""
INSERT INTO {quote_identifier(AUDIT_SCHEMA)}.{quote_identifier(AUDIT_TABLE)}
(
BatchId,
SourceFile,
TargetTable,
StartedAt,
Status
)
OUTPUT INSERTED.LoadLogId
VALUES
(
?,
?,
?,
SYSUTCDATETIME(),
N'Running'
);
""",
str(batch_id),
str(CSV_FILE),
f"{RAW_SCHEMA}.{RAW_TABLE}",
)
load_log_id = int(cursor.fetchone()[0])
connection.commit()
return load_log_id
def finish_audit(
connection: pyodbc.Connection,
load_log_id: int,
rows_loaded: int,
status: str,
error_message: str | None = None,
) -> None:
cursor = connection.cursor()
cursor.execute(
f"""
UPDATE {quote_identifier(AUDIT_SCHEMA)}.{quote_identifier(AUDIT_TABLE)}
SET
CompletedAt = SYSUTCDATETIME(),
RowsLoaded = ?,
Status = ?,
ErrorMessage = ?
WHERE LoadLogId = ?;
""",
rows_loaded,
status,
error_message,
load_log_id,
)
connection.commit()
def normalize_chunk(chunk: pd.DataFrame) -> pd.DataFrame:
normalized = chunk.copy()
normalized = normalized.where(pd.notna(normalized), None)
for column_name in normalized.columns:
normalized[column_name] = normalized[column_name].map(
lambda value: None if value is None else str(value)
)
return normalized
def load_csv(
connection: pyodbc.Connection,
source_columns: list[str],
batch_id: uuid.UUID,
) -> int:
target_columns = source_columns + ["AKPE_BatchId"]
insert_columns = ", ".join(
quote_identifier(column_name)
for column_name in target_columns
)
placeholders = ", ".join("?" for _ in target_columns)
insert_sql = f"""
INSERT INTO {quote_identifier(RAW_SCHEMA)}.{quote_identifier(RAW_TABLE)}
({insert_columns})
VALUES ({placeholders});
"""
rows_loaded = 0
cursor = connection.cursor()
cursor.fast_executemany = True
for chunk_number, chunk in enumerate(
pd.read_csv(
CSV_FILE,
dtype=str,
chunksize=CHUNK_SIZE,
encoding="utf-8-sig",
low_memory=False,
),
start=1,
):
chunk = normalize_chunk(chunk)
chunk["AKPE_BatchId"] = str(batch_id)
cursor.executemany(
insert_sql,
chunk.itertuples(index=False, name=None),
)
connection.commit()
rows_loaded += len(chunk)
print(
f"Chunk {chunk_number:,} completed | "
f"Rows loaded: {rows_loaded:,}"
)
return rows_loaded
def validate_load(connection: pyodbc.Connection) -> tuple[int, int]:
cursor = connection.cursor()
cursor.execute(
f"""
SELECT
COUNT_BIG(*) AS TotalRows,
MAX(AKPE_RowId) AS MaximumRowId
FROM {quote_identifier(RAW_SCHEMA)}.{quote_identifier(RAW_TABLE)};
"""
)
row = cursor.fetchone()
return int(row[0]), int(row[1])
def main() -> None:
started = time.perf_counter()
if not CSV_FILE.exists():
raise FileNotFoundError(f"CSV file not found: {CSV_FILE}")
batch_id = uuid.uuid4()
rows_loaded = 0
load_log_id: int | None = None
create_database_if_needed()
with pyodbc.connect(database_connection_string()) as connection:
create_schemas_and_audit_table(connection)
source_columns = read_csv_headers(CSV_FILE)
create_raw_table(connection, source_columns)
load_log_id = start_audit(connection, batch_id)
try:
rows_loaded = load_csv(
connection,
source_columns,
batch_id,
)
total_rows, maximum_row_id = validate_load(connection)
if total_rows != rows_loaded:
raise RuntimeError(
"Validation failed: loaded-row count does not match "
"the SQL Server row count."
)
finish_audit(
connection,
load_log_id,
rows_loaded,
"Completed",
)
elapsed = time.perf_counter() - started
print("\nPROCESS COMPLETED SUCCESSFULLY")
print(f"Database: {DATABASE_NAME}")
print(f"Target: {RAW_SCHEMA}.{RAW_TABLE}")
print(f"Rows transferred: {rows_loaded:,}")
print(f"Maximum AKPE_RowId: {maximum_row_id:,}")
print(f"Elapsed seconds: {elapsed:,.2f}")
except Exception as exc:
if load_log_id is not None:
finish_audit(
connection,
load_log_id,
rows_loaded,
"Failed",
str(exc),
)
raise
if __name__ == "__main__":
try:
main()
except Exception as exc:
print(f"\nPROCESS FAILED\n{exc}", file=sys.stderr)
sys.exit(1)
Validated result: 439,581 records loaded into staging.FireInspections_Raw. The process also created audit.LoadLog, preserved the original column values, and assigned a sequential AKPE_RowId.
Resultado validado: 439,581 registros cargados en staging.FireInspections_Raw. El proceso también creó audit.LoadLog, preservó los valores originales y asignó un AKPE_RowId consecutivo.
02_AKPE_SQL_Dataset_Profiler.py
A warehouse should not be designed from column names alone. Profiling reveals actual completeness, cardinality, numeric and date behavior, frequent values, and the analytical role of each field.
Un Data Warehouse no debe diseñarse solamente a partir de los nombres de columnas. El profiling revela completitud, cardinalidad, comportamiento numérico y temporal, valores frecuentes y el rol analítico de cada campo.
The profiler reads a controlled SQL sample into pandas and calculates column-level statistics. It then classifies fields as geographic, temporal, categorical, measures, identifiers, text, or technical metadata and exports the findings.
El profiler lee una muestra controlada desde SQL hacia pandas y calcula estadísticas por columna. Después clasifica los campos como geográficos, temporales, categóricos, medidas, identificadores, texto o metadata técnica y exporta los hallazgos.
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Any
import json
import re
import sys
import pandas as pd
import pyodbc
# ============================================================
# CONFIGURATION
# ============================================================
SCRIPT_FOLDER = Path(__file__).resolve().parent
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
SOURCE_SCHEMA = "staging"
SOURCE_TABLE = "FireInspections_Raw"
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
SAMPLE_SIZE = 100_000
TOP_VALUES_LIMIT = 10
OUTPUT_FOLDER = SCRIPT_FOLDER / "sql_dataset_profile_v2"
# ============================================================
# KEYWORDS
# ============================================================
TEMPORAL_KEYWORDS = [
"date", "time", "datetime", "timestamp", "year", "month",
"day", "week", "quarter", "created", "updated", "loaded",
"issued", "expired", "scheduled", "completed", "paid",
"return", "notice", "lien",
]
GEOGRAPHIC_KEYWORDS = [
"address", "street", "city", "state", "county", "zipcode",
"zip code", "zip", "postal", "neighborhood", "district",
"battalion", "station", "latitude", "longitude", "location",
"boundary", "parcel", "block", "lot",
]
CATEGORY_KEYWORDS = [
"type", "status", "category", "class", "classification",
"description", "agency", "department", "division", "program",
"service", "billable", "priority", "risk", "reason",
"source", "method", "zone",
]
IDENTIFIER_KEYWORDS = [
"id", "number", "identifier", "key", "uuid", "permit",
"complaint", "referral", "violation", "application",
]
MEASURE_KEYWORDS = [
"amount", "fee", "cost", "price", "total", "count",
"quantity", "duration", "score", "balance", "interest",
"penalty", "payment",
]
TECHNICAL_KEYWORDS = [
"akpe", "load batch", "source file", "data as of",
"data loaded", "etl", "row id", "geometry", "polygon", "shape",
]
# ============================================================
# HELPERS
# ============================================================
def print_header(title: str) -> None:
print("\n" + "=" * 88)
print(title)
print("=" * 88)
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def normalize_name(value: str) -> str:
value = value.lower().strip()
value = re.sub(r"[_\-]+", " ", value)
value = re.sub(r"\s+", " ", value)
return value
def contains_keyword(column_name: str, keywords: list[str]) -> bool:
normalized = normalize_name(column_name)
return any(keyword in normalized for keyword in keywords)
def connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
"TrustServerCertificate=yes;"
)
def connect_to_sql() -> pyodbc.Connection:
return pyodbc.connect(
connection_string(),
autocommit=True,
timeout=30,
)
# ============================================================
# SOURCE VALIDATION
# ============================================================
def validate_environment() -> None:
if ODBC_DRIVER not in pyodbc.drivers():
raise RuntimeError(
f"ODBC driver not found: {ODBC_DRIVER}\n"
f"Installed drivers: {pyodbc.drivers()}"
)
def validate_source_table(connection: pyodbc.Connection) -> None:
cursor = connection.cursor()
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
SOURCE_SCHEMA,
SOURCE_TABLE,
)
if cursor.fetchone()[0] != 1:
raise RuntimeError(
f"Table not found: "
f"{DATABASE_NAME}.{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
# ============================================================
# SQL READ
# ============================================================
def get_total_rows(connection: pyodbc.Connection) -> int:
full_table = (
f"{quote_identifier(SOURCE_SCHEMA)}."
f"{quote_identifier(SOURCE_TABLE)}"
)
cursor = connection.cursor()
cursor.execute(f"SELECT COUNT_BIG(*) FROM {full_table};")
return int(cursor.fetchone()[0])
def read_sample_to_dataframe(
connection: pyodbc.Connection,
sample_size: int,
) -> pd.DataFrame:
full_table = (
f"{quote_identifier(SOURCE_SCHEMA)}."
f"{quote_identifier(SOURCE_TABLE)}"
)
sql = f"""
SELECT TOP ({sample_size}) *
FROM {full_table}
ORDER BY [AKPE_RowId];
"""
cursor = connection.cursor()
cursor.execute(sql)
columns = [column[0] for column in cursor.description]
rows = cursor.fetchall()
dataframe = pd.DataFrame.from_records(
rows,
columns=columns,
)
cursor.close()
return dataframe
# ============================================================
# TYPE DETECTION
# ============================================================
def clean_text_series(series: pd.Series) -> pd.Series:
text = series.astype("string").str.strip()
return text.replace(
{
"": pd.NA,
"nan": pd.NA,
"NaN": pd.NA,
"None": pd.NA,
"<NA>": pd.NA,
}
)
def numeric_conversion_ratio(series: pd.Series) -> float:
cleaned = (
clean_text_series(series)
.str.replace("$", "", regex=False)
.str.replace(",", "", regex=False)
)
non_null = cleaned.dropna()
if non_null.empty:
return 0.0
converted = pd.to_numeric(non_null, errors="coerce")
return round(float(converted.notna().mean()), 4)
def integer_conversion_ratio(series: pd.Series) -> float:
cleaned = (
clean_text_series(series)
.str.replace(",", "", regex=False)
)
non_null = cleaned.dropna()
if non_null.empty:
return 0.0
converted = pd.to_numeric(non_null, errors="coerce")
valid = converted.notna() & (converted % 1 == 0)
return round(float(valid.mean()), 4)
def date_conversion_ratio(
series: pd.Series,
column_name: str,
) -> float:
if not contains_keyword(column_name, TEMPORAL_KEYWORDS):
return 0.0
cleaned = clean_text_series(series)
non_null = cleaned.dropna()
if non_null.empty:
return 0.0
converted = pd.to_datetime(
non_null,
errors="coerce",
format="mixed",
)
return round(float(converted.notna().mean()), 4)
# ============================================================
# CLASSIFICATION
# ============================================================
def classify_column(
column_name: str,
distinct_count: int,
unique_ratio: float,
null_percentage: float,
date_ratio: float,
numeric_ratio: float,
) -> tuple[str, str, int]:
normalized = normalize_name(column_name)
if contains_keyword(column_name, TECHNICAL_KEYWORDS):
return (
"Technical Metadata",
"ETL, source, load, audit, or spatial metadata.",
98,
)
if contains_keyword(column_name, TEMPORAL_KEYWORDS):
if date_ratio >= 0.90:
return (
"Temporal Dimension Candidate",
"Temporal name and strong date-conversion success.",
98,
)
return (
"Possible Temporal Dimension",
"Temporal name detected, but values require cleaning or review.",
80,
)
if contains_keyword(column_name, GEOGRAPHIC_KEYWORDS):
if "polygon" in normalized or "geometry" in normalized:
return (
"Spatial Metadata",
"Spatial geometry field.",
96,
)
if "latitude" in normalized or "longitude" in normalized:
return (
"Geographic Measure",
"Coordinate suitable for mapping.",
96,
)
return (
"Geographic Dimension Candidate",
"Geographic or administrative location field.",
96,
)
identifier_name = contains_keyword(
column_name,
IDENTIFIER_KEYWORDS,
)
if identifier_name and unique_ratio >= 0.95:
return (
"Identifier / Degenerate Dimension",
"Identifier-like field with highly unique values.",
98,
)
if identifier_name:
return (
"Operational Identifier Candidate",
"Identifier-like field requiring grain and duplicate review.",
88,
)
if (
contains_keyword(column_name, MEASURE_KEYWORDS)
and numeric_ratio >= 0.90
):
return (
"Numeric Measure Candidate",
"Measure-like name and strong numeric conversion.",
98,
)
if distinct_count == 2:
return (
"Boolean / Binary Dimension Candidate",
"Two observed values; suitable for binary analysis.",
95,
)
if (
contains_keyword(column_name, CATEGORY_KEYWORDS)
and distinct_count <= 1_000
):
return (
"Categorical Dimension Candidate",
"Category-like name and useful cardinality.",
96,
)
if distinct_count <= 100:
return (
"Categorical Dimension Candidate",
"Low-cardinality field suitable for grouping and filtering.",
90,
)
if distinct_count <= 1_000:
return (
"Possible Categorical Dimension",
"Moderate-cardinality field suitable for detailed segmentation.",
80,
)
if numeric_ratio >= 0.95:
return (
"Numeric Attribute or Measure",
"Most values are numeric; business review is required.",
84,
)
if unique_ratio >= 0.95:
return (
"Likely Free Text or Identifier",
"Most observed values are unique.",
84,
)
if null_percentage >= 95:
return (
"Sparse Field / Manual Review",
"Field is almost entirely null.",
94,
)
return (
"Manual Review Required",
"No strong analytical role was detected.",
50,
)
def suggest_sql_type(
column_name: str,
maximum_length: int,
distinct_count: int,
date_ratio: float,
numeric_ratio: float,
integer_ratio: float,
) -> str:
if (
contains_keyword(column_name, TEMPORAL_KEYWORDS)
and date_ratio >= 0.90
):
return "DATE"
if (
contains_keyword(column_name, MEASURE_KEYWORDS)
and numeric_ratio >= 0.90
):
return "DECIMAL(19,2)"
if distinct_count == 2:
return "NVARCHAR(20)"
if (
integer_ratio >= 0.98
and not contains_keyword(
column_name,
[
"zip", "number", "code", "permit",
"complaint", "referral", "violation",
],
)
):
return "BIGINT"
if numeric_ratio >= 0.98:
return "DECIMAL(19,4)"
if maximum_length <= 50:
return "NVARCHAR(50)"
if maximum_length <= 100:
return "NVARCHAR(100)"
if maximum_length <= 255:
return "NVARCHAR(255)"
if maximum_length <= 500:
return "NVARCHAR(500)"
if maximum_length <= 1_000:
return "NVARCHAR(1000)"
if maximum_length <= 4_000:
return "NVARCHAR(4000)"
return "NVARCHAR(MAX)"
# ============================================================
# PROFILING
# ============================================================
def profile_dataframe(
dataframe: pd.DataFrame,
) -> tuple[pd.DataFrame, dict[str, Any]]:
total_rows = len(dataframe)
report_rows = []
top_values_output: dict[str, Any] = {}
total_columns = len(dataframe.columns)
for position, column_name in enumerate(
dataframe.columns,
start=1,
):
print(
f"Profiling column "
f"{position:>2}/{total_columns}: "
f"{column_name}"
)
series = dataframe[column_name]
cleaned = clean_text_series(series)
non_null = cleaned.dropna()
null_count = int(cleaned.isna().sum())
non_null_count = int(non_null.shape[0])
null_percentage = (
null_count / total_rows * 100
if total_rows > 0
else 0
)
distinct_count = int(non_null.nunique(dropna=True))
unique_ratio = (
distinct_count / non_null_count
if non_null_count > 0
else 0
)
maximum_length = (
int(non_null.str.len().max())
if not non_null.empty
else 0
)
date_ratio = date_conversion_ratio(
series,
column_name,
)
numeric_ratio = numeric_conversion_ratio(series)
integer_ratio = integer_conversion_ratio(series)
top_values_series = (
non_null
.value_counts(dropna=True)
.head(TOP_VALUES_LIMIT)
)
top_values = [
{
"value": str(value),
"count": int(count),
}
for value, count in top_values_series.items()
]
top_values_output[column_name] = top_values
analytical_role, reason, confidence = classify_column(
column_name=column_name,
distinct_count=distinct_count,
unique_ratio=unique_ratio,
null_percentage=null_percentage,
date_ratio=date_ratio,
numeric_ratio=numeric_ratio,
)
suggested_type = suggest_sql_type(
column_name=column_name,
maximum_length=maximum_length,
distinct_count=distinct_count,
date_ratio=date_ratio,
numeric_ratio=numeric_ratio,
integer_ratio=integer_ratio,
)
report_rows.append(
{
"column_position": position,
"column_name": column_name,
"pandas_dtype": str(series.dtype),
"sample_rows": total_rows,
"non_null_count": non_null_count,
"null_count": null_count,
"null_percentage": round(null_percentage, 2),
"distinct_values_observed": distinct_count,
"unique_ratio_observed": round(unique_ratio, 4),
"maximum_length_observed": maximum_length,
"date_success_ratio": date_ratio,
"numeric_success_ratio": numeric_ratio,
"integer_success_ratio": integer_ratio,
"analytical_role": analytical_role,
"classification_reason": reason,
"classification_confidence": confidence,
"suggested_target_sql_type": suggested_type,
"top_values": " | ".join(
f"{item['value']} ({item['count']:,})"
for item in top_values
),
}
)
return pd.DataFrame(report_rows), top_values_output
# ============================================================
# REPORTS
# ============================================================
def build_star_schema(
profile_df: pd.DataFrame,
) -> dict[str, Any]:
def columns_for_roles(role_patterns: list[str]) -> list[str]:
mask = pd.Series(False, index=profile_df.index)
for pattern in role_patterns:
mask = mask | profile_df[
"analytical_role"
].str.contains(
pattern,
case=False,
na=False,
)
return profile_df.loc[
mask,
"column_name",
].tolist()
return {
"fact_table_candidate": "FactFireInspections",
"probable_grain": (
"One row per fire inspection record. "
"Confirm with Inspection Number."
),
"suggested_dimensions": [
"DimDate",
"DimInspectionType",
"DimInspectionStatus",
"DimReferralAgency",
"DimFireOrganization",
"DimLocation",
],
"detected_columns": {
"temporal_dimensions": columns_for_roles(["Temporal"]),
"geographic_dimensions": columns_for_roles(
["Geographic Dimension"]
),
"categorical_dimensions": columns_for_roles(
["Categorical Dimension", "Boolean"]
),
"identifiers": columns_for_roles(["Identifier"]),
"measures": columns_for_roles(["Measure"]),
"technical_metadata": columns_for_roles(
["Technical Metadata", "Spatial Metadata"]
),
},
}
def write_text_summary(
output_file: Path,
summary: dict[str, Any],
profile_df: pd.DataFrame,
star_schema: dict[str, Any],
) -> None:
with open(output_file, "w", encoding="utf-8") as file:
file.write("=" * 88 + "\n")
file.write("SQL DATASET PROFILE V2\n")
file.write("=" * 88 + "\n\n")
for key, value in summary.items():
file.write(f"{key}: {value}\n")
groups = [
("TEMPORAL DIMENSIONS", "Temporal"),
("GEOGRAPHIC DIMENSIONS", "Geographic Dimension"),
("CATEGORICAL DIMENSIONS", "Categorical Dimension|Boolean"),
("IDENTIFIERS", "Identifier"),
("MEASURES", "Measure"),
("TECHNICAL METADATA", "Technical Metadata|Spatial Metadata"),
]
for title, pattern in groups:
file.write("\n\n" + "=" * 88 + "\n")
file.write(title + "\n")
file.write("=" * 88 + "\n\n")
subset = profile_df[
profile_df["analytical_role"].str.contains(
pattern,
case=False,
na=False,
regex=True,
)
]
if subset.empty:
file.write("No candidates detected.\n")
continue
for _, row in subset.iterrows():
file.write(
f"{int(row['column_position'])}. "
f"{row['column_name']} "
f"[{row['analytical_role']}]\n"
)
file.write(
f" Suggested SQL type: "
f"{row['suggested_target_sql_type']}\n"
)
file.write(
f" Null %: {row['null_percentage']:.2f}\n"
)
file.write(
f" Distinct observed: "
f"{int(row['distinct_values_observed']):,}\n"
)
file.write(
f" Confidence: "
f"{int(row['classification_confidence'])}%\n"
)
file.write("\n\n" + "=" * 88 + "\n")
file.write("SUGGESTED STAR SCHEMA\n")
file.write("=" * 88 + "\n\n")
file.write(
f"Fact table: "
f"{star_schema['fact_table_candidate']}\n"
)
file.write(
f"Probable grain: "
f"{star_schema['probable_grain']}\n"
)
file.write("\nSuggested dimensions:\n")
for dimension in star_schema["suggested_dimensions"]:
file.write(f"- {dimension}\n")
def export_reports(
summary: dict[str, Any],
profile_df: pd.DataFrame,
top_values: dict[str, Any],
star_schema: dict[str, Any],
) -> None:
OUTPUT_FOLDER.mkdir(
parents=True,
exist_ok=True,
)
profile_file = OUTPUT_FOLDER / "sql_column_profile_v2.csv"
dimensions_file = OUTPUT_FOLDER / "dimension_candidates_v2.csv"
measures_file = OUTPUT_FOLDER / "measure_candidates_v2.csv"
identifiers_file = OUTPUT_FOLDER / "identifier_candidates_v2.csv"
top_values_file = OUTPUT_FOLDER / "column_top_values_v2.json"
profile_json_file = OUTPUT_FOLDER / "sql_table_profile_v2.json"
summary_file = OUTPUT_FOLDER / "sql_dataset_profile_summary_v2.txt"
profile_df.to_csv(
profile_file,
index=False,
encoding="utf-8-sig",
)
profile_df[
profile_df["analytical_role"].str.contains(
"Dimension|Temporal|Geographic",
case=False,
na=False,
regex=True,
)
].to_csv(
dimensions_file,
index=False,
encoding="utf-8-sig",
)
profile_df[
profile_df["analytical_role"].str.contains(
"Measure",
case=False,
na=False,
)
].to_csv(
measures_file,
index=False,
encoding="utf-8-sig",
)
profile_df[
profile_df["analytical_role"].str.contains(
"Identifier",
case=False,
na=False,
)
].to_csv(
identifiers_file,
index=False,
encoding="utf-8-sig",
)
with open(
top_values_file,
"w",
encoding="utf-8",
) as file:
json.dump(
top_values,
file,
indent=2,
ensure_ascii=False,
default=str,
)
with open(
profile_json_file,
"w",
encoding="utf-8",
) as file:
json.dump(
{
"summary": summary,
"star_schema": star_schema,
"columns": profile_df.to_dict(
orient="records"
),
},
file,
indent=2,
ensure_ascii=False,
default=str,
)
write_text_summary(
summary_file,
summary,
profile_df,
star_schema,
)
print("\nGenerated files:")
print(f"1. {profile_file}")
print(f"2. {dimensions_file}")
print(f"3. {measures_file}")
print(f"4. {identifiers_file}")
print(f"5. {top_values_file}")
print(f"6. {profile_json_file}")
print(f"7. {summary_file}")
# ============================================================
# MAIN
# ============================================================
def main() -> None:
validate_environment()
print_header("SQL DATASET PROFILER V2")
print(f"Server: {SQL_SERVER}")
print(f"Database: {DATABASE_NAME}")
print(f"Source table: {SOURCE_SCHEMA}.{SOURCE_TABLE}")
print(f"ODBC driver: {ODBC_DRIVER}")
print(f"Maximum sample size: {SAMPLE_SIZE:,}")
connection = connect_to_sql()
try:
validate_source_table(connection)
print_header("STEP 1 — TABLE VALIDATION")
total_rows = get_total_rows(connection)
actual_sample_size = min(SAMPLE_SIZE, total_rows)
print(f"Total rows in SQL: {total_rows:,}")
print(f"Rows to profile: {actual_sample_size:,}")
print_header("STEP 2 — SINGLE SQL READ")
dataframe = read_sample_to_dataframe(
connection,
actual_sample_size,
)
print(
f"Data loaded into pandas: "
f"{len(dataframe):,} rows x "
f"{len(dataframe.columns):,} columns"
)
finally:
connection.close()
print_header("STEP 3 — IN-MEMORY PROFILING")
profile_df, top_values = profile_dataframe(dataframe)
print_header("STEP 4 — STAR SCHEMA RECOMMENDATION")
star_schema = build_star_schema(profile_df)
summary = {
"Server": SQL_SERVER,
"Database": DATABASE_NAME,
"Source Table": f"{SOURCE_SCHEMA}.{SOURCE_TABLE}",
"Total Rows": f"{total_rows:,}",
"Profile Sample Rows": f"{len(dataframe):,}",
"Total Columns": f"{len(dataframe.columns):,}",
"Generated At": datetime.now().isoformat(
timespec="seconds"
),
}
export_reports(
summary,
profile_df,
top_values,
star_schema,
)
print_header("PROCESS COMPLETED SUCCESSFULLY")
print(f"Rows profiled: {len(dataframe):,}")
print(f"Columns profiled: {len(profile_df):,}")
print("\nAnalytical classifications:")
for role, count in (
profile_df["analytical_role"]
.value_counts()
.items()
):
print(f"- {role}: {count:,}")
print(
"\nMain report:\n"
f"{OUTPUT_FOLDER / 'sql_dataset_profile_summary_v2.txt'}"
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nProcess cancelled by the user.")
sys.exit(1)
except Exception as error:
print_header("PROCESS FAILED")
print(type(error).__name__)
print(error)
print(
"\nVerify SQL Server, database, table, "
"Windows permissions, and ODBC driver."
)
sys.exit(1)
Validated result: 53 columns profiled. The output identified geographic and temporal dimensions, categorical attributes, measures, identifiers, and technical metadata, creating the metadata foundation for the remaining modules.
Resultado validado: 53 columnas perfiladas. La salida identificó dimensiones geográficas y temporales, atributos categóricos, medidas, identificadores y metadata técnica, creando la base para los módulos siguientes.
03_AKPE_Data_Quality_Engine.py
Before normalization and modeling, the dataset must be evaluated with explicit rules. This separates observed data problems from transformation logic and provides a measurable baseline.
Antes de normalizar y modelar, el dataset debe evaluarse con reglas explícitas. Esto separa los problemas observados de la lógica de transformación y proporciona una línea base medible.
The engine evaluates completeness, duplicates, identifiers, dates, date sequences, amounts, boolean values, ZIP codes, and geographic consistency. It calculates a weighted quality score and creates detailed issue files with samples.
El motor evalúa completitud, duplicados, identificadores, fechas, secuencias temporales, importes, valores booleanos, códigos ZIP y consistencia geográfica. Calcula un score ponderado y crea archivos detallados con muestras.
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Any
import json
import re
import sys
import pandas as pd
import pyodbc
# ============================================================
# CONFIGURATION
# ============================================================
SCRIPT_FOLDER = Path(__file__).resolve().parent
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
SOURCE_SCHEMA = "staging"
SOURCE_TABLE = "FireInspections_Raw"
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
SAMPLE_SIZE = 100_000
TOP_ISSUES_LIMIT = 100
OUTPUT_FOLDER = SCRIPT_FOLDER / "akpe_data_quality_report"
# ============================================================
# EXPECTED BUSINESS COLUMNS
# ============================================================
DATE_COLUMNS = [
"Inspection Start Date",
"Inspection End Date",
"Return Date",
"Corrective Action Date",
"Invoice Date",
"Second Notice Date",
"Final Notice Date",
"Lien Date",
"Paid Date",
]
AMOUNT_COLUMNS = [
"Invoice Amount",
"Fee",
"Penalty Amount",
"Posting Fee",
"Interest Amount",
"Paid Amount",
]
IDENTIFIER_COLUMNS = [
"Inspection Number",
"Complaint Number",
"Permit Number",
"Referral Number",
"Violation Number",
"DBI Application Number",
]
BOOLEAN_COLUMNS = [
"Billable Inspection",
"Sent to Bureau of Delinquent Revenue",
]
GEOGRAPHIC_COLUMNS = [
"Address",
"zipcode",
"Supervisor District",
"neighborhood_district",
"location",
"Neighborhoods (old) 2",
"Zip Codes 2",
"Fire Prevention Districts 2",
"Police Districts 2",
"Supervisor Districts 2",
"Central Market/Tenderloin Boundary 2",
"Central Market/Tenderloin Boundary Polygon - Updated 2",
"Neighborhoods 2",
"SF Find Neighborhoods 2",
"Current Police Districts 2",
"Current Supervisor Districts 2",
"Analysis Neighborhoods 2",
]
TECHNICAL_COLUMNS = [
"AKPE_RowId",
"AKPE_LoadBatchId",
"AKPE_SourceFile",
"AKPE_LoadedAt",
"data_as_of",
"data_loaded_at",
]
# ============================================================
# HELPERS
# ============================================================
def print_header(title: str) -> None:
print("\n" + "=" * 92)
print(title)
print("=" * 92)
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
"TrustServerCertificate=yes;"
)
def connect_to_sql() -> pyodbc.Connection:
return pyodbc.connect(
connection_string(),
autocommit=True,
timeout=30,
)
def normalize_text_series(series: pd.Series) -> pd.Series:
normalized = series.astype("string").str.strip()
return normalized.replace(
{
"": pd.NA,
"nan": pd.NA,
"NaN": pd.NA,
"None": pd.NA,
"<NA>": pd.NA,
"NULL": pd.NA,
"null": pd.NA,
}
)
def currency_to_numeric(series: pd.Series) -> pd.Series:
cleaned = (
normalize_text_series(series)
.str.replace("$", "", regex=False)
.str.replace(",", "", regex=False)
.str.replace("(", "-", regex=False)
.str.replace(")", "", regex=False)
)
return pd.to_numeric(cleaned, errors="coerce")
def parse_dates(series: pd.Series) -> pd.Series:
cleaned = normalize_text_series(series)
return pd.to_datetime(
cleaned,
errors="coerce",
format="mixed",
)
def safe_percentage(numerator: int, denominator: int) -> float:
if denominator == 0:
return 0.0
return round(numerator / denominator * 100, 2)
def issue_severity(
issue_percentage: float,
critical: bool = False,
) -> str:
if critical and issue_percentage > 0:
return "Critical"
if issue_percentage >= 50:
return "Critical"
if issue_percentage >= 20:
return "High"
if issue_percentage >= 5:
return "Medium"
if issue_percentage > 0:
return "Low"
return "Pass"
# ============================================================
# SOURCE VALIDATION
# ============================================================
def validate_environment() -> None:
installed_drivers = pyodbc.drivers()
if ODBC_DRIVER not in installed_drivers:
raise RuntimeError(
f"ODBC driver not found: {ODBC_DRIVER}\n"
f"Installed drivers: {installed_drivers}"
)
def validate_source_table(
connection: pyodbc.Connection,
) -> None:
cursor = connection.cursor()
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
SOURCE_SCHEMA,
SOURCE_TABLE,
)
exists = cursor.fetchone()[0] == 1
if not exists:
raise RuntimeError(
f"Source table not found: "
f"{DATABASE_NAME}.{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
def get_total_rows(
connection: pyodbc.Connection,
) -> int:
full_table = (
f"{quote_identifier(SOURCE_SCHEMA)}."
f"{quote_identifier(SOURCE_TABLE)}"
)
cursor = connection.cursor()
cursor.execute(f"SELECT COUNT_BIG(*) FROM {full_table};")
return int(cursor.fetchone()[0])
def read_sample(
connection: pyodbc.Connection,
sample_size: int,
) -> pd.DataFrame:
full_table = (
f"{quote_identifier(SOURCE_SCHEMA)}."
f"{quote_identifier(SOURCE_TABLE)}"
)
sql = f"""
SELECT TOP ({sample_size}) *
FROM {full_table}
ORDER BY [AKPE_RowId];
"""
cursor = connection.cursor()
cursor.execute(sql)
columns = [item[0] for item in cursor.description]
rows = cursor.fetchall()
dataframe = pd.DataFrame.from_records(
rows,
columns=columns,
)
cursor.close()
return dataframe
# ============================================================
# GENERAL DATA QUALITY
# ============================================================
def profile_completeness(
dataframe: pd.DataFrame,
) -> pd.DataFrame:
total_rows = len(dataframe)
rows = []
for position, column in enumerate(
dataframe.columns,
start=1,
):
normalized = normalize_text_series(
dataframe[column]
)
null_count = int(normalized.isna().sum())
non_null_count = total_rows - null_count
completeness = safe_percentage(
non_null_count,
total_rows,
)
rows.append(
{
"column_position": position,
"column_name": column,
"total_rows": total_rows,
"non_null_count": non_null_count,
"null_or_blank_count": null_count,
"completeness_percentage": completeness,
"missing_percentage": round(
100 - completeness,
2,
),
"severity": issue_severity(
100 - completeness
),
}
)
return pd.DataFrame(rows)
def detect_duplicate_rows(
dataframe: pd.DataFrame,
) -> dict[str, Any]:
business_columns = [
column
for column in dataframe.columns
if column not in TECHNICAL_COLUMNS
]
duplicate_mask = dataframe.duplicated(
subset=business_columns,
keep=False,
)
duplicate_row_count = int(
duplicate_mask.sum()
)
duplicate_groups = int(
dataframe.loc[
duplicate_mask,
business_columns,
]
.drop_duplicates()
.shape[0]
)
return {
"business_columns_evaluated":
len(business_columns),
"duplicate_rows":
duplicate_row_count,
"duplicate_groups":
duplicate_groups,
"duplicate_percentage":
safe_percentage(
duplicate_row_count,
len(dataframe),
),
"severity":
issue_severity(
safe_percentage(
duplicate_row_count,
len(dataframe),
)
),
"duplicate_sample":
dataframe.loc[
duplicate_mask
].head(
TOP_ISSUES_LIMIT
),
}
# ============================================================
# IDENTIFIER QUALITY
# ============================================================
def profile_identifiers(
dataframe: pd.DataFrame,
) -> pd.DataFrame:
rows = []
total_rows = len(dataframe)
for column in IDENTIFIER_COLUMNS:
if column not in dataframe.columns:
continue
normalized = normalize_text_series(
dataframe[column]
)
non_null = normalized.dropna()
null_count = int(normalized.isna().sum())
distinct_count = int(
non_null.nunique()
)
duplicated_non_null = int(
non_null.duplicated(
keep=False
).sum()
)
rows.append(
{
"column_name": column,
"total_rows": total_rows,
"null_count": null_count,
"null_percentage": safe_percentage(
null_count,
total_rows,
),
"distinct_non_null_values":
distinct_count,
"duplicated_non_null_rows":
duplicated_non_null,
"duplicate_percentage_of_non_null":
safe_percentage(
duplicated_non_null,
len(non_null),
),
"candidate_primary_key":
(
null_count == 0
and distinct_count == total_rows
),
"severity":
issue_severity(
safe_percentage(
duplicated_non_null,
len(non_null),
),
critical=(
column
== "Inspection Number"
),
),
}
)
return pd.DataFrame(rows)
# ============================================================
# DATE QUALITY
# ============================================================
def profile_dates(
dataframe: pd.DataFrame,
) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]:
rows = []
issue_samples: dict[str, pd.DataFrame] = {}
today = pd.Timestamp.now().normalize()
for column in DATE_COLUMNS:
if column not in dataframe.columns:
continue
original = normalize_text_series(
dataframe[column]
)
parsed = parse_dates(
dataframe[column]
)
non_blank_mask = original.notna()
invalid_mask = (
non_blank_mask
& parsed.isna()
)
future_mask = (
parsed.notna()
& (parsed > today)
)
invalid_count = int(
invalid_mask.sum()
)
future_count = int(
future_mask.sum()
)
non_blank_count = int(
non_blank_mask.sum()
)
issue_percentage = safe_percentage(
invalid_count,
non_blank_count,
)
rows.append(
{
"column_name": column,
"non_blank_count":
non_blank_count,
"valid_date_count":
int(parsed.notna().sum()),
"invalid_date_count":
invalid_count,
"invalid_date_percentage":
issue_percentage,
"future_date_count":
future_count,
"minimum_valid_date":
(
parsed.min()
if parsed.notna().any()
else pd.NaT
),
"maximum_valid_date":
(
parsed.max()
if parsed.notna().any()
else pd.NaT
),
"severity":
issue_severity(
issue_percentage
),
}
)
if invalid_count > 0:
issue_samples[
f"{column}_invalid"
] = dataframe.loc[
invalid_mask
].head(
TOP_ISSUES_LIMIT
)
if future_count > 0:
issue_samples[
f"{column}_future"
] = dataframe.loc[
future_mask
].head(
TOP_ISSUES_LIMIT
)
return pd.DataFrame(rows), issue_samples
def profile_date_sequence_rules(
dataframe: pd.DataFrame,
) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]:
rules = [
(
"Inspection End Date before Inspection Start Date",
"Inspection Start Date",
"Inspection End Date",
),
(
"Return Date before Inspection Start Date",
"Inspection Start Date",
"Return Date",
),
(
"Corrective Action Date before Inspection Start Date",
"Inspection Start Date",
"Corrective Action Date",
),
(
"Paid Date before Invoice Date",
"Invoice Date",
"Paid Date",
),
(
"Second Notice Date before Invoice Date",
"Invoice Date",
"Second Notice Date",
),
(
"Final Notice Date before Second Notice Date",
"Second Notice Date",
"Final Notice Date",
),
(
"Lien Date before Final Notice Date",
"Final Notice Date",
"Lien Date",
),
]
rows = []
samples: dict[str, pd.DataFrame] = {}
for rule_name, earlier_column, later_column in rules:
if (
earlier_column not in dataframe.columns
or later_column not in dataframe.columns
):
continue
earlier = parse_dates(
dataframe[earlier_column]
)
later = parse_dates(
dataframe[later_column]
)
comparable_mask = (
earlier.notna()
& later.notna()
)
violation_mask = (
comparable_mask
& (later < earlier)
)
comparable_count = int(
comparable_mask.sum()
)
violation_count = int(
violation_mask.sum()
)
violation_percentage = safe_percentage(
violation_count,
comparable_count,
)
rows.append(
{
"rule_name": rule_name,
"earlier_column":
earlier_column,
"later_column":
later_column,
"comparable_rows":
comparable_count,
"violation_count":
violation_count,
"violation_percentage":
violation_percentage,
"severity":
issue_severity(
violation_percentage
),
}
)
if violation_count > 0:
key = re.sub(
r"[^A-Za-z0-9]+",
"_",
rule_name,
).strip("_")
samples[key] = dataframe.loc[
violation_mask
].head(
TOP_ISSUES_LIMIT
)
return pd.DataFrame(rows), samples
# ============================================================
# AMOUNT QUALITY
# ============================================================
def profile_amounts(
dataframe: pd.DataFrame,
) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]:
rows = []
samples: dict[str, pd.DataFrame] = {}
for column in AMOUNT_COLUMNS:
if column not in dataframe.columns:
continue
original = normalize_text_series(
dataframe[column]
)
numeric = currency_to_numeric(
dataframe[column]
)
non_blank_mask = original.notna()
invalid_mask = (
non_blank_mask
& numeric.isna()
)
negative_mask = (
numeric.notna()
& (numeric < 0)
)
non_blank_count = int(
non_blank_mask.sum()
)
invalid_count = int(
invalid_mask.sum()
)
negative_count = int(
negative_mask.sum()
)
invalid_percentage = safe_percentage(
invalid_count,
non_blank_count,
)
rows.append(
{
"column_name": column,
"non_blank_count":
non_blank_count,
"valid_numeric_count":
int(numeric.notna().sum()),
"invalid_numeric_count":
invalid_count,
"invalid_numeric_percentage":
invalid_percentage,
"negative_value_count":
negative_count,
"zero_value_count":
int(
(numeric == 0).sum()
),
"minimum_value":
(
float(numeric.min())
if numeric.notna().any()
else None
),
"maximum_value":
(
float(numeric.max())
if numeric.notna().any()
else None
),
"average_value":
(
round(
float(
numeric.mean()
),
2,
)
if numeric.notna().any()
else None
),
"severity":
issue_severity(
invalid_percentage
),
}
)
if invalid_count > 0:
samples[
f"{column}_invalid"
] = dataframe.loc[
invalid_mask
].head(
TOP_ISSUES_LIMIT
)
if negative_count > 0:
samples[
f"{column}_negative"
] = dataframe.loc[
negative_mask
].head(
TOP_ISSUES_LIMIT
)
return pd.DataFrame(rows), samples
def profile_financial_rules(
dataframe: pd.DataFrame,
) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]:
required = [
"Invoice Amount",
"Paid Amount",
]
if not all(
column in dataframe.columns
for column in required
):
return pd.DataFrame(), {}
invoice = currency_to_numeric(
dataframe["Invoice Amount"]
)
paid = currency_to_numeric(
dataframe["Paid Amount"]
)
comparable_mask = (
invoice.notna()
& paid.notna()
)
overpaid_mask = (
comparable_mask
& (paid > invoice)
)
rows = []
samples = {}
comparable_count = int(
comparable_mask.sum()
)
violation_count = int(
overpaid_mask.sum()
)
violation_percentage = safe_percentage(
violation_count,
comparable_count,
)
rows.append(
{
"rule_name":
"Paid Amount exceeds Invoice Amount",
"comparable_rows":
comparable_count,
"violation_count":
violation_count,
"violation_percentage":
violation_percentage,
"severity":
issue_severity(
violation_percentage
),
}
)
if violation_count > 0:
samples[
"Paid_Amount_exceeds_Invoice_Amount"
] = dataframe.loc[
overpaid_mask
].head(
TOP_ISSUES_LIMIT
)
return pd.DataFrame(rows), samples
# ============================================================
# BOOLEAN QUALITY
# ============================================================
def profile_boolean_columns(
dataframe: pd.DataFrame,
) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]:
accepted_values = {
"yes",
"no",
"y",
"n",
"true",
"false",
"1",
"0",
}
rows = []
samples: dict[str, pd.DataFrame] = {}
for column in BOOLEAN_COLUMNS:
if column not in dataframe.columns:
continue
normalized = (
normalize_text_series(
dataframe[column]
)
.str.lower()
)
non_null = normalized.dropna()
invalid_mask = (
normalized.notna()
& ~normalized.isin(
accepted_values
)
)
invalid_count = int(
invalid_mask.sum()
)
invalid_percentage = safe_percentage(
invalid_count,
len(non_null),
)
rows.append(
{
"column_name": column,
"non_null_count":
len(non_null),
"distinct_values":
int(
non_null.nunique()
),
"observed_values":
" | ".join(
sorted(
non_null.unique().tolist()
)
),
"invalid_value_count":
invalid_count,
"invalid_value_percentage":
invalid_percentage,
"severity":
issue_severity(
invalid_percentage
),
}
)
if invalid_count > 0:
samples[
f"{column}_invalid"
] = dataframe.loc[
invalid_mask
].head(
TOP_ISSUES_LIMIT
)
return pd.DataFrame(rows), samples
# ============================================================
# ZIP CODE QUALITY
# ============================================================
def profile_zip_codes(
dataframe: pd.DataFrame,
) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]:
candidate_columns = [
"zipcode",
"Zip Codes 2",
]
rows = []
samples: dict[str, pd.DataFrame] = {}
zip_pattern = re.compile(
r"^\d{5}(?:-\d{4})?$"
)
for column in candidate_columns:
if column not in dataframe.columns:
continue
normalized = normalize_text_series(
dataframe[column]
)
non_null = normalized.dropna()
valid_mask_non_null = non_null.str.match(
zip_pattern
)
invalid_values = non_null.loc[
~valid_mask_non_null
]
invalid_count = int(
len(invalid_values)
)
invalid_percentage = safe_percentage(
invalid_count,
len(non_null),
)
rows.append(
{
"column_name": column,
"non_null_count":
len(non_null),
"valid_zip_count":
int(
valid_mask_non_null.sum()
),
"invalid_zip_count":
invalid_count,
"invalid_zip_percentage":
invalid_percentage,
"severity":
issue_severity(
invalid_percentage
),
}
)
if invalid_count > 0:
invalid_index = (
invalid_values.index
)
samples[
f"{column}_invalid"
] = dataframe.loc[
invalid_index
].head(
TOP_ISSUES_LIMIT
)
return pd.DataFrame(rows), samples
# ============================================================
# GEOGRAPHIC CONSISTENCY
# ============================================================
def profile_geographic_consistency(
dataframe: pd.DataFrame,
) -> pd.DataFrame:
rows = []
comparable_pairs = [
(
"zipcode",
"Zip Codes 2",
),
(
"Supervisor District",
"Current Supervisor Districts 2",
),
(
"Fire Prevention District",
"Fire Prevention Districts 2",
),
(
"neighborhood_district",
"Analysis Neighborhoods 2",
),
]
for left_column, right_column in comparable_pairs:
if (
left_column not in dataframe.columns
or right_column not in dataframe.columns
):
continue
left = normalize_text_series(
dataframe[left_column]
)
right = normalize_text_series(
dataframe[right_column]
)
comparable_mask = (
left.notna()
& right.notna()
)
mismatch_mask = (
comparable_mask
& (left != right)
)
comparable_count = int(
comparable_mask.sum()
)
mismatch_count = int(
mismatch_mask.sum()
)
mismatch_percentage = safe_percentage(
mismatch_count,
comparable_count,
)
rows.append(
{
"left_column":
left_column,
"right_column":
right_column,
"comparable_rows":
comparable_count,
"mismatch_count":
mismatch_count,
"mismatch_percentage":
mismatch_percentage,
"severity":
issue_severity(
mismatch_percentage
),
"note":
(
"Potential mismatch only. "
"Confirm whether fields use the same coding system."
),
}
)
return pd.DataFrame(rows)
# ============================================================
# OVERALL SCORE
# ============================================================
def calculate_quality_score(
completeness_df: pd.DataFrame,
duplicate_info: dict[str, Any],
identifier_df: pd.DataFrame,
date_df: pd.DataFrame,
date_rule_df: pd.DataFrame,
amount_df: pd.DataFrame,
financial_rule_df: pd.DataFrame,
boolean_df: pd.DataFrame,
zip_df: pd.DataFrame,
) -> dict[str, Any]:
deductions = []
average_missing = (
float(
completeness_df[
"missing_percentage"
].mean()
)
if not completeness_df.empty
else 0.0
)
deductions.append(
min(
average_missing * 0.35,
20,
)
)
deductions.append(
min(
duplicate_info[
"duplicate_percentage"
] * 1.5,
20,
)
)
for dataframe, column in [
(
identifier_df,
"duplicate_percentage_of_non_null",
),
(
date_df,
"invalid_date_percentage",
),
(
date_rule_df,
"violation_percentage",
),
(
amount_df,
"invalid_numeric_percentage",
),
(
financial_rule_df,
"violation_percentage",
),
(
boolean_df,
"invalid_value_percentage",
),
(
zip_df,
"invalid_zip_percentage",
),
]:
if (
not dataframe.empty
and column in dataframe.columns
):
deductions.append(
min(
float(
dataframe[column].mean()
) * 0.15,
10,
)
)
total_deduction = sum(deductions)
score = max(
0.0,
round(
100 - total_deduction,
2,
),
)
if score >= 90:
rating = "Excellent"
elif score >= 80:
rating = "Good"
elif score >= 70:
rating = "Acceptable"
elif score >= 60:
rating = "Needs Improvement"
else:
rating = "High Risk"
return {
"quality_score": score,
"quality_rating": rating,
"method_note": (
"Heuristic score based on completeness, duplicates, "
"conversion failures, invalid sequences, and rule violations."
),
}
# ============================================================
# EXPORT
# ============================================================
def save_dataframe(
dataframe: pd.DataFrame,
filename: str,
) -> Path:
path = OUTPUT_FOLDER / filename
dataframe.to_csv(
path,
index=False,
encoding="utf-8-sig",
)
return path
def export_issue_samples(
issue_groups: list[
dict[str, pd.DataFrame]
],
) -> list[str]:
issue_folder = (
OUTPUT_FOLDER
/ "issue_samples"
)
issue_folder.mkdir(
parents=True,
exist_ok=True,
)
generated_files = []
combined: dict[str, pd.DataFrame] = {}
for group in issue_groups:
combined.update(group)
for issue_name, dataframe in combined.items():
safe_name = re.sub(
r"[^A-Za-z0-9_-]+",
"_",
issue_name,
).strip("_")
path = (
issue_folder
/ f"{safe_name}.csv"
)
dataframe.to_csv(
path,
index=False,
encoding="utf-8-sig",
)
generated_files.append(
str(path)
)
return generated_files
def write_summary(
output_file: Path,
summary: dict[str, Any],
completeness_df: pd.DataFrame,
duplicate_info: dict[str, Any],
identifier_df: pd.DataFrame,
date_df: pd.DataFrame,
date_rule_df: pd.DataFrame,
amount_df: pd.DataFrame,
financial_rule_df: pd.DataFrame,
boolean_df: pd.DataFrame,
zip_df: pd.DataFrame,
geographic_df: pd.DataFrame,
) -> None:
with open(
output_file,
"w",
encoding="utf-8",
) as file:
file.write("=" * 92 + "\n")
file.write("AKPE DATA QUALITY ENGINE\n")
file.write("=" * 92 + "\n\n")
for key, value in summary.items():
file.write(f"{key}: {value}\n")
file.write("\n")
file.write("=" * 92 + "\n")
file.write("DUPLICATE ROWS\n")
file.write("=" * 92 + "\n\n")
for key, value in duplicate_info.items():
if key != "duplicate_sample":
file.write(f"{key}: {value}\n")
sections = [
(
"LOW COMPLETENESS COLUMNS",
completeness_df.sort_values(
"missing_percentage",
ascending=False,
).head(20),
),
(
"IDENTIFIER QUALITY",
identifier_df,
),
(
"DATE QUALITY",
date_df,
),
(
"DATE SEQUENCE RULES",
date_rule_df,
),
(
"AMOUNT QUALITY",
amount_df,
),
(
"FINANCIAL RULES",
financial_rule_df,
),
(
"BOOLEAN QUALITY",
boolean_df,
),
(
"ZIP CODE QUALITY",
zip_df,
),
(
"GEOGRAPHIC CONSISTENCY",
geographic_df,
),
]
for title, dataframe in sections:
file.write("\n\n")
file.write("=" * 92 + "\n")
file.write(title + "\n")
file.write("=" * 92 + "\n\n")
if dataframe.empty:
file.write(
"No records or issues available.\n"
)
else:
file.write(
dataframe.to_string(
index=False
)
)
file.write("\n")
# ============================================================
# MAIN
# ============================================================
def main() -> None:
validate_environment()
print_header(
"AKPE DATA QUALITY ENGINE"
)
print(f"Server: {SQL_SERVER}")
print(f"Database: {DATABASE_NAME}")
print(
f"Source table: "
f"{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
print(f"ODBC driver: {ODBC_DRIVER}")
print(f"Maximum sample: {SAMPLE_SIZE:,}")
connection = connect_to_sql()
try:
validate_source_table(connection)
print_header(
"STEP 1 — SOURCE VALIDATION"
)
total_rows = get_total_rows(
connection
)
actual_sample_size = min(
SAMPLE_SIZE,
total_rows,
)
print(
f"Total rows in SQL: "
f"{total_rows:,}"
)
print(
f"Rows selected for quality review: "
f"{actual_sample_size:,}"
)
print_header(
"STEP 2 — SINGLE SQL READ"
)
dataframe = read_sample(
connection,
actual_sample_size,
)
print(
f"Data loaded into pandas: "
f"{len(dataframe):,} rows x "
f"{len(dataframe.columns):,} columns"
)
finally:
connection.close()
OUTPUT_FOLDER.mkdir(
parents=True,
exist_ok=True,
)
print_header(
"STEP 3 — COMPLETENESS"
)
completeness_df = profile_completeness(
dataframe
)
print(
f"Columns reviewed: "
f"{len(completeness_df):,}"
)
print_header(
"STEP 4 — DUPLICATES"
)
duplicate_info = detect_duplicate_rows(
dataframe
)
print(
f"Duplicate rows detected: "
f"{duplicate_info['duplicate_rows']:,}"
)
print_header(
"STEP 5 — IDENTIFIERS"
)
identifier_df = profile_identifiers(
dataframe
)
print(
f"Identifier fields reviewed: "
f"{len(identifier_df):,}"
)
print_header(
"STEP 6 — DATES"
)
date_df, date_samples = profile_dates(
dataframe
)
date_rule_df, date_rule_samples = (
profile_date_sequence_rules(
dataframe
)
)
print(
f"Date fields reviewed: "
f"{len(date_df):,}"
)
print_header(
"STEP 7 — AMOUNTS"
)
amount_df, amount_samples = (
profile_amounts(
dataframe
)
)
financial_rule_df, financial_samples = (
profile_financial_rules(
dataframe
)
)
print(
f"Amount fields reviewed: "
f"{len(amount_df):,}"
)
print_header(
"STEP 8 — BOOLEAN FIELDS"
)
boolean_df, boolean_samples = (
profile_boolean_columns(
dataframe
)
)
print(
f"Boolean fields reviewed: "
f"{len(boolean_df):,}"
)
print_header(
"STEP 9 — ZIP CODES"
)
zip_df, zip_samples = profile_zip_codes(
dataframe
)
print(
f"ZIP fields reviewed: "
f"{len(zip_df):,}"
)
print_header(
"STEP 10 — GEOGRAPHIC CONSISTENCY"
)
geographic_df = (
profile_geographic_consistency(
dataframe
)
)
print(
f"Geographic comparisons reviewed: "
f"{len(geographic_df):,}"
)
print_header(
"STEP 11 — QUALITY SCORE"
)
score = calculate_quality_score(
completeness_df,
duplicate_info,
identifier_df,
date_df,
date_rule_df,
amount_df,
financial_rule_df,
boolean_df,
zip_df,
)
print(
f"Quality score: "
f"{score['quality_score']}"
)
print(
f"Quality rating: "
f"{score['quality_rating']}"
)
print_header(
"STEP 12 — EXPORT REPORTS"
)
generated_files = []
generated_files.append(
str(
save_dataframe(
completeness_df,
"01_column_completeness.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
identifier_df,
"02_identifier_quality.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
date_df,
"03_date_quality.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
date_rule_df,
"04_date_sequence_rules.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
amount_df,
"05_amount_quality.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
financial_rule_df,
"06_financial_rules.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
boolean_df,
"07_boolean_quality.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
zip_df,
"08_zip_code_quality.csv",
)
)
)
generated_files.append(
str(
save_dataframe(
geographic_df,
"09_geographic_consistency.csv",
)
)
)
duplicate_sample = duplicate_info[
"duplicate_sample"
]
if not duplicate_sample.empty:
generated_files.append(
str(
save_dataframe(
duplicate_sample,
"10_duplicate_rows_sample.csv",
)
)
)
issue_sample_files = export_issue_samples(
[
date_samples,
date_rule_samples,
amount_samples,
financial_samples,
boolean_samples,
zip_samples,
]
)
generated_files.extend(
issue_sample_files
)
summary = {
"Server": SQL_SERVER,
"Database": DATABASE_NAME,
"Source Table":
f"{SOURCE_SCHEMA}.{SOURCE_TABLE}",
"Total SQL Rows":
f"{total_rows:,}",
"Sample Rows Reviewed":
f"{len(dataframe):,}",
"Columns Reviewed":
f"{len(dataframe.columns):,}",
"Duplicate Rows":
f"{duplicate_info['duplicate_rows']:,}",
"Quality Score":
score["quality_score"],
"Quality Rating":
score["quality_rating"],
"Generated At":
datetime.now().isoformat(
timespec="seconds"
),
"Scoring Note":
score["method_note"],
}
summary_file = (
OUTPUT_FOLDER
/ "akpe_data_quality_summary.txt"
)
write_summary(
summary_file,
summary,
completeness_df,
duplicate_info,
identifier_df,
date_df,
date_rule_df,
amount_df,
financial_rule_df,
boolean_df,
zip_df,
geographic_df,
)
generated_files.append(
str(summary_file)
)
json_file = (
OUTPUT_FOLDER
/ "akpe_data_quality_report.json"
)
duplicate_json = {
key: value
for key, value in duplicate_info.items()
if key != "duplicate_sample"
}
with open(
json_file,
"w",
encoding="utf-8",
) as file:
json.dump(
{
"summary": summary,
"duplicate_analysis":
duplicate_json,
"completeness":
completeness_df.to_dict(
orient="records"
),
"identifier_quality":
identifier_df.to_dict(
orient="records"
),
"date_quality":
date_df.to_dict(
orient="records"
),
"date_sequence_rules":
date_rule_df.to_dict(
orient="records"
),
"amount_quality":
amount_df.to_dict(
orient="records"
),
"financial_rules":
financial_rule_df.to_dict(
orient="records"
),
"boolean_quality":
boolean_df.to_dict(
orient="records"
),
"zip_quality":
zip_df.to_dict(
orient="records"
),
"geographic_consistency":
geographic_df.to_dict(
orient="records"
),
"issue_sample_files":
issue_sample_files,
},
file,
indent=2,
ensure_ascii=False,
default=str,
)
generated_files.append(
str(json_file)
)
print("\nGenerated files:")
for position, filename in enumerate(
generated_files,
start=1,
):
print(
f"{position}. {filename}"
)
print_header(
"PROCESS COMPLETED SUCCESSFULLY"
)
print(
f"Rows reviewed: "
f"{len(dataframe):,}"
)
print(
f"Columns reviewed: "
f"{len(dataframe.columns):,}"
)
print(
f"Quality score: "
f"{score['quality_score']}"
)
print(
f"Quality rating: "
f"{score['quality_rating']}"
)
print(
"\nMain report:\n"
f"{summary_file}"
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(
"\nProcess cancelled by the user."
)
sys.exit(1)
except Exception as error:
print_header(
"PROCESS FAILED"
)
print(
type(error).__name__
)
print(error)
print(
"\nVerify SQL Server, database, table, "
"Windows permissions, ODBC driver, "
"and expected column names."
)
sys.exit(1)
Validated result: 100,000 rows and 53 columns reviewed. The final Data Quality Score was 81.4 — Good, with detailed issue reports and representative samples.
Resultado validado: 100,000 filas y 53 columnas revisadas. El Data Quality Score final fue 81.4 — Good, acompañado por reportes detallados y muestras representativas.
04_AKPE_Normalization_Engine.py
Raw staging preserves source fidelity, but analytical models require stable data types, standardized values, and reusable business fields.
El staging raw preserva la fidelidad del origen, pero los modelos analíticos necesitan tipos estables, valores estandarizados y campos de negocio reutilizables.
Python orchestrates a SQL transformation that converts text into dates, decimals, integers, and bits; standardizes categorical values; derives useful fields; and writes a separate clean table without modifying the raw source.
Python coordina una transformación SQL que convierte texto en fechas, decimales, enteros y bits; estandariza valores categóricos; deriva campos útiles; y escribe una tabla clean separada sin modificar el origen raw.
from __future__ import annotations
from datetime import datetime
from pathlib import Path
import sys
import uuid
import pyodbc
# ============================================================
# CONFIGURATION
# ============================================================
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
SOURCE_SCHEMA = "staging"
SOURCE_TABLE = "FireInspections_Raw"
TARGET_SCHEMA = "staging"
TARGET_TABLE = "FireInspections_Clean"
AUDIT_SCHEMA = "audit"
AUDIT_TABLE = "NormalizationLog"
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
RECREATE_TARGET_TABLE = True
# ============================================================
# HELPERS
# ============================================================
def print_header(title: str) -> None:
print("\n" + "=" * 92)
print(title)
print("=" * 92)
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
"TrustServerCertificate=yes;"
)
def connect_to_sql() -> pyodbc.Connection:
return pyodbc.connect(
connection_string(),
autocommit=False,
timeout=60,
)
def validate_environment() -> None:
installed_drivers = pyodbc.drivers()
if ODBC_DRIVER not in installed_drivers:
raise RuntimeError(
f"ODBC driver not found: {ODBC_DRIVER}\n"
f"Installed drivers: {installed_drivers}"
)
def validate_source_table(
connection: pyodbc.Connection,
) -> None:
cursor = connection.cursor()
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
SOURCE_SCHEMA,
SOURCE_TABLE,
)
if cursor.fetchone()[0] != 1:
raise RuntimeError(
f"Source table not found: "
f"{DATABASE_NAME}.{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
def ensure_audit_schema_and_table(
connection: pyodbc.Connection,
) -> None:
cursor = connection.cursor()
cursor.execute(
f"""
IF NOT EXISTS
(
SELECT 1
FROM sys.schemas
WHERE name = N'{AUDIT_SCHEMA}'
)
BEGIN
EXEC(
'CREATE SCHEMA {quote_identifier(AUDIT_SCHEMA)}'
);
END;
"""
)
full_audit_table = (
f"{quote_identifier(AUDIT_SCHEMA)}."
f"{quote_identifier(AUDIT_TABLE)}"
)
cursor.execute(
f"""
IF OBJECT_ID(
N'{AUDIT_SCHEMA}.{AUDIT_TABLE}',
N'U'
) IS NULL
BEGIN
CREATE TABLE {full_audit_table}
(
NormalizationLogId BIGINT IDENTITY(1,1)
CONSTRAINT PK_NormalizationLog PRIMARY KEY,
NormalizationBatchId UNIQUEIDENTIFIER NOT NULL,
SourceSchema SYSNAME NOT NULL,
SourceTable SYSNAME NOT NULL,
TargetSchema SYSNAME NOT NULL,
TargetTable SYSNAME NOT NULL,
StartedAt DATETIME2(0) NOT NULL,
CompletedAt DATETIME2(0) NULL,
SourceRowCount BIGINT NULL,
TargetRowCount BIGINT NULL,
InvalidDateValues BIGINT NULL,
InvalidAmountValues BIGINT NULL,
InvalidBooleanValues BIGINT NULL,
Status NVARCHAR(30) NOT NULL,
ErrorMessage NVARCHAR(MAX) NULL
);
END;
"""
)
connection.commit()
def insert_audit_start(
connection: pyodbc.Connection,
batch_id: str,
started_at: datetime,
) -> None:
full_audit_table = (
f"{quote_identifier(AUDIT_SCHEMA)}."
f"{quote_identifier(AUDIT_TABLE)}"
)
cursor = connection.cursor()
cursor.execute(
f"""
INSERT INTO {full_audit_table}
(
NormalizationBatchId,
SourceSchema,
SourceTable,
TargetSchema,
TargetTable,
StartedAt,
Status
)
VALUES (?, ?, ?, ?, ?, ?, ?);
""",
batch_id,
SOURCE_SCHEMA,
SOURCE_TABLE,
TARGET_SCHEMA,
TARGET_TABLE,
started_at,
"Running",
)
connection.commit()
def update_audit_success(
connection: pyodbc.Connection,
batch_id: str,
source_rows: int,
target_rows: int,
invalid_dates: int,
invalid_amounts: int,
invalid_booleans: int,
) -> None:
full_audit_table = (
f"{quote_identifier(AUDIT_SCHEMA)}."
f"{quote_identifier(AUDIT_TABLE)}"
)
cursor = connection.cursor()
cursor.execute(
f"""
UPDATE {full_audit_table}
SET
CompletedAt = SYSDATETIME(),
SourceRowCount = ?,
TargetRowCount = ?,
InvalidDateValues = ?,
InvalidAmountValues = ?,
InvalidBooleanValues = ?,
Status = N'Completed',
ErrorMessage = NULL
WHERE NormalizationBatchId = ?;
""",
source_rows,
target_rows,
invalid_dates,
invalid_amounts,
invalid_booleans,
batch_id,
)
connection.commit()
def update_audit_failure(
connection: pyodbc.Connection,
batch_id: str,
error_message: str,
) -> None:
full_audit_table = (
f"{quote_identifier(AUDIT_SCHEMA)}."
f"{quote_identifier(AUDIT_TABLE)}"
)
cursor = connection.cursor()
cursor.execute(
f"""
UPDATE {full_audit_table}
SET
CompletedAt = SYSDATETIME(),
Status = N'Failed',
ErrorMessage = ?
WHERE NormalizationBatchId = ?;
""",
error_message,
batch_id,
)
connection.commit()
# ============================================================
# SQL EXPRESSIONS
# ============================================================
def null_trim_expression(column_name: str) -> str:
column = quote_identifier(column_name)
return (
f"NULLIF("
f"LTRIM(RTRIM(CONVERT(NVARCHAR(4000), {column}))), "
f"N''"
f")"
)
def normalized_text_expression(column_name: str) -> str:
return null_trim_expression(column_name)
def normalized_upper_expression(column_name: str) -> str:
return (
f"UPPER({null_trim_expression(column_name)})"
)
def date_expression(column_name: str) -> str:
source = null_trim_expression(column_name)
return (
"COALESCE("
f"TRY_CONVERT(DATE, {source}, 101), "
f"TRY_CONVERT(DATE, {source}, 120), "
f"TRY_CONVERT(DATE, {source}, 126), "
f"TRY_CONVERT(DATE, {source})"
")"
)
def amount_expression(column_name: str) -> str:
source = null_trim_expression(column_name)
cleaned = (
f"REPLACE("
f"REPLACE("
f"REPLACE("
f"REPLACE({source}, N'$', N''), "
f"N',', N''), "
f"N'(', N'-'), "
f"N')', N''"
f")"
)
return (
f"TRY_CONVERT(DECIMAL(19,2), {cleaned})"
)
def boolean_expression(column_name: str) -> str:
source = normalized_upper_expression(column_name)
return f"""
CASE
WHEN {source} IN (N'YES', N'Y', N'TRUE', N'1')
THEN CAST(1 AS BIT)
WHEN {source} IN (N'NO', N'N', N'FALSE', N'0')
THEN CAST(0 AS BIT)
ELSE NULL
END
"""
def zip_expression(column_name: str) -> str:
source = null_trim_expression(column_name)
return f"""
CASE
WHEN {source} IS NULL
THEN NULL
WHEN {source} LIKE N'[0-9][0-9][0-9][0-9][0-9]'
THEN {source}
WHEN {source} LIKE N'[0-9][0-9][0-9][0-9][0-9]-[0-9][0-9][0-9][0-9]'
THEN {source}
ELSE {source}
END
"""
# ============================================================
# TARGET STRUCTURE
# ============================================================
def create_target_table(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 2 — CREATE CLEAN TABLE")
cursor = connection.cursor()
full_target = (
f"{quote_identifier(TARGET_SCHEMA)}."
f"{quote_identifier(TARGET_TABLE)}"
)
cursor.execute(
f"""
IF NOT EXISTS
(
SELECT 1
FROM sys.schemas
WHERE name = N'{TARGET_SCHEMA}'
)
BEGIN
EXEC(
'CREATE SCHEMA {quote_identifier(TARGET_SCHEMA)}'
);
END;
"""
)
if RECREATE_TARGET_TABLE:
cursor.execute(
f"""
IF OBJECT_ID(
N'{TARGET_SCHEMA}.{TARGET_TABLE}',
N'U'
) IS NOT NULL
BEGIN
DROP TABLE {full_target};
END;
"""
)
cursor.execute(
f"""
CREATE TABLE {full_target}
(
CleanRowId BIGINT IDENTITY(1,1)
CONSTRAINT PK_FireInspections_Clean PRIMARY KEY,
NormalizationBatchId UNIQUEIDENTIFIER NOT NULL,
SourceAKPERowId BIGINT NOT NULL,
SourceLoadBatchId UNIQUEIDENTIFIER NULL,
SourceFile NVARCHAR(1000) NULL,
SourceLoadedAt DATETIME2(0) NULL,
InspectionNumber NVARCHAR(100) NULL,
InspectionType NVARCHAR(100) NULL,
InspectionTypeDescription NVARCHAR(500) NULL,
Address NVARCHAR(500) NULL,
ZipCode NVARCHAR(20) NULL,
Battalion NVARCHAR(100) NULL,
StationArea NVARCHAR(100) NULL,
FirePreventionDistrict NVARCHAR(100) NULL,
BillableInspection BIT NULL,
InspectionStartDate DATE NULL,
InspectionEndDate DATE NULL,
InspectionStatus NVARCHAR(100) NULL,
ReturnDate DATE NULL,
CorrectiveActionDate DATE NULL,
ReferralAgency NVARCHAR(255) NULL,
ComplaintNumber NVARCHAR(100) NULL,
PermitNumber NVARCHAR(100) NULL,
ReferralNumber NVARCHAR(100) NULL,
ViolationNumber NVARCHAR(100) NULL,
DBIApplicationNumber NVARCHAR(100) NULL,
InvoiceDate DATE NULL,
SecondNoticeDate DATE NULL,
FinalNoticeDate DATE NULL,
LienDate DATE NULL,
SentToBureauOfDelinquentRevenue BIT NULL,
InvoiceAmount DECIMAL(19,2) NULL,
Fee DECIMAL(19,2) NULL,
PenaltyAmount DECIMAL(19,2) NULL,
PostingFee DECIMAL(19,2) NULL,
InterestAmount DECIMAL(19,2) NULL,
PaidAmount DECIMAL(19,2) NULL,
PaidDate DATE NULL,
SupervisorDistrict NVARCHAR(100) NULL,
NeighborhoodDistrict NVARCHAR(255) NULL,
Location NVARCHAR(1000) NULL,
DataAsOf DATETIME2(0) NULL,
DataLoadedAt DATETIME2(0) NULL,
NeighborhoodsOld2 NVARCHAR(255) NULL,
ZipCodes2 NVARCHAR(100) NULL,
FirePreventionDistricts2 NVARCHAR(100) NULL,
PoliceDistricts2 NVARCHAR(100) NULL,
SupervisorDistricts2 NVARCHAR(100) NULL,
CentralMarketTenderloinBoundary2 NVARCHAR(100) NULL,
CentralMarketTenderloinBoundaryPolygonUpdated2 NVARCHAR(MAX) NULL,
Neighborhoods2 NVARCHAR(255) NULL,
SFFindNeighborhoods2 NVARCHAR(255) NULL,
CurrentPoliceDistricts2 NVARCHAR(100) NULL,
CurrentSupervisorDistricts2 NVARCHAR(100) NULL,
AnalysisNeighborhoods2 NVARCHAR(255) NULL,
InvalidDateCount TINYINT NOT NULL,
InvalidAmountCount TINYINT NOT NULL,
InvalidBooleanCount TINYINT NOT NULL,
HasNormalizationIssue AS
(
CASE
WHEN
InvalidDateCount > 0
OR InvalidAmountCount > 0
OR InvalidBooleanCount > 0
THEN CONVERT(BIT, 1)
ELSE CONVERT(BIT, 0)
END
),
NormalizedAt DATETIME2(0) NOT NULL
CONSTRAINT DF_FireInspections_Clean_NormalizedAt
DEFAULT SYSDATETIME()
);
"""
)
connection.commit()
print(
f"Clean table ready: "
f"{DATABASE_NAME}.{TARGET_SCHEMA}.{TARGET_TABLE}"
)
# ============================================================
# NORMALIZATION LOAD
# ============================================================
def normalize_and_load(
connection: pyodbc.Connection,
batch_id: str,
) -> None:
print_header("STEP 3 — NORMALIZE AND LOAD")
full_source = (
f"{quote_identifier(SOURCE_SCHEMA)}."
f"{quote_identifier(SOURCE_TABLE)}"
)
full_target = (
f"{quote_identifier(TARGET_SCHEMA)}."
f"{quote_identifier(TARGET_TABLE)}"
)
date_columns = [
"Inspection Start Date",
"Inspection End Date",
"Return Date",
"Corrective Action Date",
"Invoice Date",
"Second Notice Date",
"Final Notice Date",
"Lien Date",
"Paid Date",
]
amount_columns = [
"Invoice Amount",
"Fee",
"Penalty Amount",
"Posting Fee",
"Interest Amount",
"Paid Amount",
]
boolean_columns = [
"Billable Inspection",
"Sent to Bureau of Delinquent Revenue",
]
invalid_date_terms = []
for column_name in date_columns:
source = null_trim_expression(column_name)
parsed = date_expression(column_name)
invalid_date_terms.append(
f"""
CASE
WHEN {source} IS NOT NULL
AND {parsed} IS NULL
THEN 1
ELSE 0
END
"""
)
invalid_amount_terms = []
for column_name in amount_columns:
source = null_trim_expression(column_name)
parsed = amount_expression(column_name)
invalid_amount_terms.append(
f"""
CASE
WHEN {source} IS NOT NULL
AND {parsed} IS NULL
THEN 1
ELSE 0
END
"""
)
invalid_boolean_terms = []
for column_name in boolean_columns:
source = null_trim_expression(column_name)
parsed = boolean_expression(column_name)
invalid_boolean_terms.append(
f"""
CASE
WHEN {source} IS NOT NULL
AND {parsed} IS NULL
THEN 1
ELSE 0
END
"""
)
invalid_date_expression = (
" + ".join(invalid_date_terms)
)
invalid_amount_expression = (
" + ".join(invalid_amount_terms)
)
invalid_boolean_expression = (
" + ".join(invalid_boolean_terms)
)
cursor = connection.cursor()
cursor.execute(
f"""
INSERT INTO {full_target}
(
NormalizationBatchId,
SourceAKPERowId,
SourceLoadBatchId,
SourceFile,
SourceLoadedAt,
InspectionNumber,
InspectionType,
InspectionTypeDescription,
Address,
ZipCode,
Battalion,
StationArea,
FirePreventionDistrict,
BillableInspection,
InspectionStartDate,
InspectionEndDate,
InspectionStatus,
ReturnDate,
CorrectiveActionDate,
ReferralAgency,
ComplaintNumber,
PermitNumber,
ReferralNumber,
ViolationNumber,
DBIApplicationNumber,
InvoiceDate,
SecondNoticeDate,
FinalNoticeDate,
LienDate,
SentToBureauOfDelinquentRevenue,
InvoiceAmount,
Fee,
PenaltyAmount,
PostingFee,
InterestAmount,
PaidAmount,
PaidDate,
SupervisorDistrict,
NeighborhoodDistrict,
Location,
DataAsOf,
DataLoadedAt,
NeighborhoodsOld2,
ZipCodes2,
FirePreventionDistricts2,
PoliceDistricts2,
SupervisorDistricts2,
CentralMarketTenderloinBoundary2,
CentralMarketTenderloinBoundaryPolygonUpdated2,
Neighborhoods2,
SFFindNeighborhoods2,
CurrentPoliceDistricts2,
CurrentSupervisorDistricts2,
AnalysisNeighborhoods2,
InvalidDateCount,
InvalidAmountCount,
InvalidBooleanCount,
NormalizedAt
)
SELECT
?,
[AKPE_RowId],
[AKPE_LoadBatchId],
{normalized_text_expression("AKPE_SourceFile")},
TRY_CONVERT(
DATETIME2(0),
[AKPE_LoadedAt]
),
{normalized_text_expression("Inspection Number")},
{normalized_text_expression("Inspection Type")},
{normalized_text_expression("Inspection Type Description")},
{normalized_text_expression("Address")},
{zip_expression("zipcode")},
{normalized_text_expression("Battalion")},
{normalized_text_expression("Station Area")},
{normalized_text_expression("Fire Prevention District")},
{boolean_expression("Billable Inspection")},
{date_expression("Inspection Start Date")},
{date_expression("Inspection End Date")},
{normalized_text_expression("Inspection Status")},
{date_expression("Return Date")},
{date_expression("Corrective Action Date")},
{normalized_text_expression("Referral Agency")},
{normalized_text_expression("Complaint Number")},
{normalized_text_expression("Permit Number")},
{normalized_text_expression("Referral Number")},
{normalized_text_expression("Violation Number")},
{normalized_text_expression("DBI Application Number")},
{date_expression("Invoice Date")},
{date_expression("Second Notice Date")},
{date_expression("Final Notice Date")},
{date_expression("Lien Date")},
{boolean_expression("Sent to Bureau of Delinquent Revenue")},
{amount_expression("Invoice Amount")},
{amount_expression("Fee")},
{amount_expression("Penalty Amount")},
{amount_expression("Posting Fee")},
{amount_expression("Interest Amount")},
{amount_expression("Paid Amount")},
{date_expression("Paid Date")},
{normalized_text_expression("Supervisor District")},
{normalized_text_expression("neighborhood_district")},
{normalized_text_expression("location")},
COALESCE(
TRY_CONVERT(
DATETIME2(0),
{null_trim_expression("data_as_of")},
120
),
TRY_CONVERT(
DATETIME2(0),
{null_trim_expression("data_as_of")}
)
),
COALESCE(
TRY_CONVERT(
DATETIME2(0),
{null_trim_expression("data_loaded_at")},
120
),
TRY_CONVERT(
DATETIME2(0),
{null_trim_expression("data_loaded_at")}
)
),
{normalized_text_expression("Neighborhoods (old) 2")},
{normalized_text_expression("Zip Codes 2")},
{normalized_text_expression("Fire Prevention Districts 2")},
{normalized_text_expression("Police Districts 2")},
{normalized_text_expression("Supervisor Districts 2")},
{normalized_text_expression("Central Market/Tenderloin Boundary 2")},
{normalized_text_expression("Central Market/Tenderloin Boundary Polygon - Updated 2")},
{normalized_text_expression("Neighborhoods 2")},
{normalized_text_expression("SF Find Neighborhoods 2")},
{normalized_text_expression("Current Police Districts 2")},
{normalized_text_expression("Current Supervisor Districts 2")},
{normalized_text_expression("Analysis Neighborhoods 2")},
CONVERT(
TINYINT,
{invalid_date_expression}
),
CONVERT(
TINYINT,
{invalid_amount_expression}
),
CONVERT(
TINYINT,
{invalid_boolean_expression}
),
SYSDATETIME()
FROM {full_source};
""",
batch_id,
)
connection.commit()
print("Normalization load completed.")
# ============================================================
# INDEXES
# ============================================================
def create_indexes(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 4 — CREATE INDEXES")
full_target = (
f"{quote_identifier(TARGET_SCHEMA)}."
f"{quote_identifier(TARGET_TABLE)}"
)
cursor = connection.cursor()
index_statements = [
f"""
CREATE NONCLUSTERED INDEX
IX_FireInspections_Clean_InspectionNumber
ON {full_target}
(
InspectionNumber
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FireInspections_Clean_StartDate
ON {full_target}
(
InspectionStartDate
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FireInspections_Clean_Status
ON {full_target}
(
InspectionStatus
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FireInspections_Clean_Type
ON {full_target}
(
InspectionType
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FireInspections_Clean_ZipCode
ON {full_target}
(
ZipCode
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FireInspections_Clean_NormalizationIssue
ON {full_target}
(
HasNormalizationIssue
);
""",
]
for statement in index_statements:
cursor.execute(statement)
connection.commit()
print("Indexes created successfully.")
# ============================================================
# VALIDATION
# ============================================================
def validate_normalization(
connection: pyodbc.Connection,
) -> dict[str, int]:
print_header("STEP 5 — VALIDATION")
full_source = (
f"{quote_identifier(SOURCE_SCHEMA)}."
f"{quote_identifier(SOURCE_TABLE)}"
)
full_target = (
f"{quote_identifier(TARGET_SCHEMA)}."
f"{quote_identifier(TARGET_TABLE)}"
)
cursor = connection.cursor()
cursor.execute(
f"SELECT COUNT_BIG(*) FROM {full_source};"
)
source_rows = int(
cursor.fetchone()[0]
)
cursor.execute(
f"SELECT COUNT_BIG(*) FROM {full_target};"
)
target_rows = int(
cursor.fetchone()[0]
)
cursor.execute(
f"""
SELECT
SUM(CONVERT(BIGINT, InvalidDateCount)),
SUM(CONVERT(BIGINT, InvalidAmountCount)),
SUM(CONVERT(BIGINT, InvalidBooleanCount)),
SUM(
CASE
WHEN HasNormalizationIssue = 1
THEN 1
ELSE 0
END
)
FROM {full_target};
"""
)
result = cursor.fetchone()
invalid_dates = int(
result[0] or 0
)
invalid_amounts = int(
result[1] or 0
)
invalid_booleans = int(
result[2] or 0
)
rows_with_issues = int(
result[3] or 0
)
print(f"Source rows: {source_rows:,}")
print(f"Target rows: {target_rows:,}")
print(f"Invalid date values: {invalid_dates:,}")
print(f"Invalid amount values: {invalid_amounts:,}")
print(f"Invalid boolean values: {invalid_booleans:,}")
print(f"Rows with normalization issues: {rows_with_issues:,}")
if source_rows != target_rows:
raise RuntimeError(
"Row-count validation failed. "
f"Source: {source_rows:,}; "
f"Target: {target_rows:,}."
)
return {
"source_rows": source_rows,
"target_rows": target_rows,
"invalid_dates": invalid_dates,
"invalid_amounts": invalid_amounts,
"invalid_booleans": invalid_booleans,
"rows_with_issues": rows_with_issues,
}
# ============================================================
# MAIN
# ============================================================
def main() -> None:
validate_environment()
print_header(
"AKPE NORMALIZATION ENGINE"
)
print(f"Server: {SQL_SERVER}")
print(f"Database: {DATABASE_NAME}")
print(
f"Source: "
f"{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
print(
f"Target: "
f"{TARGET_SCHEMA}.{TARGET_TABLE}"
)
print(f"ODBC driver: {ODBC_DRIVER}")
batch_id = str(
uuid.uuid4()
)
started_at = datetime.now().replace(
microsecond=0
)
connection = connect_to_sql()
try:
validate_source_table(
connection
)
ensure_audit_schema_and_table(
connection
)
insert_audit_start(
connection,
batch_id,
started_at,
)
try:
print_header(
"STEP 1 — SOURCE CONFIRMED"
)
print(
f"Source table confirmed: "
f"{DATABASE_NAME}."
f"{SOURCE_SCHEMA}."
f"{SOURCE_TABLE}"
)
create_target_table(
connection
)
normalize_and_load(
connection,
batch_id,
)
create_indexes(
connection
)
validation = validate_normalization(
connection
)
update_audit_success(
connection,
batch_id,
validation["source_rows"],
validation["target_rows"],
validation["invalid_dates"],
validation["invalid_amounts"],
validation["invalid_booleans"],
)
except Exception as processing_error:
connection.rollback()
update_audit_failure(
connection,
batch_id,
str(processing_error),
)
raise
finally:
connection.close()
print_header(
"PROCESS COMPLETED SUCCESSFULLY"
)
print(f"Normalization batch ID: {batch_id}")
print(
f"Clean table: "
f"{DATABASE_NAME}."
f"{TARGET_SCHEMA}."
f"{TARGET_TABLE}"
)
print(
f"Rows normalized: "
f"{validation['target_rows']:,}"
)
print(
f"Rows with issues: "
f"{validation['rows_with_issues']:,}"
)
print(
"\nNext validation queries:\n"
)
print(
f"""
USE {quote_identifier(DATABASE_NAME)};
GO
SELECT COUNT_BIG(*) AS TotalCleanRows
FROM {quote_identifier(TARGET_SCHEMA)}.
{quote_identifier(TARGET_TABLE)};
GO
SELECT TOP (20) *
FROM {quote_identifier(TARGET_SCHEMA)}.
{quote_identifier(TARGET_TABLE)}
WHERE HasNormalizationIssue = 1
ORDER BY CleanRowId;
GO
SELECT
HasNormalizationIssue,
COUNT_BIG(*) AS TotalRows
FROM {quote_identifier(TARGET_SCHEMA)}.
{quote_identifier(TARGET_TABLE)}
GROUP BY HasNormalizationIssue;
GO
SELECT *
FROM {quote_identifier(AUDIT_SCHEMA)}.
{quote_identifier(AUDIT_TABLE)}
ORDER BY NormalizationLogId DESC;
GO
""".strip()
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(
"\nProcess cancelled by the user."
)
sys.exit(1)
except Exception as error:
print_header(
"PROCESS FAILED"
)
print(
type(error).__name__
)
print(error)
print(
"\nVerify SQL Server, database, source table, "
"column names, permissions, and ODBC driver."
)
sys.exit(1)
Validated result: staging.FireInspections_Clean created with 439,581 rows. The validation reported zero rows with normalization issues, and the execution was recorded in audit.NormalizationLog.
Resultado validado: staging.FireInspections_Clean creada con 439,581 filas. La validación reportó cero filas con problemas de normalización y la ejecución quedó registrada en audit.NormalizationLog.
05_AKPE_Semantic_Dimension_Builder.py
Dimensions give business meaning and stable filtering paths to the fact table. Surrogate keys isolate the analytical model from source-system changes.
Las dimensiones aportan significado de negocio y rutas estables de filtrado para la tabla de hechos. Las surrogate keys aíslan el modelo analítico de cambios en el sistema fuente.
The builder creates the dw schema, a complete date dimension, business dimensions, unknown members, unique indexes, and audit records. Distinct normalized values are loaded from the clean staging table.
El builder crea el schema dw, una dimensión de fecha completa, dimensiones de negocio, miembros desconocidos, índices únicos y registros de auditoría. Los valores normalizados se cargan desde la tabla clean.
from __future__ import annotations
from datetime import datetime
import sys
import uuid
import pyodbc
# ============================================================
# CONFIGURATION
# ============================================================
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
SOURCE_SCHEMA = "staging"
SOURCE_TABLE = "FireInspections_Clean"
DW_SCHEMA = "dw"
AUDIT_SCHEMA = "audit"
AUDIT_TABLE = "DimensionBuildLog"
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
RECREATE_DIMENSIONS = True
# ============================================================
# DIMENSION NAMES
# ============================================================
DIM_DATE = "DimDate"
DIM_INSPECTION_TYPE = "DimInspectionType"
DIM_INSPECTION_STATUS = "DimInspectionStatus"
DIM_REFERRAL_AGENCY = "DimReferralAgency"
DIM_FIRE_ORGANIZATION = "DimFireOrganization"
DIM_LOCATION = "DimLocation"
# ============================================================
# HELPERS
# ============================================================
def print_header(title: str) -> None:
print("\n" + "=" * 96)
print(title)
print("=" * 96)
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
"TrustServerCertificate=yes;"
)
def connect_to_sql() -> pyodbc.Connection:
return pyodbc.connect(
connection_string(),
autocommit=False,
timeout=60,
)
def validate_environment() -> None:
installed_drivers = pyodbc.drivers()
if ODBC_DRIVER not in installed_drivers:
raise RuntimeError(
f"ODBC driver not found: {ODBC_DRIVER}\n"
f"Installed drivers: {installed_drivers}"
)
def full_name(schema_name: str, table_name: str) -> str:
return (
f"{quote_identifier(schema_name)}."
f"{quote_identifier(table_name)}"
)
# ============================================================
# SOURCE VALIDATION
# ============================================================
def validate_source_table(
connection: pyodbc.Connection,
) -> None:
cursor = connection.cursor()
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
SOURCE_SCHEMA,
SOURCE_TABLE,
)
if cursor.fetchone()[0] != 1:
raise RuntimeError(
f"Source table not found: "
f"{DATABASE_NAME}.{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
# ============================================================
# SCHEMAS AND AUDIT
# ============================================================
def create_schemas_and_audit(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 1 — CREATE DW AND AUDIT STRUCTURE")
cursor = connection.cursor()
for schema_name in [DW_SCHEMA, AUDIT_SCHEMA]:
cursor.execute(
f"""
IF NOT EXISTS
(
SELECT 1
FROM sys.schemas
WHERE name = N'{schema_name}'
)
BEGIN
EXEC(
'CREATE SCHEMA {quote_identifier(schema_name)}'
);
END;
"""
)
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
IF OBJECT_ID(
N'{AUDIT_SCHEMA}.{AUDIT_TABLE}',
N'U'
) IS NULL
BEGIN
CREATE TABLE {audit_table}
(
DimensionBuildLogId BIGINT IDENTITY(1,1)
CONSTRAINT PK_DimensionBuildLog PRIMARY KEY,
DimensionBuildBatchId UNIQUEIDENTIFIER NOT NULL,
SourceSchema SYSNAME NOT NULL,
SourceTable SYSNAME NOT NULL,
StartedAt DATETIME2(0) NOT NULL,
CompletedAt DATETIME2(0) NULL,
DimDateRows BIGINT NULL,
DimInspectionTypeRows BIGINT NULL,
DimInspectionStatusRows BIGINT NULL,
DimReferralAgencyRows BIGINT NULL,
DimFireOrganizationRows BIGINT NULL,
DimLocationRows BIGINT NULL,
Status NVARCHAR(30) NOT NULL,
ErrorMessage NVARCHAR(MAX) NULL
);
END;
"""
)
connection.commit()
print(f"Schema ready: {DW_SCHEMA}")
print(f"Audit table ready: {AUDIT_SCHEMA}.{AUDIT_TABLE}")
def insert_audit_start(
connection: pyodbc.Connection,
batch_id: str,
started_at: datetime,
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
INSERT INTO {audit_table}
(
DimensionBuildBatchId,
SourceSchema,
SourceTable,
StartedAt,
Status
)
VALUES (?, ?, ?, ?, N'Running');
""",
batch_id,
SOURCE_SCHEMA,
SOURCE_TABLE,
started_at,
)
connection.commit()
def update_audit_success(
connection: pyodbc.Connection,
batch_id: str,
counts: dict[str, int],
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
DimDateRows = ?,
DimInspectionTypeRows = ?,
DimInspectionStatusRows = ?,
DimReferralAgencyRows = ?,
DimFireOrganizationRows = ?,
DimLocationRows = ?,
Status = N'Completed',
ErrorMessage = NULL
WHERE DimensionBuildBatchId = ?;
""",
counts[DIM_DATE],
counts[DIM_INSPECTION_TYPE],
counts[DIM_INSPECTION_STATUS],
counts[DIM_REFERRAL_AGENCY],
counts[DIM_FIRE_ORGANIZATION],
counts[DIM_LOCATION],
batch_id,
)
connection.commit()
def update_audit_failure(
connection: pyodbc.Connection,
batch_id: str,
error_message: str,
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
Status = N'Failed',
ErrorMessage = ?
WHERE DimensionBuildBatchId = ?;
""",
error_message,
batch_id,
)
connection.commit()
# ============================================================
# DROP DIMENSIONS
# ============================================================
def drop_dimensions_if_requested(
connection: pyodbc.Connection,
) -> None:
if not RECREATE_DIMENSIONS:
return
print_header("STEP 2 — RESET DIMENSIONS")
cursor = connection.cursor()
# Drop in reverse dependency order.
dimension_names = [
DIM_LOCATION,
DIM_FIRE_ORGANIZATION,
DIM_REFERRAL_AGENCY,
DIM_INSPECTION_STATUS,
DIM_INSPECTION_TYPE,
DIM_DATE,
]
for table_name in dimension_names:
cursor.execute(
f"""
IF OBJECT_ID(
N'{DW_SCHEMA}.{table_name}',
N'U'
) IS NOT NULL
BEGIN
DROP TABLE {full_name(DW_SCHEMA, table_name)};
END;
"""
)
connection.commit()
print("Existing dimension tables removed.")
# ============================================================
# DIMDATE
# ============================================================
def create_dim_date(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 3 — BUILD DimDate")
cursor = connection.cursor()
source = full_name(
SOURCE_SCHEMA,
SOURCE_TABLE,
)
target = full_name(
DW_SCHEMA,
DIM_DATE,
)
cursor.execute(
f"""
CREATE TABLE {target}
(
DateKey INT NOT NULL
CONSTRAINT PK_DimDate PRIMARY KEY,
FullDate DATE NOT NULL
CONSTRAINT UQ_DimDate_FullDate UNIQUE,
CalendarYear SMALLINT NOT NULL,
CalendarQuarter TINYINT NOT NULL,
QuarterName NVARCHAR(20) NOT NULL,
CalendarMonth TINYINT NOT NULL,
MonthName NVARCHAR(20) NOT NULL,
MonthShortName NVARCHAR(10) NOT NULL,
DayOfMonth TINYINT NOT NULL,
DayOfWeekNumber TINYINT NOT NULL,
DayOfWeekName NVARCHAR(20) NOT NULL,
WeekOfYear TINYINT NOT NULL,
YearMonthNumber INT NOT NULL,
YearMonthLabel CHAR(7) NOT NULL,
IsWeekend BIT NOT NULL
);
"""
)
cursor.execute(
f"""
DECLARE @MinimumDate DATE;
DECLARE @MaximumDate DATE;
SELECT
@MinimumDate = MIN(DateValue),
@MaximumDate = MAX(DateValue)
FROM
(
SELECT InspectionStartDate AS DateValue FROM {source}
UNION ALL
SELECT InspectionEndDate FROM {source}
UNION ALL
SELECT ReturnDate FROM {source}
UNION ALL
SELECT CorrectiveActionDate FROM {source}
UNION ALL
SELECT InvoiceDate FROM {source}
UNION ALL
SELECT SecondNoticeDate FROM {source}
UNION ALL
SELECT FinalNoticeDate FROM {source}
UNION ALL
SELECT LienDate FROM {source}
UNION ALL
SELECT PaidDate FROM {source}
) AS AllDates
WHERE DateValue IS NOT NULL;
IF @MinimumDate IS NULL
SET @MinimumDate = '2000-01-01';
IF @MaximumDate IS NULL
SET @MaximumDate = CAST(GETDATE() AS DATE);
;WITH DateSeries AS
(
SELECT @MinimumDate AS FullDate
UNION ALL
SELECT DATEADD(DAY, 1, FullDate)
FROM DateSeries
WHERE FullDate < @MaximumDate
)
INSERT INTO {target}
(
DateKey,
FullDate,
CalendarYear,
CalendarQuarter,
QuarterName,
CalendarMonth,
MonthName,
MonthShortName,
DayOfMonth,
DayOfWeekNumber,
DayOfWeekName,
WeekOfYear,
YearMonthNumber,
YearMonthLabel,
IsWeekend
)
SELECT
CONVERT(
INT,
CONVERT(
CHAR(8),
FullDate,
112
)
) AS DateKey,
FullDate,
DATEPART(YEAR, FullDate),
DATEPART(QUARTER, FullDate),
CONCAT(
N'Q',
DATEPART(QUARTER, FullDate)
),
DATEPART(MONTH, FullDate),
DATENAME(MONTH, FullDate),
LEFT(
DATENAME(MONTH, FullDate),
3
),
DATEPART(DAY, FullDate),
CASE
WHEN DATENAME(WEEKDAY, FullDate) = N'Sunday' THEN 1
WHEN DATENAME(WEEKDAY, FullDate) = N'Monday' THEN 2
WHEN DATENAME(WEEKDAY, FullDate) = N'Tuesday' THEN 3
WHEN DATENAME(WEEKDAY, FullDate) = N'Wednesday' THEN 4
WHEN DATENAME(WEEKDAY, FullDate) = N'Thursday' THEN 5
WHEN DATENAME(WEEKDAY, FullDate) = N'Friday' THEN 6
WHEN DATENAME(WEEKDAY, FullDate) = N'Saturday' THEN 7
END,
DATENAME(WEEKDAY, FullDate),
DATEPART(ISO_WEEK, FullDate),
DATEPART(YEAR, FullDate) * 100
+ DATEPART(MONTH, FullDate),
CONVERT(
CHAR(7),
FullDate,
120
),
CASE
WHEN DATENAME(WEEKDAY, FullDate)
IN (N'Saturday', N'Sunday')
THEN CAST(1 AS BIT)
ELSE CAST(0 AS BIT)
END
FROM DateSeries
OPTION (MAXRECURSION 0);
"""
)
connection.commit()
print("DimDate created and populated.")
# ============================================================
# DIMENSION TABLES
# ============================================================
def create_dim_inspection_type(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 4 — BUILD DimInspectionType")
cursor = connection.cursor()
source = full_name(SOURCE_SCHEMA, SOURCE_TABLE)
target = full_name(DW_SCHEMA, DIM_INSPECTION_TYPE)
cursor.execute(
f"""
CREATE TABLE {target}
(
InspectionTypeKey INT IDENTITY(1,1)
CONSTRAINT PK_DimInspectionType PRIMARY KEY,
InspectionType NVARCHAR(100) NOT NULL,
InspectionTypeDescription NVARCHAR(500) NULL,
IsUnknown BIT NOT NULL
CONSTRAINT DF_DimInspectionType_IsUnknown
DEFAULT 0,
CONSTRAINT UQ_DimInspectionType
UNIQUE
(
InspectionType,
InspectionTypeDescription
)
);
SET IDENTITY_INSERT {target} ON;
INSERT INTO {target}
(
InspectionTypeKey,
InspectionType,
InspectionTypeDescription,
IsUnknown
)
VALUES
(
-1,
N'Unknown',
N'Unknown or missing inspection type',
1
);
SET IDENTITY_INSERT {target} OFF;
INSERT INTO {target}
(
InspectionType,
InspectionTypeDescription,
IsUnknown
)
SELECT DISTINCT
COALESCE(
NULLIF(
LTRIM(RTRIM(InspectionType)),
N''
),
N'Unknown'
),
NULLIF(
LTRIM(
RTRIM(
InspectionTypeDescription
)
),
N''
),
0
FROM {source}
WHERE
InspectionType IS NOT NULL
OR InspectionTypeDescription IS NOT NULL;
"""
)
connection.commit()
print("DimInspectionType created and populated.")
def create_dim_inspection_status(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 5 — BUILD DimInspectionStatus")
cursor = connection.cursor()
source = full_name(SOURCE_SCHEMA, SOURCE_TABLE)
target = full_name(DW_SCHEMA, DIM_INSPECTION_STATUS)
cursor.execute(
f"""
CREATE TABLE {target}
(
InspectionStatusKey INT IDENTITY(1,1)
CONSTRAINT PK_DimInspectionStatus PRIMARY KEY,
InspectionStatus NVARCHAR(100) NOT NULL,
BillableInspection BIT NULL,
SentToBureauOfDelinquentRevenue BIT NULL,
IsUnknown BIT NOT NULL
CONSTRAINT DF_DimInspectionStatus_IsUnknown
DEFAULT 0,
CONSTRAINT UQ_DimInspectionStatus
UNIQUE
(
InspectionStatus,
BillableInspection,
SentToBureauOfDelinquentRevenue
)
);
SET IDENTITY_INSERT {target} ON;
INSERT INTO {target}
(
InspectionStatusKey,
InspectionStatus,
BillableInspection,
SentToBureauOfDelinquentRevenue,
IsUnknown
)
VALUES
(
-1,
N'Unknown',
NULL,
NULL,
1
);
SET IDENTITY_INSERT {target} OFF;
INSERT INTO {target}
(
InspectionStatus,
BillableInspection,
SentToBureauOfDelinquentRevenue,
IsUnknown
)
SELECT DISTINCT
COALESCE(
NULLIF(
LTRIM(RTRIM(InspectionStatus)),
N''
),
N'Unknown'
),
BillableInspection,
SentToBureauOfDelinquentRevenue,
0
FROM {source}
WHERE
InspectionStatus IS NOT NULL
OR BillableInspection IS NOT NULL
OR SentToBureauOfDelinquentRevenue IS NOT NULL;
"""
)
connection.commit()
print("DimInspectionStatus created and populated.")
def create_dim_referral_agency(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 6 — BUILD DimReferralAgency")
cursor = connection.cursor()
source = full_name(SOURCE_SCHEMA, SOURCE_TABLE)
target = full_name(DW_SCHEMA, DIM_REFERRAL_AGENCY)
cursor.execute(
f"""
CREATE TABLE {target}
(
ReferralAgencyKey INT IDENTITY(1,1)
CONSTRAINT PK_DimReferralAgency PRIMARY KEY,
ReferralAgency NVARCHAR(255) NOT NULL,
IsUnknown BIT NOT NULL
CONSTRAINT DF_DimReferralAgency_IsUnknown
DEFAULT 0,
CONSTRAINT UQ_DimReferralAgency
UNIQUE (ReferralAgency)
);
SET IDENTITY_INSERT {target} ON;
INSERT INTO {target}
(
ReferralAgencyKey,
ReferralAgency,
IsUnknown
)
VALUES
(
-1,
N'Unknown',
1
);
SET IDENTITY_INSERT {target} OFF;
INSERT INTO {target}
(
ReferralAgency,
IsUnknown
)
SELECT DISTINCT
LTRIM(RTRIM(ReferralAgency)),
0
FROM {source}
WHERE
NULLIF(
LTRIM(RTRIM(ReferralAgency)),
N''
) IS NOT NULL;
"""
)
connection.commit()
print("DimReferralAgency created and populated.")
def create_dim_fire_organization(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 7 — BUILD DimFireOrganization")
cursor = connection.cursor()
source = full_name(SOURCE_SCHEMA, SOURCE_TABLE)
target = full_name(DW_SCHEMA, DIM_FIRE_ORGANIZATION)
cursor.execute(
f"""
CREATE TABLE {target}
(
FireOrganizationKey INT IDENTITY(1,1)
CONSTRAINT PK_DimFireOrganization PRIMARY KEY,
Battalion NVARCHAR(100) NULL,
StationArea NVARCHAR(100) NULL,
FirePreventionDistrict NVARCHAR(100) NULL,
IsUnknown BIT NOT NULL
CONSTRAINT DF_DimFireOrganization_IsUnknown
DEFAULT 0,
OrganizationNaturalKey AS
(
CONCAT(
COALESCE(Battalion, N''),
N'|',
COALESCE(StationArea, N''),
N'|',
COALESCE(FirePreventionDistrict, N'')
)
) PERSISTED
);
CREATE UNIQUE NONCLUSTERED INDEX
UX_DimFireOrganization_NaturalKey
ON {target}
(
OrganizationNaturalKey
);
SET IDENTITY_INSERT {target} ON;
INSERT INTO {target}
(
FireOrganizationKey,
Battalion,
StationArea,
FirePreventionDistrict,
IsUnknown
)
VALUES
(
-1,
N'Unknown',
N'Unknown',
N'Unknown',
1
);
SET IDENTITY_INSERT {target} OFF;
INSERT INTO {target}
(
Battalion,
StationArea,
FirePreventionDistrict,
IsUnknown
)
SELECT DISTINCT
NULLIF(
LTRIM(RTRIM(Battalion)),
N''
),
NULLIF(
LTRIM(RTRIM(StationArea)),
N''
),
NULLIF(
LTRIM(
RTRIM(
FirePreventionDistrict
)
),
N''
),
0
FROM {source}
WHERE
Battalion IS NOT NULL
OR StationArea IS NOT NULL
OR FirePreventionDistrict IS NOT NULL;
"""
)
connection.commit()
print("DimFireOrganization created and populated.")
def create_dim_location(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 8 — BUILD DimLocation")
cursor = connection.cursor()
source = full_name(SOURCE_SCHEMA, SOURCE_TABLE)
target = full_name(DW_SCHEMA, DIM_LOCATION)
cursor.execute(
f"""
CREATE TABLE {target}
(
LocationKey INT IDENTITY(1,1)
CONSTRAINT PK_DimLocation PRIMARY KEY,
Address NVARCHAR(500) NULL,
ZipCode NVARCHAR(20) NULL,
NeighborhoodDistrict NVARCHAR(255) NULL,
AnalysisNeighborhood NVARCHAR(255) NULL,
Neighborhood NVARCHAR(255) NULL,
SupervisorDistrict NVARCHAR(100) NULL,
CurrentSupervisorDistrict NVARCHAR(100) NULL,
PoliceDistrict NVARCHAR(100) NULL,
CurrentPoliceDistrict NVARCHAR(100) NULL,
LocationText NVARCHAR(1000) NULL,
IsUnknown BIT NOT NULL
CONSTRAINT DF_DimLocation_IsUnknown
DEFAULT 0,
LocationNaturalKey AS
(
CONVERT(
CHAR(64),
HASHBYTES(
'SHA2_256',
CONCAT(
COALESCE(Address, N''),
N'|',
COALESCE(ZipCode, N''),
N'|',
COALESCE(NeighborhoodDistrict, N''),
N'|',
COALESCE(AnalysisNeighborhood, N''),
N'|',
COALESCE(SupervisorDistrict, N''),
N'|',
COALESCE(PoliceDistrict, N'')
)
),
2
)
) PERSISTED
);
CREATE UNIQUE NONCLUSTERED INDEX
UX_DimLocation_NaturalKey
ON {target}
(
LocationNaturalKey
);
SET IDENTITY_INSERT {target} ON;
INSERT INTO {target}
(
LocationKey,
Address,
ZipCode,
NeighborhoodDistrict,
AnalysisNeighborhood,
Neighborhood,
SupervisorDistrict,
CurrentSupervisorDistrict,
PoliceDistrict,
CurrentPoliceDistrict,
LocationText,
IsUnknown
)
VALUES
(
-1,
N'Unknown',
N'Unknown',
N'Unknown',
N'Unknown',
N'Unknown',
N'Unknown',
N'Unknown',
N'Unknown',
N'Unknown',
N'Unknown',
1
);
SET IDENTITY_INSERT {target} OFF;
INSERT INTO {target}
(
Address,
ZipCode,
NeighborhoodDistrict,
AnalysisNeighborhood,
Neighborhood,
SupervisorDistrict,
CurrentSupervisorDistrict,
PoliceDistrict,
CurrentPoliceDistrict,
LocationText,
IsUnknown
)
SELECT DISTINCT
NULLIF(
LTRIM(RTRIM(Address)),
N''
),
NULLIF(
LTRIM(RTRIM(ZipCode)),
N''
),
NULLIF(
LTRIM(
RTRIM(
NeighborhoodDistrict
)
),
N''
),
NULLIF(
LTRIM(
RTRIM(
AnalysisNeighborhoods2
)
),
N''
),
COALESCE(
NULLIF(
LTRIM(
RTRIM(
Neighborhoods2
)
),
N''
),
NULLIF(
LTRIM(
RTRIM(
SFFindNeighborhoods2
)
),
N''
),
NULLIF(
LTRIM(
RTRIM(
NeighborhoodsOld2
)
),
N''
)
),
NULLIF(
LTRIM(
RTRIM(
SupervisorDistrict
)
),
N''
),
NULLIF(
LTRIM(
RTRIM(
CurrentSupervisorDistricts2
)
),
N''
),
NULLIF(
LTRIM(
RTRIM(
PoliceDistricts2
)
),
N''
),
NULLIF(
LTRIM(
RTRIM(
CurrentPoliceDistricts2
)
),
N''
),
NULLIF(
LTRIM(RTRIM(Location)),
N''
),
0
FROM {source}
WHERE
Address IS NOT NULL
OR ZipCode IS NOT NULL
OR NeighborhoodDistrict IS NOT NULL
OR AnalysisNeighborhoods2 IS NOT NULL
OR Neighborhoods2 IS NOT NULL
OR SupervisorDistrict IS NOT NULL
OR PoliceDistricts2 IS NOT NULL
OR Location IS NOT NULL;
"""
)
connection.commit()
print("DimLocation created and populated.")
# ============================================================
# VALIDATION
# ============================================================
def count_rows(
connection: pyodbc.Connection,
schema_name: str,
table_name: str,
) -> int:
cursor = connection.cursor()
cursor.execute(
f"""
SELECT COUNT_BIG(*)
FROM {full_name(schema_name, table_name)};
"""
)
return int(cursor.fetchone()[0])
def validate_dimensions(
connection: pyodbc.Connection,
) -> dict[str, int]:
print_header("STEP 9 — VALIDATE DIMENSIONS")
dimension_names = [
DIM_DATE,
DIM_INSPECTION_TYPE,
DIM_INSPECTION_STATUS,
DIM_REFERRAL_AGENCY,
DIM_FIRE_ORGANIZATION,
DIM_LOCATION,
]
counts = {}
for dimension_name in dimension_names:
row_count = count_rows(
connection,
DW_SCHEMA,
dimension_name,
)
counts[dimension_name] = row_count
print(
f"{dimension_name}: "
f"{row_count:,} rows"
)
if row_count == 0:
raise RuntimeError(
f"Dimension is empty: "
f"{DW_SCHEMA}.{dimension_name}"
)
return counts
# ============================================================
# MAIN
# ============================================================
def main() -> None:
validate_environment()
print_header(
"AKPE SEMANTIC DIMENSION BUILDER"
)
print(f"Server: {SQL_SERVER}")
print(f"Database: {DATABASE_NAME}")
print(
f"Source: "
f"{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
print(f"Target schema: {DW_SCHEMA}")
print(f"ODBC driver: {ODBC_DRIVER}")
batch_id = str(uuid.uuid4())
started_at = datetime.now().replace(
microsecond=0
)
connection = connect_to_sql()
try:
validate_source_table(connection)
create_schemas_and_audit(
connection
)
insert_audit_start(
connection,
batch_id,
started_at,
)
try:
drop_dimensions_if_requested(
connection
)
create_dim_date(connection)
create_dim_inspection_type(connection)
create_dim_inspection_status(connection)
create_dim_referral_agency(connection)
create_dim_fire_organization(connection)
create_dim_location(connection)
counts = validate_dimensions(
connection
)
update_audit_success(
connection,
batch_id,
counts,
)
except Exception as processing_error:
connection.rollback()
update_audit_failure(
connection,
batch_id,
str(processing_error),
)
raise
finally:
connection.close()
print_header(
"PROCESS COMPLETED SUCCESSFULLY"
)
print(
f"Dimension build batch ID: "
f"{batch_id}"
)
print("\nCreated dimensions:")
for dimension_name, row_count in counts.items():
print(
f"- {DW_SCHEMA}.{dimension_name}: "
f"{row_count:,} rows"
)
print(
"\nNext validation query:\n"
)
print(
f"""
USE {quote_identifier(DATABASE_NAME)};
GO
SELECT COUNT_BIG(*) AS DimDateRows
FROM {full_name(DW_SCHEMA, DIM_DATE)};
GO
SELECT TOP (20) *
FROM {full_name(DW_SCHEMA, DIM_INSPECTION_TYPE)}
ORDER BY InspectionTypeKey;
GO
SELECT TOP (20) *
FROM {full_name(DW_SCHEMA, DIM_INSPECTION_STATUS)}
ORDER BY InspectionStatusKey;
GO
SELECT TOP (20) *
FROM {full_name(DW_SCHEMA, DIM_REFERRAL_AGENCY)}
ORDER BY ReferralAgencyKey;
GO
SELECT TOP (20) *
FROM {full_name(DW_SCHEMA, DIM_FIRE_ORGANIZATION)}
ORDER BY FireOrganizationKey;
GO
SELECT TOP (20) *
FROM {full_name(DW_SCHEMA, DIM_LOCATION)}
ORDER BY LocationKey;
GO
SELECT *
FROM {full_name(AUDIT_SCHEMA, AUDIT_TABLE)}
ORDER BY DimensionBuildLogId DESC;
GO
""".strip()
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(
"\nProcess cancelled by the user."
)
sys.exit(1)
except Exception as error:
print_header(
"PROCESS FAILED"
)
print(type(error).__name__)
print(error)
print(
"\nVerify SQL Server, database, source table, "
"column names, permissions, and ODBC driver."
)
sys.exit(1)
Validated result: dw.DimDate, dw.DimInspectionType, dw.DimInspectionStatus, dw.DimReferralAgency, dw.DimFireOrganization, and dw.DimLocation were created and populated.
Resultado validado: se crearon y poblaron dw.DimDate, dw.DimInspectionType, dw.DimInspectionStatus, dw.DimReferralAgency, dw.DimFireOrganization y dw.DimLocation.
06_AKPE_Fact_Builder.py
The fact table establishes the analytical grain and connects operational events to every descriptive dimension. It becomes the central table consumed by Power BI.
La tabla de hechos establece la granularidad analítica y conecta los eventos operativos con cada dimensión descriptiva. Se convierte en la tabla central consumida por Power BI.
The builder joins normalized source records to each dimension, resolves surrogate keys, assigns unknown keys where necessary, preserves operational identifiers, loads additive measures, and creates foreign keys and performance indexes.
El builder relaciona los registros normalizados con cada dimensión, resuelve surrogate keys, asigna claves unknown cuando es necesario, preserva identificadores operativos, carga medidas aditivas y crea foreign keys e índices.
from __future__ import annotations
from datetime import datetime
import sys
import uuid
import pyodbc
# ============================================================
# CONFIGURATION
# ============================================================
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
SOURCE_SCHEMA = "staging"
SOURCE_TABLE = "FireInspections_Clean"
DW_SCHEMA = "dw"
FACT_TABLE = "FactFireInspection"
AUDIT_SCHEMA = "audit"
AUDIT_TABLE = "FactBuildLog"
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
RECREATE_FACT_TABLE = True
# ============================================================
# HELPERS
# ============================================================
def print_header(title: str) -> None:
print("\n" + "=" * 96)
print(title)
print("=" * 96)
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def full_name(schema_name: str, table_name: str) -> str:
return (
f"{quote_identifier(schema_name)}."
f"{quote_identifier(table_name)}"
)
def connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
"TrustServerCertificate=yes;"
)
def connect_to_sql() -> pyodbc.Connection:
return pyodbc.connect(
connection_string(),
autocommit=False,
timeout=60,
)
def validate_environment() -> None:
installed_drivers = pyodbc.drivers()
if ODBC_DRIVER not in installed_drivers:
raise RuntimeError(
f"ODBC driver not found: {ODBC_DRIVER}\n"
f"Installed drivers: {installed_drivers}"
)
# ============================================================
# VALIDATION
# ============================================================
def validate_required_tables(
connection: pyodbc.Connection,
) -> None:
required_tables = [
(SOURCE_SCHEMA, SOURCE_TABLE),
(DW_SCHEMA, "DimDate"),
(DW_SCHEMA, "DimInspectionType"),
(DW_SCHEMA, "DimInspectionStatus"),
(DW_SCHEMA, "DimReferralAgency"),
(DW_SCHEMA, "DimFireOrganization"),
(DW_SCHEMA, "DimLocation"),
]
cursor = connection.cursor()
missing_tables = []
for schema_name, table_name in required_tables:
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
schema_name,
table_name,
)
exists = cursor.fetchone()[0] == 1
if not exists:
missing_tables.append(
f"{schema_name}.{table_name}"
)
if missing_tables:
raise RuntimeError(
"Required tables are missing:\n"
+ "\n".join(missing_tables)
)
# ============================================================
# AUDIT STRUCTURE
# ============================================================
def create_audit_table(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 1 — CREATE AUDIT STRUCTURE")
cursor = connection.cursor()
cursor.execute(
f"""
IF NOT EXISTS
(
SELECT 1
FROM sys.schemas
WHERE name = N'{AUDIT_SCHEMA}'
)
BEGIN
EXEC(
'CREATE SCHEMA {quote_identifier(AUDIT_SCHEMA)}'
);
END;
"""
)
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
IF OBJECT_ID(
N'{AUDIT_SCHEMA}.{AUDIT_TABLE}',
N'U'
) IS NULL
BEGIN
CREATE TABLE {audit_table}
(
FactBuildLogId BIGINT IDENTITY(1,1)
CONSTRAINT PK_FactBuildLog PRIMARY KEY,
FactBuildBatchId UNIQUEIDENTIFIER NOT NULL,
SourceSchema SYSNAME NOT NULL,
SourceTable SYSNAME NOT NULL,
TargetSchema SYSNAME NOT NULL,
TargetTable SYSNAME NOT NULL,
StartedAt DATETIME2(0) NOT NULL,
CompletedAt DATETIME2(0) NULL,
SourceRowCount BIGINT NULL,
FactRowCount BIGINT NULL,
UnknownInspectionTypeRows BIGINT NULL,
UnknownInspectionStatusRows BIGINT NULL,
UnknownReferralAgencyRows BIGINT NULL,
UnknownFireOrganizationRows BIGINT NULL,
UnknownLocationRows BIGINT NULL,
Status NVARCHAR(30) NOT NULL,
ErrorMessage NVARCHAR(MAX) NULL
);
END;
"""
)
connection.commit()
print(
f"Audit table ready: "
f"{AUDIT_SCHEMA}.{AUDIT_TABLE}"
)
def insert_audit_start(
connection: pyodbc.Connection,
batch_id: str,
started_at: datetime,
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
INSERT INTO {audit_table}
(
FactBuildBatchId,
SourceSchema,
SourceTable,
TargetSchema,
TargetTable,
StartedAt,
Status
)
VALUES (?, ?, ?, ?, ?, ?, N'Running');
""",
batch_id,
SOURCE_SCHEMA,
SOURCE_TABLE,
DW_SCHEMA,
FACT_TABLE,
started_at,
)
connection.commit()
def update_audit_success(
connection: pyodbc.Connection,
batch_id: str,
validation: dict[str, int],
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
SourceRowCount = ?,
FactRowCount = ?,
UnknownInspectionTypeRows = ?,
UnknownInspectionStatusRows = ?,
UnknownReferralAgencyRows = ?,
UnknownFireOrganizationRows = ?,
UnknownLocationRows = ?,
Status = N'Completed',
ErrorMessage = NULL
WHERE FactBuildBatchId = ?;
""",
validation["source_rows"],
validation["fact_rows"],
validation["unknown_inspection_type_rows"],
validation["unknown_inspection_status_rows"],
validation["unknown_referral_agency_rows"],
validation["unknown_fire_organization_rows"],
validation["unknown_location_rows"],
batch_id,
)
connection.commit()
def update_audit_failure(
connection: pyodbc.Connection,
batch_id: str,
error_message: str,
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
Status = N'Failed',
ErrorMessage = ?
WHERE FactBuildBatchId = ?;
""",
error_message,
batch_id,
)
connection.commit()
# ============================================================
# FACT TABLE
# ============================================================
def create_fact_table(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 2 — CREATE FACT TABLE")
cursor = connection.cursor()
fact_table = full_name(
DW_SCHEMA,
FACT_TABLE,
)
if RECREATE_FACT_TABLE:
cursor.execute(
f"""
IF OBJECT_ID(
N'{DW_SCHEMA}.{FACT_TABLE}',
N'U'
) IS NOT NULL
BEGIN
DROP TABLE {fact_table};
END;
"""
)
cursor.execute(
f"""
CREATE TABLE {fact_table}
(
FactFireInspectionKey BIGINT IDENTITY(1,1)
CONSTRAINT PK_FactFireInspection PRIMARY KEY,
FactBuildBatchId UNIQUEIDENTIFIER NOT NULL,
SourceCleanRowId BIGINT NOT NULL,
SourceAKPERowId BIGINT NOT NULL,
SourceLoadBatchId UNIQUEIDENTIFIER NULL,
NormalizationBatchId UNIQUEIDENTIFIER NOT NULL,
InspectionTypeKey INT NOT NULL,
InspectionStatusKey INT NOT NULL,
ReferralAgencyKey INT NOT NULL,
FireOrganizationKey INT NOT NULL,
LocationKey INT NOT NULL,
InspectionStartDateKey INT NULL,
InspectionEndDateKey INT NULL,
ReturnDateKey INT NULL,
CorrectiveActionDateKey INT NULL,
InvoiceDateKey INT NULL,
SecondNoticeDateKey INT NULL,
FinalNoticeDateKey INT NULL,
LienDateKey INT NULL,
PaidDateKey INT NULL,
InspectionNumber NVARCHAR(100) NULL,
ComplaintNumber NVARCHAR(100) NULL,
PermitNumber NVARCHAR(100) NULL,
ReferralNumber NVARCHAR(100) NULL,
ViolationNumber NVARCHAR(100) NULL,
DBIApplicationNumber NVARCHAR(100) NULL,
InspectionCount INT NOT NULL
CONSTRAINT DF_FactFireInspection_InspectionCount
DEFAULT 1,
InvoiceAmount DECIMAL(19,2) NULL,
Fee DECIMAL(19,2) NULL,
PenaltyAmount DECIMAL(19,2) NULL,
PostingFee DECIMAL(19,2) NULL,
InterestAmount DECIMAL(19,2) NULL,
PaidAmount DECIMAL(19,2) NULL,
OutstandingBalance AS
(
ISNULL(InvoiceAmount, 0)
+ ISNULL(Fee, 0)
+ ISNULL(PenaltyAmount, 0)
+ ISNULL(PostingFee, 0)
+ ISNULL(InterestAmount, 0)
- ISNULL(PaidAmount, 0)
) PERSISTED,
InspectionDurationDays AS
(
CASE
WHEN
InspectionStartDateKey IS NOT NULL
AND InspectionEndDateKey IS NOT NULL
THEN DATEDIFF(
DAY,
CONVERT(
DATE,
CONVERT(
CHAR(8),
InspectionStartDateKey
),
112
),
CONVERT(
DATE,
CONVERT(
CHAR(8),
InspectionEndDateKey
),
112
)
)
ELSE NULL
END
) PERSISTED,
DaysToPayment AS
(
CASE
WHEN
InvoiceDateKey IS NOT NULL
AND PaidDateKey IS NOT NULL
THEN DATEDIFF(
DAY,
CONVERT(
DATE,
CONVERT(
CHAR(8),
InvoiceDateKey
),
112
),
CONVERT(
DATE,
CONVERT(
CHAR(8),
PaidDateKey
),
112
)
)
ELSE NULL
END
) PERSISTED,
HasNormalizationIssue BIT NOT NULL,
FactCreatedAt DATETIME2(0) NOT NULL
CONSTRAINT DF_FactFireInspection_FactCreatedAt
DEFAULT SYSDATETIME(),
CONSTRAINT FK_FactFireInspection_DimInspectionType
FOREIGN KEY (InspectionTypeKey)
REFERENCES {full_name(DW_SCHEMA, "DimInspectionType")}
(InspectionTypeKey),
CONSTRAINT FK_FactFireInspection_DimInspectionStatus
FOREIGN KEY (InspectionStatusKey)
REFERENCES {full_name(DW_SCHEMA, "DimInspectionStatus")}
(InspectionStatusKey),
CONSTRAINT FK_FactFireInspection_DimReferralAgency
FOREIGN KEY (ReferralAgencyKey)
REFERENCES {full_name(DW_SCHEMA, "DimReferralAgency")}
(ReferralAgencyKey),
CONSTRAINT FK_FactFireInspection_DimFireOrganization
FOREIGN KEY (FireOrganizationKey)
REFERENCES {full_name(DW_SCHEMA, "DimFireOrganization")}
(FireOrganizationKey),
CONSTRAINT FK_FactFireInspection_DimLocation
FOREIGN KEY (LocationKey)
REFERENCES {full_name(DW_SCHEMA, "DimLocation")}
(LocationKey),
CONSTRAINT FK_FactFireInspection_InspectionStartDate
FOREIGN KEY (InspectionStartDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_InspectionEndDate
FOREIGN KEY (InspectionEndDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_ReturnDate
FOREIGN KEY (ReturnDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_CorrectiveActionDate
FOREIGN KEY (CorrectiveActionDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_InvoiceDate
FOREIGN KEY (InvoiceDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_SecondNoticeDate
FOREIGN KEY (SecondNoticeDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_FinalNoticeDate
FOREIGN KEY (FinalNoticeDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_LienDate
FOREIGN KEY (LienDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey),
CONSTRAINT FK_FactFireInspection_PaidDate
FOREIGN KEY (PaidDateKey)
REFERENCES {full_name(DW_SCHEMA, "DimDate")}
(DateKey)
);
"""
)
connection.commit()
print(
f"Fact table ready: "
f"{DATABASE_NAME}.{DW_SCHEMA}.{FACT_TABLE}"
)
# ============================================================
# FACT LOAD
# ============================================================
def load_fact_table(
connection: pyodbc.Connection,
batch_id: str,
) -> None:
print_header("STEP 3 — LOAD FACT TABLE")
cursor = connection.cursor()
source = full_name(
SOURCE_SCHEMA,
SOURCE_TABLE,
)
fact = full_name(
DW_SCHEMA,
FACT_TABLE,
)
dim_type = full_name(
DW_SCHEMA,
"DimInspectionType",
)
dim_status = full_name(
DW_SCHEMA,
"DimInspectionStatus",
)
dim_agency = full_name(
DW_SCHEMA,
"DimReferralAgency",
)
dim_org = full_name(
DW_SCHEMA,
"DimFireOrganization",
)
dim_location = full_name(
DW_SCHEMA,
"DimLocation",
)
cursor.execute(
f"""
INSERT INTO {fact}
(
FactBuildBatchId,
SourceCleanRowId,
SourceAKPERowId,
SourceLoadBatchId,
NormalizationBatchId,
InspectionTypeKey,
InspectionStatusKey,
ReferralAgencyKey,
FireOrganizationKey,
LocationKey,
InspectionStartDateKey,
InspectionEndDateKey,
ReturnDateKey,
CorrectiveActionDateKey,
InvoiceDateKey,
SecondNoticeDateKey,
FinalNoticeDateKey,
LienDateKey,
PaidDateKey,
InspectionNumber,
ComplaintNumber,
PermitNumber,
ReferralNumber,
ViolationNumber,
DBIApplicationNumber,
InspectionCount,
InvoiceAmount,
Fee,
PenaltyAmount,
PostingFee,
InterestAmount,
PaidAmount,
HasNormalizationIssue,
FactCreatedAt
)
SELECT
?,
src.CleanRowId,
src.SourceAKPERowId,
src.SourceLoadBatchId,
src.NormalizationBatchId,
COALESCE(dimType.InspectionTypeKey, -1),
COALESCE(dimStatus.InspectionStatusKey, -1),
COALESCE(dimAgency.ReferralAgencyKey, -1),
COALESCE(dimOrg.FireOrganizationKey, -1),
COALESCE(dimLocation.LocationKey, -1),
CASE
WHEN src.InspectionStartDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.InspectionStartDate,
112
)
)
END,
CASE
WHEN src.InspectionEndDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.InspectionEndDate,
112
)
)
END,
CASE
WHEN src.ReturnDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.ReturnDate,
112
)
)
END,
CASE
WHEN src.CorrectiveActionDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.CorrectiveActionDate,
112
)
)
END,
CASE
WHEN src.InvoiceDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.InvoiceDate,
112
)
)
END,
CASE
WHEN src.SecondNoticeDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.SecondNoticeDate,
112
)
)
END,
CASE
WHEN src.FinalNoticeDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.FinalNoticeDate,
112
)
)
END,
CASE
WHEN src.LienDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.LienDate,
112
)
)
END,
CASE
WHEN src.PaidDate IS NULL
THEN NULL
ELSE CONVERT(
INT,
CONVERT(
CHAR(8),
src.PaidDate,
112
)
)
END,
src.InspectionNumber,
src.ComplaintNumber,
src.PermitNumber,
src.ReferralNumber,
src.ViolationNumber,
src.DBIApplicationNumber,
1,
src.InvoiceAmount,
src.Fee,
src.PenaltyAmount,
src.PostingFee,
src.InterestAmount,
src.PaidAmount,
src.HasNormalizationIssue,
SYSDATETIME()
FROM {source} AS src
LEFT JOIN {dim_type} AS dimType
ON dimType.InspectionType =
COALESCE(
NULLIF(
LTRIM(RTRIM(src.InspectionType)),
N''
),
N'Unknown'
)
AND
(
dimType.InspectionTypeDescription =
NULLIF(
LTRIM(
RTRIM(
src.InspectionTypeDescription
)
),
N''
)
OR
(
dimType.InspectionTypeDescription IS NULL
AND
NULLIF(
LTRIM(
RTRIM(
src.InspectionTypeDescription
)
),
N''
) IS NULL
)
)
LEFT JOIN {dim_status} AS dimStatus
ON dimStatus.InspectionStatus =
COALESCE(
NULLIF(
LTRIM(RTRIM(src.InspectionStatus)),
N''
),
N'Unknown'
)
AND
(
dimStatus.BillableInspection =
src.BillableInspection
OR
(
dimStatus.BillableInspection IS NULL
AND src.BillableInspection IS NULL
)
)
AND
(
dimStatus.SentToBureauOfDelinquentRevenue =
src.SentToBureauOfDelinquentRevenue
OR
(
dimStatus.SentToBureauOfDelinquentRevenue IS NULL
AND
src.SentToBureauOfDelinquentRevenue IS NULL
)
)
LEFT JOIN {dim_agency} AS dimAgency
ON dimAgency.ReferralAgency =
NULLIF(
LTRIM(RTRIM(src.ReferralAgency)),
N''
)
LEFT JOIN {dim_org} AS dimOrg
ON dimOrg.OrganizationNaturalKey =
CONCAT(
COALESCE(
NULLIF(
LTRIM(RTRIM(src.Battalion)),
N''
),
N''
),
N'|',
COALESCE(
NULLIF(
LTRIM(RTRIM(src.StationArea)),
N''
),
N''
),
N'|',
COALESCE(
NULLIF(
LTRIM(
RTRIM(
src.FirePreventionDistrict
)
),
N''
),
N''
)
)
LEFT JOIN {dim_location} AS dimLocation
ON dimLocation.LocationNaturalKey =
CONVERT(
CHAR(64),
HASHBYTES(
'SHA2_256',
CONCAT(
COALESCE(
NULLIF(
LTRIM(RTRIM(src.Address)),
N''
),
N''
),
N'|',
COALESCE(
NULLIF(
LTRIM(RTRIM(src.ZipCode)),
N''
),
N''
),
N'|',
COALESCE(
NULLIF(
LTRIM(
RTRIM(
src.NeighborhoodDistrict
)
),
N''
),
N''
),
N'|',
COALESCE(
NULLIF(
LTRIM(
RTRIM(
src.AnalysisNeighborhoods2
)
),
N''
),
N''
),
N'|',
COALESCE(
NULLIF(
LTRIM(
RTRIM(
src.SupervisorDistrict
)
),
N''
),
N''
),
N'|',
COALESCE(
NULLIF(
LTRIM(
RTRIM(
src.PoliceDistricts2
)
),
N''
),
N''
)
)
),
2
);
""",
batch_id,
)
connection.commit()
print("Fact table load completed.")
# ============================================================
# INDEXES
# ============================================================
def create_indexes(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 4 — CREATE FACT INDEXES")
cursor = connection.cursor()
fact = full_name(
DW_SCHEMA,
FACT_TABLE,
)
index_statements = [
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_InspectionStartDateKey
ON {fact}
(
InspectionStartDateKey
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_InspectionTypeKey
ON {fact}
(
InspectionTypeKey
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_InspectionStatusKey
ON {fact}
(
InspectionStatusKey
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_LocationKey
ON {fact}
(
LocationKey
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_FireOrganizationKey
ON {fact}
(
FireOrganizationKey
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_ReferralAgencyKey
ON {fact}
(
ReferralAgencyKey
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_InspectionNumber
ON {fact}
(
InspectionNumber
);
""",
f"""
CREATE NONCLUSTERED INDEX
IX_FactFireInspection_FactBuildBatchId
ON {fact}
(
FactBuildBatchId
);
""",
]
for statement in index_statements:
cursor.execute(statement)
connection.commit()
print("Fact indexes created successfully.")
# ============================================================
# VALIDATION
# ============================================================
def validate_fact_table(
connection: pyodbc.Connection,
) -> dict[str, int]:
print_header("STEP 5 — VALIDATE FACT TABLE")
cursor = connection.cursor()
source = full_name(
SOURCE_SCHEMA,
SOURCE_TABLE,
)
fact = full_name(
DW_SCHEMA,
FACT_TABLE,
)
cursor.execute(
f"SELECT COUNT_BIG(*) FROM {source};"
)
source_rows = int(
cursor.fetchone()[0]
)
cursor.execute(
f"SELECT COUNT_BIG(*) FROM {fact};"
)
fact_rows = int(
cursor.fetchone()[0]
)
cursor.execute(
f"""
SELECT
SUM(
CASE
WHEN InspectionTypeKey = -1
THEN 1
ELSE 0
END
),
SUM(
CASE
WHEN InspectionStatusKey = -1
THEN 1
ELSE 0
END
),
SUM(
CASE
WHEN ReferralAgencyKey = -1
THEN 1
ELSE 0
END
),
SUM(
CASE
WHEN FireOrganizationKey = -1
THEN 1
ELSE 0
END
),
SUM(
CASE
WHEN LocationKey = -1
THEN 1
ELSE 0
END
)
FROM {fact};
"""
)
result = cursor.fetchone()
validation = {
"source_rows": source_rows,
"fact_rows": fact_rows,
"unknown_inspection_type_rows":
int(result[0] or 0),
"unknown_inspection_status_rows":
int(result[1] or 0),
"unknown_referral_agency_rows":
int(result[2] or 0),
"unknown_fire_organization_rows":
int(result[3] or 0),
"unknown_location_rows":
int(result[4] or 0),
}
for key, value in validation.items():
print(
f"{key}: {value:,}"
)
if source_rows != fact_rows:
raise RuntimeError(
"Fact row-count validation failed. "
f"Source: {source_rows:,}; "
f"Fact: {fact_rows:,}."
)
return validation
# ============================================================
# MAIN
# ============================================================
def main() -> None:
validate_environment()
print_header(
"AKPE FACT BUILDER"
)
print(f"Server: {SQL_SERVER}")
print(f"Database: {DATABASE_NAME}")
print(
f"Source: "
f"{SOURCE_SCHEMA}.{SOURCE_TABLE}"
)
print(
f"Target: "
f"{DW_SCHEMA}.{FACT_TABLE}"
)
print(f"ODBC driver: {ODBC_DRIVER}")
batch_id = str(
uuid.uuid4()
)
started_at = datetime.now().replace(
microsecond=0
)
connection = connect_to_sql()
try:
validate_required_tables(
connection
)
create_audit_table(
connection
)
insert_audit_start(
connection,
batch_id,
started_at,
)
try:
create_fact_table(
connection
)
load_fact_table(
connection,
batch_id,
)
create_indexes(
connection
)
validation = validate_fact_table(
connection
)
update_audit_success(
connection,
batch_id,
validation,
)
except Exception as processing_error:
connection.rollback()
update_audit_failure(
connection,
batch_id,
str(processing_error),
)
raise
finally:
connection.close()
print_header(
"PROCESS COMPLETED SUCCESSFULLY"
)
print(
f"Fact build batch ID: "
f"{batch_id}"
)
print(
f"Fact table: "
f"{DATABASE_NAME}.{DW_SCHEMA}.{FACT_TABLE}"
)
print(
f"Rows created: "
f"{validation['fact_rows']:,}"
)
print(
"\nNext validation queries:\n"
)
print(
f"""
USE {quote_identifier(DATABASE_NAME)};
GO
SELECT COUNT_BIG(*) AS FactRows
FROM {full_name(DW_SCHEMA, FACT_TABLE)};
GO
SELECT TOP (20) *
FROM {full_name(DW_SCHEMA, FACT_TABLE)}
ORDER BY FactFireInspectionKey;
GO
SELECT
InspectionTypeKey,
COUNT_BIG(*) AS TotalInspections
FROM {full_name(DW_SCHEMA, FACT_TABLE)}
GROUP BY InspectionTypeKey
ORDER BY TotalInspections DESC;
GO
SELECT
SUM(InvoiceAmount) AS TotalInvoiceAmount,
SUM(PaidAmount) AS TotalPaidAmount,
SUM(OutstandingBalance) AS TotalOutstandingBalance
FROM {full_name(DW_SCHEMA, FACT_TABLE)};
GO
SELECT *
FROM {full_name(AUDIT_SCHEMA, AUDIT_TABLE)}
ORDER BY FactBuildLogId DESC;
GO
""".strip()
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(
"\nProcess cancelled by the user."
)
sys.exit(1)
except Exception as error:
print_header(
"PROCESS FAILED"
)
print(type(error).__name__)
print(error)
print(
"\nVerify SQL Server, database, clean table, "
"dimension tables, permissions, and ODBC driver."
)
sys.exit(1)
Validated result: dw.FactFireInspection created with 439,581 rows. The table contains dimensional keys, date keys, operational identifiers, inspection counts, invoice amounts, paid amounts, balances, and supporting indexes.
Resultado validado: dw.FactFireInspection creada con 439,581 filas. La tabla contiene claves dimensionales, claves de fecha, identificadores operativos, conteos, importes, pagos, saldos e índices.
07_AKPE_Semantic_Layer_Builder.py
The star schema is optimized for analysis, while business-facing views make common questions easier to validate and consume across Power BI, Excel, and SQL.
El esquema estrella está optimizado para análisis, mientras las vistas de negocio facilitan validar y consumir preguntas comunes desde Power BI, Excel y SQL.
The semantic builder creates a mart schema and publishes reusable views for executive totals, monthly trends, inspection types, statuses, neighborhoods, outstanding balances, referral agencies, and fire organizations.
El semantic builder crea el schema mart y publica vistas reutilizables para totales ejecutivos, tendencias mensuales, tipos, estados, neighborhoods, saldos pendientes, agencias y organizaciones.
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Any
import csv
import json
import sys
import uuid
import pyodbc
SCRIPT_FOLDER = Path(__file__).resolve().parent
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
DW_SCHEMA = "dw"
MART_SCHEMA = "mart"
AUDIT_SCHEMA = "audit"
FACT_TABLE = "FactFireInspection"
AUDIT_TABLE = "SemanticLayerBuildLog"
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
OUTPUT_FOLDER = SCRIPT_FOLDER / "semantic_layer_output"
RECREATE_VIEWS = True
REQUIRED_TABLES = [
(DW_SCHEMA, "FactFireInspection"),
(DW_SCHEMA, "DimDate"),
(DW_SCHEMA, "DimInspectionType"),
(DW_SCHEMA, "DimInspectionStatus"),
(DW_SCHEMA, "DimReferralAgency"),
(DW_SCHEMA, "DimFireOrganization"),
(DW_SCHEMA, "DimLocation"),
]
VIEW_NAMES = [
"vwFactFireInspection",
"vwMonthlyInspectionTrend",
"vwInspectionTypeSummary",
"vwInspectionStatusSummary",
"vwNeighborhoodSummary",
"vwReferralAgencySummary",
"vwFireOrganizationSummary",
"vwOutstandingBalances",
"vwExecutiveDashboard",
]
def print_header(title: str) -> None:
print("\n" + "=" * 100)
print(title)
print("=" * 100)
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def full_name(schema_name: str, object_name: str) -> str:
return f"{quote_identifier(schema_name)}.{quote_identifier(object_name)}"
def connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
"TrustServerCertificate=yes;"
)
def connect_to_sql() -> pyodbc.Connection:
return pyodbc.connect(
connection_string(),
autocommit=False,
timeout=60,
)
def validate_environment() -> None:
installed_drivers = pyodbc.drivers()
if ODBC_DRIVER not in installed_drivers:
raise RuntimeError(
f"ODBC driver not found: {ODBC_DRIVER}\n"
f"Installed drivers: {installed_drivers}"
)
def validate_required_tables(connection: pyodbc.Connection) -> None:
cursor = connection.cursor()
missing = []
for schema_name, table_name in REQUIRED_TABLES:
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
schema_name,
table_name,
)
if cursor.fetchone()[0] != 1:
missing.append(f"{schema_name}.{table_name}")
if missing:
raise RuntimeError(
"Required Data Warehouse tables are missing:\n"
+ "\n".join(missing)
)
def create_schemas_and_audit_table(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 1 — CREATE MART AND AUDIT STRUCTURE")
cursor = connection.cursor()
for schema_name in [MART_SCHEMA, AUDIT_SCHEMA]:
cursor.execute(
f"""
IF NOT EXISTS
(
SELECT 1
FROM sys.schemas
WHERE name = N'{schema_name}'
)
BEGIN
EXEC(
'CREATE SCHEMA {quote_identifier(schema_name)}'
);
END;
"""
)
audit_table = full_name(AUDIT_SCHEMA, AUDIT_TABLE)
cursor.execute(
f"""
IF OBJECT_ID(
N'{AUDIT_SCHEMA}.{AUDIT_TABLE}',
N'U'
) IS NULL
BEGIN
CREATE TABLE {audit_table}
(
SemanticLayerBuildLogId BIGINT IDENTITY(1,1)
CONSTRAINT PK_SemanticLayerBuildLog PRIMARY KEY,
SemanticLayerBuildBatchId UNIQUEIDENTIFIER NOT NULL,
StartedAt DATETIME2(0) NOT NULL,
CompletedAt DATETIME2(0) NULL,
ViewCount INT NULL,
DataDictionaryRows INT NULL,
RelationshipRows INT NULL,
MeasureRows INT NULL,
DashboardRecommendationRows INT NULL,
Status NVARCHAR(30) NOT NULL,
ErrorMessage NVARCHAR(MAX) NULL
);
END;
"""
)
connection.commit()
print(f"Schema ready: {MART_SCHEMA}")
print(f"Audit table ready: {AUDIT_SCHEMA}.{AUDIT_TABLE}")
def insert_audit_start(
connection: pyodbc.Connection,
batch_id: str,
started_at: datetime,
) -> None:
cursor = connection.cursor()
audit_table = full_name(AUDIT_SCHEMA, AUDIT_TABLE)
cursor.execute(
f"""
INSERT INTO {audit_table}
(
SemanticLayerBuildBatchId,
StartedAt,
Status
)
VALUES (?, ?, N'Running');
""",
batch_id,
started_at,
)
connection.commit()
def update_audit_success(
connection: pyodbc.Connection,
batch_id: str,
counts: dict[str, int],
) -> None:
cursor = connection.cursor()
audit_table = full_name(AUDIT_SCHEMA, AUDIT_TABLE)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
ViewCount = ?,
DataDictionaryRows = ?,
RelationshipRows = ?,
MeasureRows = ?,
DashboardRecommendationRows = ?,
Status = N'Completed',
ErrorMessage = NULL
WHERE SemanticLayerBuildBatchId = ?;
""",
counts["views"],
counts["data_dictionary_rows"],
counts["relationship_rows"],
counts["measure_rows"],
counts["dashboard_rows"],
batch_id,
)
connection.commit()
def update_audit_failure(
connection: pyodbc.Connection,
batch_id: str,
error_message: str,
) -> None:
cursor = connection.cursor()
audit_table = full_name(AUDIT_SCHEMA, AUDIT_TABLE)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
Status = N'Failed',
ErrorMessage = ?
WHERE SemanticLayerBuildBatchId = ?;
""",
error_message,
batch_id,
)
connection.commit()
def drop_views_if_requested(
connection: pyodbc.Connection,
) -> None:
if not RECREATE_VIEWS:
return
print_header("STEP 2 — RESET SEMANTIC VIEWS")
cursor = connection.cursor()
for view_name in VIEW_NAMES:
cursor.execute(
f"""
IF OBJECT_ID(
N'{MART_SCHEMA}.{view_name}',
N'V'
) IS NOT NULL
BEGIN
DROP VIEW {full_name(MART_SCHEMA, view_name)};
END;
"""
)
connection.commit()
print("Existing semantic views removed.")
def create_views(connection: pyodbc.Connection) -> None:
print_header("STEP 3 — CREATE BUSINESS VIEWS")
cursor = connection.cursor()
fact = full_name(DW_SCHEMA, "FactFireInspection")
dim_date = full_name(DW_SCHEMA, "DimDate")
dim_type = full_name(DW_SCHEMA, "DimInspectionType")
dim_status = full_name(DW_SCHEMA, "DimInspectionStatus")
dim_agency = full_name(DW_SCHEMA, "DimReferralAgency")
dim_org = full_name(DW_SCHEMA, "DimFireOrganization")
dim_location = full_name(DW_SCHEMA, "DimLocation")
statements = []
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwFactFireInspection")}
AS
SELECT
fact.FactFireInspectionKey,
fact.InspectionNumber,
fact.ComplaintNumber,
fact.PermitNumber,
fact.ReferralNumber,
fact.ViolationNumber,
fact.DBIApplicationNumber,
typeDim.InspectionType,
typeDim.InspectionTypeDescription,
statusDim.InspectionStatus,
statusDim.BillableInspection,
statusDim.SentToBureauOfDelinquentRevenue,
agencyDim.ReferralAgency,
orgDim.Battalion,
orgDim.StationArea,
orgDim.FirePreventionDistrict,
locationDim.Address,
locationDim.ZipCode,
locationDim.NeighborhoodDistrict,
locationDim.AnalysisNeighborhood,
locationDim.Neighborhood,
locationDim.SupervisorDistrict,
locationDim.CurrentSupervisorDistrict,
locationDim.PoliceDistrict,
locationDim.CurrentPoliceDistrict,
locationDim.LocationText,
startDate.FullDate AS InspectionStartDate,
startDate.CalendarYear AS InspectionStartYear,
startDate.CalendarQuarter AS InspectionStartQuarter,
startDate.QuarterName AS InspectionStartQuarterName,
startDate.CalendarMonth AS InspectionStartMonthNumber,
startDate.MonthName AS InspectionStartMonthName,
startDate.YearMonthLabel AS InspectionStartYearMonth,
endDate.FullDate AS InspectionEndDate,
returnDate.FullDate AS ReturnDate,
correctiveDate.FullDate AS CorrectiveActionDate,
invoiceDate.FullDate AS InvoiceDate,
secondNoticeDate.FullDate AS SecondNoticeDate,
finalNoticeDate.FullDate AS FinalNoticeDate,
lienDate.FullDate AS LienDate,
paidDate.FullDate AS PaidDate,
fact.InspectionCount,
fact.InvoiceAmount,
fact.Fee,
fact.PenaltyAmount,
fact.PostingFee,
fact.InterestAmount,
fact.PaidAmount,
fact.OutstandingBalance,
fact.InspectionDurationDays,
fact.DaysToPayment,
fact.HasNormalizationIssue,
fact.FactCreatedAt
FROM {fact} AS fact
INNER JOIN {dim_type} AS typeDim
ON fact.InspectionTypeKey = typeDim.InspectionTypeKey
INNER JOIN {dim_status} AS statusDim
ON fact.InspectionStatusKey = statusDim.InspectionStatusKey
INNER JOIN {dim_agency} AS agencyDim
ON fact.ReferralAgencyKey = agencyDim.ReferralAgencyKey
INNER JOIN {dim_org} AS orgDim
ON fact.FireOrganizationKey = orgDim.FireOrganizationKey
INNER JOIN {dim_location} AS locationDim
ON fact.LocationKey = locationDim.LocationKey
LEFT JOIN {dim_date} AS startDate
ON fact.InspectionStartDateKey = startDate.DateKey
LEFT JOIN {dim_date} AS endDate
ON fact.InspectionEndDateKey = endDate.DateKey
LEFT JOIN {dim_date} AS returnDate
ON fact.ReturnDateKey = returnDate.DateKey
LEFT JOIN {dim_date} AS correctiveDate
ON fact.CorrectiveActionDateKey = correctiveDate.DateKey
LEFT JOIN {dim_date} AS invoiceDate
ON fact.InvoiceDateKey = invoiceDate.DateKey
LEFT JOIN {dim_date} AS secondNoticeDate
ON fact.SecondNoticeDateKey = secondNoticeDate.DateKey
LEFT JOIN {dim_date} AS finalNoticeDate
ON fact.FinalNoticeDateKey = finalNoticeDate.DateKey
LEFT JOIN {dim_date} AS lienDate
ON fact.LienDateKey = lienDate.DateKey
LEFT JOIN {dim_date} AS paidDate
ON fact.PaidDateKey = paidDate.DateKey;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwMonthlyInspectionTrend")}
AS
SELECT
dateDim.CalendarYear,
dateDim.CalendarQuarter,
dateDim.QuarterName,
dateDim.CalendarMonth,
dateDim.MonthName,
dateDim.YearMonthNumber,
dateDim.YearMonthLabel,
SUM(fact.InspectionCount) AS TotalInspections,
COUNT_BIG(DISTINCT fact.InspectionNumber)
AS DistinctInspectionNumbers,
SUM(fact.InvoiceAmount) AS TotalInvoiceAmount,
SUM(fact.PaidAmount) AS TotalPaidAmount,
SUM(fact.OutstandingBalance)
AS TotalOutstandingBalance,
AVG(CONVERT(DECIMAL(19,4), fact.InspectionDurationDays))
AS AverageInspectionDurationDays,
AVG(CONVERT(DECIMAL(19,4), fact.DaysToPayment))
AS AverageDaysToPayment
FROM {fact} AS fact
INNER JOIN {dim_date} AS dateDim
ON fact.InspectionStartDateKey = dateDim.DateKey
GROUP BY
dateDim.CalendarYear,
dateDim.CalendarQuarter,
dateDim.QuarterName,
dateDim.CalendarMonth,
dateDim.MonthName,
dateDim.YearMonthNumber,
dateDim.YearMonthLabel;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwInspectionTypeSummary")}
AS
SELECT
typeDim.InspectionType,
typeDim.InspectionTypeDescription,
SUM(fact.InspectionCount) AS TotalInspections,
SUM(fact.InvoiceAmount) AS TotalInvoiceAmount,
SUM(fact.PaidAmount) AS TotalPaidAmount,
SUM(fact.OutstandingBalance) AS TotalOutstandingBalance,
AVG(CONVERT(DECIMAL(19,4), fact.InspectionDurationDays))
AS AverageInspectionDurationDays
FROM {fact} AS fact
INNER JOIN {dim_type} AS typeDim
ON fact.InspectionTypeKey = typeDim.InspectionTypeKey
GROUP BY
typeDim.InspectionType,
typeDim.InspectionTypeDescription;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwInspectionStatusSummary")}
AS
SELECT
statusDim.InspectionStatus,
statusDim.BillableInspection,
statusDim.SentToBureauOfDelinquentRevenue,
SUM(fact.InspectionCount) AS TotalInspections,
SUM(fact.InvoiceAmount) AS TotalInvoiceAmount,
SUM(fact.PaidAmount) AS TotalPaidAmount,
SUM(fact.OutstandingBalance) AS TotalOutstandingBalance
FROM {fact} AS fact
INNER JOIN {dim_status} AS statusDim
ON fact.InspectionStatusKey = statusDim.InspectionStatusKey
GROUP BY
statusDim.InspectionStatus,
statusDim.BillableInspection,
statusDim.SentToBureauOfDelinquentRevenue;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwNeighborhoodSummary")}
AS
SELECT
locationDim.AnalysisNeighborhood,
locationDim.Neighborhood,
locationDim.NeighborhoodDistrict,
locationDim.ZipCode,
locationDim.SupervisorDistrict,
locationDim.PoliceDistrict,
SUM(fact.InspectionCount) AS TotalInspections,
SUM(fact.InvoiceAmount) AS TotalInvoiceAmount,
SUM(fact.PaidAmount) AS TotalPaidAmount,
SUM(fact.OutstandingBalance) AS TotalOutstandingBalance,
AVG(CONVERT(DECIMAL(19,4), fact.InspectionDurationDays))
AS AverageInspectionDurationDays
FROM {fact} AS fact
INNER JOIN {dim_location} AS locationDim
ON fact.LocationKey = locationDim.LocationKey
GROUP BY
locationDim.AnalysisNeighborhood,
locationDim.Neighborhood,
locationDim.NeighborhoodDistrict,
locationDim.ZipCode,
locationDim.SupervisorDistrict,
locationDim.PoliceDistrict;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwReferralAgencySummary")}
AS
SELECT
agencyDim.ReferralAgency,
SUM(fact.InspectionCount) AS TotalInspections,
SUM(fact.InvoiceAmount) AS TotalInvoiceAmount,
SUM(fact.PaidAmount) AS TotalPaidAmount,
SUM(fact.OutstandingBalance) AS TotalOutstandingBalance,
AVG(CONVERT(DECIMAL(19,4), fact.InspectionDurationDays))
AS AverageInspectionDurationDays
FROM {fact} AS fact
INNER JOIN {dim_agency} AS agencyDim
ON fact.ReferralAgencyKey = agencyDim.ReferralAgencyKey
GROUP BY
agencyDim.ReferralAgency;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwFireOrganizationSummary")}
AS
SELECT
orgDim.Battalion,
orgDim.StationArea,
orgDim.FirePreventionDistrict,
SUM(fact.InspectionCount) AS TotalInspections,
SUM(fact.InvoiceAmount) AS TotalInvoiceAmount,
SUM(fact.PaidAmount) AS TotalPaidAmount,
SUM(fact.OutstandingBalance) AS TotalOutstandingBalance,
AVG(CONVERT(DECIMAL(19,4), fact.InspectionDurationDays))
AS AverageInspectionDurationDays
FROM {fact} AS fact
INNER JOIN {dim_org} AS orgDim
ON fact.FireOrganizationKey = orgDim.FireOrganizationKey
GROUP BY
orgDim.Battalion,
orgDim.StationArea,
orgDim.FirePreventionDistrict;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwOutstandingBalances")}
AS
SELECT
fact.FactFireInspectionKey,
fact.InspectionNumber,
fact.InvoiceAmount,
fact.Fee,
fact.PenaltyAmount,
fact.PostingFee,
fact.InterestAmount,
fact.PaidAmount,
fact.OutstandingBalance,
statusDim.InspectionStatus,
agencyDim.ReferralAgency,
locationDim.Address,
locationDim.ZipCode,
locationDim.AnalysisNeighborhood,
invoiceDate.FullDate AS InvoiceDate,
paidDate.FullDate AS PaidDate,
fact.DaysToPayment
FROM {fact} AS fact
INNER JOIN {dim_status} AS statusDim
ON fact.InspectionStatusKey = statusDim.InspectionStatusKey
INNER JOIN {dim_agency} AS agencyDim
ON fact.ReferralAgencyKey = agencyDim.ReferralAgencyKey
INNER JOIN {dim_location} AS locationDim
ON fact.LocationKey = locationDim.LocationKey
LEFT JOIN {dim_date} AS invoiceDate
ON fact.InvoiceDateKey = invoiceDate.DateKey
LEFT JOIN {dim_date} AS paidDate
ON fact.PaidDateKey = paidDate.DateKey
WHERE fact.OutstandingBalance <> 0;
"""
)
statements.append(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwExecutiveDashboard")}
AS
SELECT
SUM(fact.InspectionCount) AS TotalInspections,
COUNT_BIG(DISTINCT fact.InspectionNumber)
AS DistinctInspectionNumbers,
SUM(fact.InvoiceAmount) AS TotalInvoiceAmount,
SUM(fact.PaidAmount) AS TotalPaidAmount,
SUM(fact.OutstandingBalance) AS TotalOutstandingBalance,
CASE
WHEN SUM(fact.InvoiceAmount) = 0
THEN NULL
ELSE
SUM(fact.PaidAmount)
/ NULLIF(SUM(fact.InvoiceAmount), 0)
END AS CollectionRate,
AVG(CONVERT(DECIMAL(19,4), fact.InspectionDurationDays))
AS AverageInspectionDurationDays,
AVG(CONVERT(DECIMAL(19,4), fact.DaysToPayment))
AS AverageDaysToPayment,
SUM(
CASE
WHEN fact.OutstandingBalance > 0
THEN 1
ELSE 0
END
) AS InspectionsWithOutstandingBalance,
SUM(
CASE
WHEN fact.HasNormalizationIssue = 1
THEN 1
ELSE 0
END
) AS RowsWithNormalizationIssues
FROM {fact} AS fact;
"""
)
for statement in statements:
cursor.execute(statement)
connection.commit()
print(f"Created {len(VIEW_NAMES)} semantic views.")
def read_table_columns(
connection: pyodbc.Connection,
) -> list[dict[str, Any]]:
cursor = connection.cursor()
cursor.execute(
"""
SELECT
c.TABLE_SCHEMA,
c.TABLE_NAME,
c.ORDINAL_POSITION,
c.COLUMN_NAME,
c.DATA_TYPE,
c.CHARACTER_MAXIMUM_LENGTH,
c.NUMERIC_PRECISION,
c.NUMERIC_SCALE,
c.IS_NULLABLE
FROM INFORMATION_SCHEMA.COLUMNS AS c
WHERE c.TABLE_SCHEMA = ?
AND c.TABLE_NAME IN
(
'FactFireInspection',
'DimDate',
'DimInspectionType',
'DimInspectionStatus',
'DimReferralAgency',
'DimFireOrganization',
'DimLocation'
)
ORDER BY
c.TABLE_SCHEMA,
c.TABLE_NAME,
c.ORDINAL_POSITION;
""",
DW_SCHEMA,
)
rows = []
for row in cursor.fetchall():
rows.append(
{
"schema_name": row[0],
"table_name": row[1],
"column_position": int(row[2]),
"column_name": row[3],
"sql_data_type": row[4],
"maximum_length": row[5],
"numeric_precision": row[6],
"numeric_scale": row[7],
"is_nullable": row[8],
}
)
return rows
def read_relationships(
connection: pyodbc.Connection,
) -> list[dict[str, Any]]:
cursor = connection.cursor()
cursor.execute(
"""
SELECT
fk.name AS foreign_key_name,
OBJECT_SCHEMA_NAME(fk.parent_object_id) AS fact_schema,
OBJECT_NAME(fk.parent_object_id) AS fact_table,
parentColumn.name AS fact_column,
OBJECT_SCHEMA_NAME(fk.referenced_object_id) AS dimension_schema,
OBJECT_NAME(fk.referenced_object_id) AS dimension_table,
referencedColumn.name AS dimension_column
FROM sys.foreign_keys AS fk
INNER JOIN sys.foreign_key_columns AS fkc
ON fk.object_id = fkc.constraint_object_id
INNER JOIN sys.columns AS parentColumn
ON parentColumn.object_id = fk.parent_object_id
AND parentColumn.column_id = fkc.parent_column_id
INNER JOIN sys.columns AS referencedColumn
ON referencedColumn.object_id = fk.referenced_object_id
AND referencedColumn.column_id = fkc.referenced_column_id
WHERE
OBJECT_SCHEMA_NAME(fk.parent_object_id) = ?
AND OBJECT_NAME(fk.parent_object_id) = ?
ORDER BY fk.name;
""",
DW_SCHEMA,
FACT_TABLE,
)
rows = []
for row in cursor.fetchall():
rows.append(
{
"relationship_name": row[0],
"fact_schema": row[1],
"fact_table": row[2],
"fact_column": row[3],
"dimension_schema": row[4],
"dimension_table": row[5],
"dimension_column": row[6],
"cardinality": "Many-to-One",
"filter_direction": "Single",
}
)
return rows
def business_definitions() -> dict[str, dict[str, str]]:
return {
"FactFireInspection": {
"business_name": "Fire Inspection Fact",
"purpose": (
"Stores one analytical row for each normalized "
"fire inspection record."
),
"grain": "One row per source fire inspection record.",
},
"DimDate": {
"business_name": "Date Dimension",
"purpose": (
"Provides reusable calendar attributes for all "
"inspection, invoice, notice, lien, and payment dates."
),
"grain": "One row per calendar date.",
},
"DimInspectionType": {
"business_name": "Inspection Type Dimension",
"purpose": "Classifies inspections by type and description.",
"grain": (
"One row per unique inspection type and description."
),
},
"DimInspectionStatus": {
"business_name": "Inspection Status Dimension",
"purpose": (
"Classifies inspection status and related billing flags."
),
"grain": (
"One row per unique combination of status and billing flags."
),
},
"DimReferralAgency": {
"business_name": "Referral Agency Dimension",
"purpose": (
"Identifies the agency that referred or originated "
"the inspection."
),
"grain": "One row per referral agency.",
},
"DimFireOrganization": {
"business_name": "Fire Organization Dimension",
"purpose": (
"Groups inspections by battalion, station area, "
"and fire-prevention district."
),
"grain": (
"One row per unique fire organizational combination."
),
},
"DimLocation": {
"business_name": "Location Dimension",
"purpose": (
"Provides address, ZIP, neighborhood, supervisor, "
"and police-district attributes."
),
"grain": (
"One row per unique standardized inspection location."
),
},
}
def recommended_measures() -> list[dict[str, str]]:
return [
{
"measure_name": "Total Inspections",
"dax_expression": (
"SUM('FactFireInspection'[InspectionCount])"
),
"format": "#,0",
"business_definition": "Total number of inspection fact rows.",
},
{
"measure_name": "Distinct Inspections",
"dax_expression": (
"DISTINCTCOUNT('FactFireInspection'[InspectionNumber])"
),
"format": "#,0",
"business_definition": (
"Distinct count of inspection numbers."
),
},
{
"measure_name": "Total Invoice Amount",
"dax_expression": (
"SUM('FactFireInspection'[InvoiceAmount])"
),
"format": "$#,0.00",
"business_definition": (
"Total invoice amount associated with inspections."
),
},
{
"measure_name": "Total Paid Amount",
"dax_expression": (
"SUM('FactFireInspection'[PaidAmount])"
),
"format": "$#,0.00",
"business_definition": (
"Total amount paid against inspection invoices."
),
},
{
"measure_name": "Outstanding Balance",
"dax_expression": (
"SUM('FactFireInspection'[OutstandingBalance])"
),
"format": "$#,0.00",
"business_definition": "Remaining balance after payments.",
},
{
"measure_name": "Collection Rate",
"dax_expression": (
"DIVIDE([Total Paid Amount], [Total Invoice Amount])"
),
"format": "0.00%",
"business_definition": (
"Percentage of invoiced amount collected."
),
},
{
"measure_name": "Average Invoice Amount",
"dax_expression": (
"AVERAGE('FactFireInspection'[InvoiceAmount])"
),
"format": "$#,0.00",
"business_definition": (
"Average invoice amount per fact row."
),
},
{
"measure_name": "Average Inspection Duration",
"dax_expression": (
"AVERAGE('FactFireInspection'[InspectionDurationDays])"
),
"format": "0.00",
"business_definition": (
"Average number of days between inspection start and end."
),
},
{
"measure_name": "Average Days to Payment",
"dax_expression": (
"AVERAGE('FactFireInspection'[DaysToPayment])"
),
"format": "0.00",
"business_definition": (
"Average number of days between invoice and payment."
),
},
{
"measure_name": "Inspections with Balance",
"dax_expression": (
"CALCULATE("
"[Total Inspections], "
"'FactFireInspection'[OutstandingBalance] > 0"
")"
),
"format": "#,0",
"business_definition": (
"Inspection rows with a positive outstanding balance."
),
},
{
"measure_name": "Normalization Issue Rows",
"dax_expression": (
"CALCULATE("
"[Total Inspections], "
"'FactFireInspection'[HasNormalizationIssue] = TRUE()"
")"
),
"format": "#,0",
"business_definition": (
"Fact rows that retained a normalization issue flag."
),
},
{
"measure_name": "Inspection Growth %",
"dax_expression": (
"VAR CurrentValue = [Total Inspections] "
"VAR PreviousValue = "
"CALCULATE("
"[Total Inspections], "
"DATEADD('DimDate'[FullDate], -1, YEAR)"
") "
"RETURN "
"DIVIDE(CurrentValue - PreviousValue, PreviousValue)"
),
"format": "0.00%",
"business_definition": "Year-over-year inspection growth.",
},
]
def dashboard_recommendations() -> list[dict[str, str]]:
return [
{
"page_name": "Executive Summary",
"purpose": (
"Provide leadership with the primary inspection "
"and collection KPIs."
),
"recommended_visuals": (
"KPI cards; monthly trend; status distribution; "
"outstanding balance by neighborhood"
),
"recommended_slicers": (
"Year; Quarter; Inspection Type; Inspection Status"
),
},
{
"page_name": "Inspection Trends",
"purpose": (
"Analyze inspection volume and growth over time."
),
"recommended_visuals": (
"Monthly line chart; year-over-year comparison; "
"inspection type trend"
),
"recommended_slicers": (
"Year; Month; Inspection Type; Battalion"
),
},
{
"page_name": "Geographic Analysis",
"purpose": (
"Identify inspection volume and financial exposure "
"by geographic area."
),
"recommended_visuals": (
"Map; neighborhood bar chart; ZIP matrix; "
"supervisor-district summary"
),
"recommended_slicers": (
"Neighborhood; ZIP Code; Supervisor District; "
"Police District"
),
},
{
"page_name": "Operational Performance",
"purpose": (
"Analyze duration, status, station, and battalion performance."
),
"recommended_visuals": (
"Average duration cards; status funnel; "
"battalion comparison; station matrix"
),
"recommended_slicers": (
"Battalion; Station Area; Fire Prevention District; Status"
),
},
{
"page_name": "Collections",
"purpose": (
"Analyze invoicing, payment, collection rate, "
"and outstanding balances."
),
"recommended_visuals": (
"Invoice vs paid trend; collection-rate gauge; "
"outstanding balance table; aging distribution"
),
"recommended_slicers": (
"Invoice Year; Paid Year; Referral Agency; Status"
),
},
{
"page_name": "Referral Agencies",
"purpose": (
"Compare inspection volume and financial outcomes "
"by referral agency."
),
"recommended_visuals": (
"Agency ranking; inspection count; invoice amount; "
"average duration"
),
"recommended_slicers": (
"Referral Agency; Inspection Type; Year"
),
},
{
"page_name": "Inspection Types",
"purpose": (
"Evaluate volume, duration, and financial outcomes "
"by inspection type."
),
"recommended_visuals": (
"Type ranking; duration scatterplot; "
"invoice and payment comparison"
),
"recommended_slicers": (
"Inspection Type; Status; Year; Neighborhood"
),
},
{
"page_name": "Data Quality",
"purpose": (
"Monitor normalization flags and model completeness."
),
"recommended_visuals": (
"Issue-row card; issue trend; affected fields table"
),
"recommended_slicers": (
"Has Normalization Issue; Load Batch; Fact Build Batch"
),
},
]
def write_csv_file(
path: Path,
rows: list[dict[str, Any]],
) -> None:
if not rows:
path.write_text("", encoding="utf-8")
return
with open(
path,
"w",
newline="",
encoding="utf-8-sig",
) as file:
writer = csv.DictWriter(
file,
fieldnames=list(rows[0].keys()),
)
writer.writeheader()
writer.writerows(rows)
def write_html_documentation(
path: Path,
semantic_model: dict[str, Any],
) -> None:
tables_html = ""
for table in semantic_model["tables"]:
tables_html += f"""
<section class="card">
<h2>{table['table_name']}</h2>
<p><strong>Business name:</strong> {table['business_name']}</p>
<p><strong>Purpose:</strong> {table['purpose']}</p>
<p><strong>Grain:</strong> {table['grain']}</p>
</section>
"""
relationships_html = ""
for relationship in semantic_model["relationships"]:
relationships_html += f"""
<tr>
<td>{relationship['fact_table']}</td>
<td>{relationship['fact_column']}</td>
<td>{relationship['dimension_table']}</td>
<td>{relationship['dimension_column']}</td>
<td>{relationship['cardinality']}</td>
</tr>
"""
measures_html = ""
for measure in semantic_model["recommended_measures"]:
measures_html += f"""
<tr>
<td>{measure['measure_name']}</td>
<td><code>{measure['dax_expression']}</code></td>
<td>{measure['format']}</td>
<td>{measure['business_definition']}</td>
</tr>
"""
dashboard_html = ""
for page in semantic_model["dashboard_recommendations"]:
dashboard_html += f"""
<section class="card">
<h3>{page['page_name']}</h3>
<p><strong>Purpose:</strong> {page['purpose']}</p>
<p><strong>Visuals:</strong> {page['recommended_visuals']}</p>
<p><strong>Slicers:</strong> {page['recommended_slicers']}</p>
</section>
"""
html = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AKPE Semantic Layer Documentation</title>
<style>
body {{
margin: 0;
font-family: Arial, Helvetica, sans-serif;
background: #f4f7f5;
color: #1f2d27;
line-height: 1.55;
}}
header {{
background: #173f35;
color: white;
padding: 32px 24px;
}}
main {{
max-width: 1200px;
margin: auto;
padding: 24px;
}}
h1, h2, h3 {{
margin-top: 0;
}}
.card {{
background: white;
border: 1px solid #d9e3de;
border-radius: 12px;
padding: 20px;
margin-bottom: 18px;
box-shadow: 0 4px 14px rgba(0,0,0,.05);
}}
table {{
width: 100%;
border-collapse: collapse;
background: white;
margin-bottom: 24px;
}}
th, td {{
border: 1px solid #d9e3de;
padding: 10px;
text-align: left;
vertical-align: top;
}}
th {{
background: #e8f1ed;
}}
code {{
white-space: pre-wrap;
word-break: break-word;
}}
footer {{
padding: 24px;
text-align: center;
color: #51665d;
}}
</style>
</head>
<body>
<header>
<h1>AKPE Semantic Layer Documentation</h1>
<p>Database: {semantic_model['database']}</p>
<p>Generated at: {semantic_model['generated_at']}</p>
</header>
<main>
<section class="card">
<h2>Model Summary</h2>
<p><strong>Fact table:</strong> {semantic_model['fact_table']}</p>
<p><strong>Dimensions:</strong> {", ".join(semantic_model['dimensions'])}</p>
<p><strong>Semantic schema:</strong> {semantic_model['semantic_schema']}</p>
</section>
<h2>Business Tables</h2>
{tables_html}
<h2>Relationships</h2>
<table>
<thead>
<tr>
<th>Fact Table</th>
<th>Fact Column</th>
<th>Dimension Table</th>
<th>Dimension Column</th>
<th>Cardinality</th>
</tr>
</thead>
<tbody>
{relationships_html}
</tbody>
</table>
<h2>Recommended Power BI Measures</h2>
<table>
<thead>
<tr>
<th>Measure</th>
<th>DAX</th>
<th>Format</th>
<th>Business Definition</th>
</tr>
</thead>
<tbody>
{measures_html}
</tbody>
</table>
<h2>Dashboard Recommendations</h2>
{dashboard_html}
</main>
<footer>
AKPE Semantic Layer Builder
</footer>
</body>
</html>
"""
path.write_text(html, encoding="utf-8")
def export_semantic_package(
connection: pyodbc.Connection,
) -> dict[str, int]:
print_header("STEP 4 — EXPORT SEMANTIC PACKAGE")
OUTPUT_FOLDER.mkdir(
parents=True,
exist_ok=True,
)
columns = read_table_columns(connection)
relationships = read_relationships(connection)
definitions = business_definitions()
measures = recommended_measures()
dashboards = dashboard_recommendations()
table_metadata = []
for table_name, metadata in definitions.items():
table_metadata.append(
{
"schema_name": DW_SCHEMA,
"table_name": table_name,
"business_name": metadata["business_name"],
"purpose": metadata["purpose"],
"grain": metadata["grain"],
}
)
semantic_model = {
"model_name": "Fire Open Data Semantic Model",
"database": DATABASE_NAME,
"fact_table": f"{DW_SCHEMA}.{FACT_TABLE}",
"dimensions": [
f"{DW_SCHEMA}.DimDate",
f"{DW_SCHEMA}.DimInspectionType",
f"{DW_SCHEMA}.DimInspectionStatus",
f"{DW_SCHEMA}.DimReferralAgency",
f"{DW_SCHEMA}.DimFireOrganization",
f"{DW_SCHEMA}.DimLocation",
],
"semantic_schema": MART_SCHEMA,
"semantic_views": [
f"{MART_SCHEMA}.{name}"
for name in VIEW_NAMES
],
"generated_at": datetime.now().isoformat(
timespec="seconds"
),
"tables": table_metadata,
"columns": columns,
"relationships": relationships,
"recommended_measures": measures,
"dashboard_recommendations": dashboards,
}
data_dictionary_file = (
OUTPUT_FOLDER / "business_data_dictionary.csv"
)
relationships_file = (
OUTPUT_FOLDER / "model_relationships.csv"
)
measures_file = (
OUTPUT_FOLDER / "recommended_power_bi_measures.csv"
)
dashboards_file = (
OUTPUT_FOLDER / "dashboard_recommendations.csv"
)
semantic_json_file = (
OUTPUT_FOLDER / "semantic_model.json"
)
html_file = (
OUTPUT_FOLDER / "semantic_model_documentation.html"
)
write_csv_file(data_dictionary_file, columns)
write_csv_file(relationships_file, relationships)
write_csv_file(measures_file, measures)
write_csv_file(dashboards_file, dashboards)
with open(
semantic_json_file,
"w",
encoding="utf-8",
) as file:
json.dump(
semantic_model,
file,
indent=2,
ensure_ascii=False,
default=str,
)
write_html_documentation(
html_file,
semantic_model,
)
print("Generated files:")
print(f"1. {data_dictionary_file}")
print(f"2. {relationships_file}")
print(f"3. {measures_file}")
print(f"4. {dashboards_file}")
print(f"5. {semantic_json_file}")
print(f"6. {html_file}")
return {
"views": len(VIEW_NAMES),
"data_dictionary_rows": len(columns),
"relationship_rows": len(relationships),
"measure_rows": len(measures),
"dashboard_rows": len(dashboards),
}
def validate_views(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 5 — VALIDATE SEMANTIC VIEWS")
cursor = connection.cursor()
for view_name in VIEW_NAMES:
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.VIEWS
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
MART_SCHEMA,
view_name,
)
if cursor.fetchone()[0] != 1:
raise RuntimeError(
f"Semantic view was not created: "
f"{MART_SCHEMA}.{view_name}"
)
print(f"Confirmed: {MART_SCHEMA}.{view_name}")
def main() -> None:
validate_environment()
print_header("AKPE SEMANTIC LAYER BUILDER")
print(f"Server: {SQL_SERVER}")
print(f"Database: {DATABASE_NAME}")
print(f"Data Warehouse schema: {DW_SCHEMA}")
print(f"Semantic schema: {MART_SCHEMA}")
print(f"ODBC driver: {ODBC_DRIVER}")
batch_id = str(uuid.uuid4())
started_at = datetime.now().replace(
microsecond=0
)
connection = connect_to_sql()
try:
validate_required_tables(connection)
create_schemas_and_audit_table(connection)
insert_audit_start(
connection,
batch_id,
started_at,
)
try:
drop_views_if_requested(connection)
create_views(connection)
counts = export_semantic_package(connection)
validate_views(connection)
update_audit_success(
connection,
batch_id,
counts,
)
except Exception as processing_error:
connection.rollback()
update_audit_failure(
connection,
batch_id,
str(processing_error),
)
raise
finally:
connection.close()
print_header("PROCESS COMPLETED SUCCESSFULLY")
print(f"Semantic build batch ID: {batch_id}")
print(f"Views created: {counts['views']}")
print(
f"Business dictionary rows: "
f"{counts['data_dictionary_rows']}"
)
print(
f"Relationships documented: "
f"{counts['relationship_rows']}"
)
print(
f"Recommended measures: "
f"{counts['measure_rows']}"
)
print(
f"Dashboard recommendations: "
f"{counts['dashboard_rows']}"
)
print(
"\nMain semantic model file:\n"
f"{OUTPUT_FOLDER / 'semantic_model.json'}"
)
print(
"\nHTML documentation:\n"
f"{OUTPUT_FOLDER / 'semantic_model_documentation.html'}"
)
print(
"\nNext SQL validation queries:\n"
)
print(
f"""
USE {quote_identifier(DATABASE_NAME)};
GO
SELECT *
FROM {full_name(MART_SCHEMA, "vwExecutiveDashboard")};
GO
SELECT TOP (24) *
FROM {full_name(MART_SCHEMA, "vwMonthlyInspectionTrend")}
ORDER BY YearMonthNumber DESC;
GO
SELECT TOP (20) *
FROM {full_name(MART_SCHEMA, "vwNeighborhoodSummary")}
ORDER BY TotalInspections DESC;
GO
SELECT TOP (20) *
FROM {full_name(MART_SCHEMA, "vwOutstandingBalances")}
ORDER BY OutstandingBalance DESC;
GO
SELECT *
FROM {full_name(AUDIT_SCHEMA, AUDIT_TABLE)}
ORDER BY SemanticLayerBuildLogId DESC;
GO
""".strip()
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nProcess cancelled by the user.")
sys.exit(1)
except Exception as error:
print_header("PROCESS FAILED")
print(type(error).__name__)
print(error)
print(
"\nVerify SQL Server, database, Data Warehouse tables, "
"permissions, object names, and ODBC driver."
)
sys.exit(1)
Validated result: nine semantic views were created, including mart.vwExecutiveDashboard, mart.vwMonthlyInspectionTrend, mart.vwNeighborhoodSummary, and mart.vwOutstandingBalances.
Resultado validado: se crearon nueve vistas semánticas, incluyendo mart.vwExecutiveDashboard, mart.vwMonthlyInspectionTrend, mart.vwNeighborhoodSummary y mart.vwOutstandingBalances.
08_AKPE_PowerBI_Model_Generator.py
Power BI connects directly to the SQL Server fact and dimensions. The final module documents how that model should be implemented, governed, validated, and maintained.
Power BI conecta directamente con la fact y las dimensiones en SQL Server. El módulo final documenta cómo debe implementarse, gobernarse, validarse y mantenerse ese modelo.
The generator inspects the SQL model and produces table definitions, column metadata, relationships, DAX measures, hierarchies, data categories, hidden-column recommendations, sort-by rules, dashboard pages, a checklist, JSON, and HTML documentation.
El generador inspecciona el modelo SQL y produce definiciones de tablas, metadata de columnas, relaciones, medidas DAX, jerarquías, categorías, recomendaciones de columnas ocultas, sort-by, páginas, checklist, JSON y documentación HTML.
from __future__ import annotations
from datetime import datetime
from pathlib import Path
from typing import Any
import csv
import json
import sys
import uuid
import pyodbc
# ============================================================
# CONFIGURATION
# ============================================================
SCRIPT_FOLDER = Path(__file__).resolve().parent
SQL_SERVER = r"JCDCOMPUTER"
DATABASE_NAME = "FireOpenData"
DW_SCHEMA = "dw"
MART_SCHEMA = "mart"
AUDIT_SCHEMA = "audit"
FACT_TABLE = "FactFireInspection"
DATE_TABLE = "DimDate"
AUDIT_TABLE = "PowerBIModelBuildLog"
ODBC_DRIVER = "ODBC Driver 17 for SQL Server"
OUTPUT_FOLDER = SCRIPT_FOLDER / "power_bi_model_output"
POWER_BI_MODEL_NAME = "Fire Open Data Analytics"
# The script creates Power BI model-definition artifacts and documentation.
# It does not create a .pbix file directly.
CREATE_SQL_VALIDATION_VIEWS = True
# ============================================================
# EXPECTED MODEL OBJECTS
# ============================================================
DIMENSION_TABLES = [
"DimDate",
"DimInspectionType",
"DimInspectionStatus",
"DimReferralAgency",
"DimFireOrganization",
"DimLocation",
]
REQUIRED_TABLES = [
(DW_SCHEMA, FACT_TABLE),
*[(DW_SCHEMA, table_name) for table_name in DIMENSION_TABLES],
]
VALIDATION_VIEW_NAMES = [
"vwPowerBIModelRowCounts",
"vwPowerBIModelRelationshipQuality",
"vwPowerBIModelFinancialValidation",
]
# ============================================================
# HELPERS
# ============================================================
def print_header(title: str) -> None:
print("\n" + "=" * 100)
print(title)
print("=" * 100)
def quote_identifier(value: str) -> str:
return "[" + value.replace("]", "]]") + "]"
def full_name(schema_name: str, object_name: str) -> str:
return (
f"{quote_identifier(schema_name)}."
f"{quote_identifier(object_name)}"
)
def connection_string() -> str:
return (
f"DRIVER={{{ODBC_DRIVER}}};"
f"SERVER={SQL_SERVER};"
f"DATABASE={DATABASE_NAME};"
"Trusted_Connection=yes;"
"Encrypt=yes;"
"TrustServerCertificate=yes;"
)
def connect_to_sql() -> pyodbc.Connection:
return pyodbc.connect(
connection_string(),
autocommit=False,
timeout=60,
)
def validate_environment() -> None:
installed_drivers = pyodbc.drivers()
if ODBC_DRIVER not in installed_drivers:
raise RuntimeError(
f"ODBC driver not found: {ODBC_DRIVER}\n"
f"Installed drivers: {installed_drivers}"
)
# ============================================================
# SOURCE VALIDATION
# ============================================================
def validate_required_tables(
connection: pyodbc.Connection,
) -> None:
cursor = connection.cursor()
missing = []
for schema_name, table_name in REQUIRED_TABLES:
cursor.execute(
"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = ?
AND TABLE_NAME = ?;
""",
schema_name,
table_name,
)
if cursor.fetchone()[0] != 1:
missing.append(
f"{schema_name}.{table_name}"
)
if missing:
raise RuntimeError(
"Required Data Warehouse tables are missing:\n"
+ "\n".join(missing)
)
# ============================================================
# AUDIT
# ============================================================
def create_audit_table(
connection: pyodbc.Connection,
) -> None:
print_header("STEP 1 — CREATE POWER BI MODEL AUDIT")
cursor = connection.cursor()
cursor.execute(
f"""
IF NOT EXISTS
(
SELECT 1
FROM sys.schemas
WHERE name = N'{AUDIT_SCHEMA}'
)
BEGIN
EXEC(
'CREATE SCHEMA {quote_identifier(AUDIT_SCHEMA)}'
);
END;
"""
)
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
IF OBJECT_ID(
N'{AUDIT_SCHEMA}.{AUDIT_TABLE}',
N'U'
) IS NULL
BEGIN
CREATE TABLE {audit_table}
(
PowerBIModelBuildLogId BIGINT IDENTITY(1,1)
CONSTRAINT PK_PowerBIModelBuildLog PRIMARY KEY,
PowerBIModelBuildBatchId UNIQUEIDENTIFIER NOT NULL,
ModelName NVARCHAR(255) NOT NULL,
StartedAt DATETIME2(0) NOT NULL,
CompletedAt DATETIME2(0) NULL,
TableCount INT NULL,
RelationshipCount INT NULL,
MeasureCount INT NULL,
HierarchyCount INT NULL,
DataCategoryCount INT NULL,
ValidationViewCount INT NULL,
Status NVARCHAR(30) NOT NULL,
ErrorMessage NVARCHAR(MAX) NULL
);
END;
"""
)
connection.commit()
print(
f"Audit table ready: "
f"{AUDIT_SCHEMA}.{AUDIT_TABLE}"
)
def insert_audit_start(
connection: pyodbc.Connection,
batch_id: str,
started_at: datetime,
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
INSERT INTO {audit_table}
(
PowerBIModelBuildBatchId,
ModelName,
StartedAt,
Status
)
VALUES (?, ?, ?, N'Running');
""",
batch_id,
POWER_BI_MODEL_NAME,
started_at,
)
connection.commit()
def update_audit_success(
connection: pyodbc.Connection,
batch_id: str,
counts: dict[str, int],
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
TableCount = ?,
RelationshipCount = ?,
MeasureCount = ?,
HierarchyCount = ?,
DataCategoryCount = ?,
ValidationViewCount = ?,
Status = N'Completed',
ErrorMessage = NULL
WHERE PowerBIModelBuildBatchId = ?;
""",
counts["tables"],
counts["relationships"],
counts["measures"],
counts["hierarchies"],
counts["data_categories"],
counts["validation_views"],
batch_id,
)
connection.commit()
def update_audit_failure(
connection: pyodbc.Connection,
batch_id: str,
error_message: str,
) -> None:
cursor = connection.cursor()
audit_table = full_name(
AUDIT_SCHEMA,
AUDIT_TABLE,
)
cursor.execute(
f"""
UPDATE {audit_table}
SET
CompletedAt = SYSDATETIME(),
Status = N'Failed',
ErrorMessage = ?
WHERE PowerBIModelBuildBatchId = ?;
""",
error_message,
batch_id,
)
connection.commit()
# ============================================================
# SQL METADATA
# ============================================================
def read_model_columns(
connection: pyodbc.Connection,
) -> list[dict[str, Any]]:
cursor = connection.cursor()
placeholders = ", ".join(
"?" for _ in [FACT_TABLE, *DIMENSION_TABLES]
)
cursor.execute(
f"""
SELECT
c.TABLE_SCHEMA,
c.TABLE_NAME,
c.ORDINAL_POSITION,
c.COLUMN_NAME,
c.DATA_TYPE,
c.CHARACTER_MAXIMUM_LENGTH,
c.NUMERIC_PRECISION,
c.NUMERIC_SCALE,
c.IS_NULLABLE
FROM INFORMATION_SCHEMA.COLUMNS AS c
WHERE c.TABLE_SCHEMA = ?
AND c.TABLE_NAME IN ({placeholders})
ORDER BY
CASE
WHEN c.TABLE_NAME = ? THEN 0
ELSE 1
END,
c.TABLE_NAME,
c.ORDINAL_POSITION;
""",
DW_SCHEMA,
FACT_TABLE,
*DIMENSION_TABLES,
FACT_TABLE,
)
rows = []
for row in cursor.fetchall():
rows.append(
{
"schema_name": row[0],
"table_name": row[1],
"column_position": int(row[2]),
"column_name": row[3],
"sql_data_type": row[4],
"maximum_length": row[5],
"numeric_precision": row[6],
"numeric_scale": row[7],
"is_nullable": row[8],
}
)
return rows
def read_relationships(
connection: pyodbc.Connection,
) -> list[dict[str, Any]]:
cursor = connection.cursor()
cursor.execute(
"""
SELECT
fk.name,
OBJECT_SCHEMA_NAME(fk.parent_object_id),
OBJECT_NAME(fk.parent_object_id),
parentColumn.name,
OBJECT_SCHEMA_NAME(fk.referenced_object_id),
OBJECT_NAME(fk.referenced_object_id),
referencedColumn.name
FROM sys.foreign_keys AS fk
INNER JOIN sys.foreign_key_columns AS fkc
ON fk.object_id = fkc.constraint_object_id
INNER JOIN sys.columns AS parentColumn
ON parentColumn.object_id = fk.parent_object_id
AND parentColumn.column_id = fkc.parent_column_id
INNER JOIN sys.columns AS referencedColumn
ON referencedColumn.object_id = fk.referenced_object_id
AND referencedColumn.column_id = fkc.referenced_column_id
WHERE
OBJECT_SCHEMA_NAME(fk.parent_object_id) = ?
AND OBJECT_NAME(fk.parent_object_id) = ?
ORDER BY fk.name;
""",
DW_SCHEMA,
FACT_TABLE,
)
rows = []
for row in cursor.fetchall():
rows.append(
{
"relationship_name": row[0],
"from_table": row[2],
"from_column": row[3],
"to_table": row[5],
"to_column": row[6],
"cardinality": "Many-to-One",
"cross_filter_direction": "Single",
"is_active": True,
}
)
return rows
# ============================================================
# POWER BI MODEL DEFINITION
# ============================================================
def table_configuration() -> list[dict[str, Any]]:
return [
{
"table_name": FACT_TABLE,
"table_type": "Fact",
"storage_mode": "Import",
"source_schema": DW_SCHEMA,
"source_object": FACT_TABLE,
"hide_in_report_view": False,
"description": (
"Central fact table containing one row per fire "
"inspection record."
),
},
{
"table_name": "DimDate",
"table_type": "Dimension",
"storage_mode": "Import",
"source_schema": DW_SCHEMA,
"source_object": "DimDate",
"hide_in_report_view": False,
"description": (
"Reusable role-playing calendar dimension."
),
},
{
"table_name": "DimInspectionType",
"table_type": "Dimension",
"storage_mode": "Import",
"source_schema": DW_SCHEMA,
"source_object": "DimInspectionType",
"hide_in_report_view": False,
"description": (
"Inspection type and description."
),
},
{
"table_name": "DimInspectionStatus",
"table_type": "Dimension",
"storage_mode": "Import",
"source_schema": DW_SCHEMA,
"source_object": "DimInspectionStatus",
"hide_in_report_view": False,
"description": (
"Inspection status and billing flags."
),
},
{
"table_name": "DimReferralAgency",
"table_type": "Dimension",
"storage_mode": "Import",
"source_schema": DW_SCHEMA,
"source_object": "DimReferralAgency",
"hide_in_report_view": False,
"description": (
"Agency that referred or originated the inspection."
),
},
{
"table_name": "DimFireOrganization",
"table_type": "Dimension",
"storage_mode": "Import",
"source_schema": DW_SCHEMA,
"source_object": "DimFireOrganization",
"hide_in_report_view": False,
"description": (
"Battalion, station area, and fire-prevention district."
),
},
{
"table_name": "DimLocation",
"table_type": "Dimension",
"storage_mode": "Import",
"source_schema": DW_SCHEMA,
"source_object": "DimLocation",
"hide_in_report_view": False,
"description": (
"Address, ZIP code, neighborhood, and district attributes."
),
},
]
def measure_definitions() -> list[dict[str, str]]:
return [
{
"measure_name": "Total Inspections",
"home_table": FACT_TABLE,
"display_folder": "Inspection Volume",
"dax_expression": (
"SUM('FactFireInspection'[InspectionCount])"
),
"format_string": "#,0",
"description": (
"Total number of inspection fact rows."
),
},
{
"measure_name": "Distinct Inspections",
"home_table": FACT_TABLE,
"display_folder": "Inspection Volume",
"dax_expression": (
"DISTINCTCOUNT("
"'FactFireInspection'[InspectionNumber]"
")"
),
"format_string": "#,0",
"description": (
"Distinct count of inspection numbers."
),
},
{
"measure_name": "Total Invoice Amount",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"SUM('FactFireInspection'[InvoiceAmount])"
),
"format_string": "$#,0.00",
"description": (
"Total invoice amount."
),
},
{
"measure_name": "Total Fees",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"SUM('FactFireInspection'[Fee])"
),
"format_string": "$#,0.00",
"description": (
"Total inspection fees."
),
},
{
"measure_name": "Total Penalties",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"SUM('FactFireInspection'[PenaltyAmount])"
),
"format_string": "$#,0.00",
"description": (
"Total penalties."
),
},
{
"measure_name": "Total Interest",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"SUM('FactFireInspection'[InterestAmount])"
),
"format_string": "$#,0.00",
"description": (
"Total interest charged."
),
},
{
"measure_name": "Total Paid Amount",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"SUM('FactFireInspection'[PaidAmount])"
),
"format_string": "$#,0.00",
"description": (
"Total paid amount."
),
},
{
"measure_name": "Outstanding Balance",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"SUM('FactFireInspection'[OutstandingBalance])"
),
"format_string": "$#,0.00",
"description": (
"Total unpaid balance."
),
},
{
"measure_name": "Collection Rate",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"DIVIDE("
"[Total Paid Amount], "
"[Total Invoice Amount]"
")"
),
"format_string": "0.00%",
"description": (
"Paid amount divided by invoice amount."
),
},
{
"measure_name": "Average Invoice Amount",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"AVERAGE('FactFireInspection'[InvoiceAmount])"
),
"format_string": "$#,0.00",
"description": (
"Average invoice amount."
),
},
{
"measure_name": "Average Inspection Duration",
"home_table": FACT_TABLE,
"display_folder": "Operational Performance",
"dax_expression": (
"AVERAGE("
"'FactFireInspection'[InspectionDurationDays]"
")"
),
"format_string": "0.00",
"description": (
"Average inspection duration in days."
),
},
{
"measure_name": "Average Days to Payment",
"home_table": FACT_TABLE,
"display_folder": "Operational Performance",
"dax_expression": (
"AVERAGE("
"'FactFireInspection'[DaysToPayment]"
")"
),
"format_string": "0.00",
"description": (
"Average days between invoice and payment."
),
},
{
"measure_name": "Inspections with Outstanding Balance",
"home_table": FACT_TABLE,
"display_folder": "Financial",
"dax_expression": (
"CALCULATE("
"[Total Inspections], "
"'FactFireInspection'[OutstandingBalance] > 0"
")"
),
"format_string": "#,0",
"description": (
"Inspection rows with positive outstanding balance."
),
},
{
"measure_name": "Normalization Issue Rows",
"home_table": FACT_TABLE,
"display_folder": "Data Quality",
"dax_expression": (
"CALCULATE("
"[Total Inspections], "
"'FactFireInspection'[HasNormalizationIssue] = TRUE()"
")"
),
"format_string": "#,0",
"description": (
"Rows with normalization issues."
),
},
{
"measure_name": "Inspection Growth YoY %",
"home_table": FACT_TABLE,
"display_folder": "Time Intelligence",
"dax_expression": (
"VAR CurrentValue = [Total Inspections] "
"VAR PreviousValue = "
"CALCULATE("
"[Total Inspections], "
"DATEADD('DimDate'[FullDate], -1, YEAR)"
") "
"RETURN "
"DIVIDE(CurrentValue - PreviousValue, PreviousValue)"
),
"format_string": "0.00%",
"description": (
"Year-over-year inspection growth."
),
},
{
"measure_name": "Inspections YTD",
"home_table": FACT_TABLE,
"display_folder": "Time Intelligence",
"dax_expression": (
"TOTALYTD("
"[Total Inspections], "
"'DimDate'[FullDate]"
")"
),
"format_string": "#,0",
"description": (
"Year-to-date inspections."
),
},
{
"measure_name": "Invoice Amount YTD",
"home_table": FACT_TABLE,
"display_folder": "Time Intelligence",
"dax_expression": (
"TOTALYTD("
"[Total Invoice Amount], "
"'DimDate'[FullDate]"
")"
),
"format_string": "$#,0.00",
"description": (
"Year-to-date invoice amount."
),
},
]
def hierarchy_definitions() -> list[dict[str, Any]]:
return [
{
"hierarchy_name": "Calendar Hierarchy",
"table_name": "DimDate",
"levels": [
"CalendarYear",
"QuarterName",
"MonthName",
"FullDate",
],
},
{
"hierarchy_name": "Fire Organization Hierarchy",
"table_name": "DimFireOrganization",
"levels": [
"Battalion",
"StationArea",
"FirePreventionDistrict",
],
},
{
"hierarchy_name": "Geographic Hierarchy",
"table_name": "DimLocation",
"levels": [
"SupervisorDistrict",
"AnalysisNeighborhood",
"ZipCode",
"Address",
],
},
{
"hierarchy_name": "Inspection Type Hierarchy",
"table_name": "DimInspectionType",
"levels": [
"InspectionType",
"InspectionTypeDescription",
],
},
]
def data_category_definitions() -> list[dict[str, str]]:
return [
{
"table_name": "DimLocation",
"column_name": "Address",
"data_category": "Address",
},
{
"table_name": "DimLocation",
"column_name": "ZipCode",
"data_category": "PostalCode",
},
{
"table_name": "DimLocation",
"column_name": "Neighborhood",
"data_category": "Place",
},
{
"table_name": "DimLocation",
"column_name": "AnalysisNeighborhood",
"data_category": "Place",
},
{
"table_name": "DimDate",
"column_name": "FullDate",
"data_category": "Date",
},
]
def sort_by_column_definitions() -> list[dict[str, str]]:
return [
{
"table_name": "DimDate",
"column_name": "MonthName",
"sort_by_column": "CalendarMonth",
},
{
"table_name": "DimDate",
"column_name": "MonthShortName",
"sort_by_column": "CalendarMonth",
},
{
"table_name": "DimDate",
"column_name": "QuarterName",
"sort_by_column": "CalendarQuarter",
},
{
"table_name": "DimDate",
"column_name": "YearMonthLabel",
"sort_by_column": "YearMonthNumber",
},
{
"table_name": "DimDate",
"column_name": "DayOfWeekName",
"sort_by_column": "DayOfWeekNumber",
},
]
def hidden_column_definitions() -> list[dict[str, str]]:
hidden_columns = [
(FACT_TABLE, "FactFireInspectionKey"),
(FACT_TABLE, "FactBuildBatchId"),
(FACT_TABLE, "SourceCleanRowId"),
(FACT_TABLE, "SourceAKPERowId"),
(FACT_TABLE, "SourceLoadBatchId"),
(FACT_TABLE, "NormalizationBatchId"),
(FACT_TABLE, "InspectionTypeKey"),
(FACT_TABLE, "InspectionStatusKey"),
(FACT_TABLE, "ReferralAgencyKey"),
(FACT_TABLE, "FireOrganizationKey"),
(FACT_TABLE, "LocationKey"),
(FACT_TABLE, "InspectionStartDateKey"),
(FACT_TABLE, "InspectionEndDateKey"),
(FACT_TABLE, "ReturnDateKey"),
(FACT_TABLE, "CorrectiveActionDateKey"),
(FACT_TABLE, "InvoiceDateKey"),
(FACT_TABLE, "SecondNoticeDateKey"),
(FACT_TABLE, "FinalNoticeDateKey"),
(FACT_TABLE, "LienDateKey"),
(FACT_TABLE, "PaidDateKey"),
("DimDate", "DateKey"),
("DimInspectionType", "InspectionTypeKey"),
("DimInspectionStatus", "InspectionStatusKey"),
("DimReferralAgency", "ReferralAgencyKey"),
("DimFireOrganization", "FireOrganizationKey"),
("DimLocation", "LocationKey"),
]
return [
{
"table_name": table_name,
"column_name": column_name,
"is_hidden": "True",
}
for table_name, column_name in hidden_columns
]
def dashboard_pages() -> list[dict[str, str]]:
return [
{
"page_name": "Executive Summary",
"page_purpose": (
"Leadership overview of inspection volume, "
"financial performance, and outstanding balances."
),
"recommended_visuals": (
"KPI cards; monthly trend; status bar chart; "
"outstanding balance by neighborhood"
),
"recommended_slicers": (
"Year; Quarter; Inspection Type; Inspection Status"
),
},
{
"page_name": "Inspection Trends",
"page_purpose": (
"Analyze inspection volume over time."
),
"recommended_visuals": (
"Monthly line chart; YoY growth chart; "
"inspection type trend"
),
"recommended_slicers": (
"Year; Month; Inspection Type; Battalion"
),
},
{
"page_name": "Geographic Analysis",
"page_purpose": (
"Analyze inspection activity and financial exposure "
"by location."
),
"recommended_visuals": (
"Map; neighborhood ranking; ZIP matrix; "
"supervisor-district comparison"
),
"recommended_slicers": (
"Neighborhood; ZIP Code; Supervisor District; "
"Police District"
),
},
{
"page_name": "Operational Performance",
"page_purpose": (
"Evaluate duration and operational workload."
),
"recommended_visuals": (
"Average duration cards; battalion matrix; "
"station comparison; status distribution"
),
"recommended_slicers": (
"Battalion; Station Area; Fire Prevention District"
),
},
{
"page_name": "Collections",
"page_purpose": (
"Analyze invoicing, payment, collection rate, "
"and outstanding balances."
),
"recommended_visuals": (
"Invoice versus paid trend; collection-rate KPI; "
"outstanding balance table"
),
"recommended_slicers": (
"Invoice Year; Paid Year; Referral Agency; Status"
),
},
{
"page_name": "Data Quality",
"page_purpose": (
"Monitor normalization and model-quality indicators."
),
"recommended_visuals": (
"Issue rows card; issue percentage; affected records table"
),
"recommended_slicers": (
"Has Normalization Issue"
),
},
]
# ============================================================
# VALIDATION VIEWS
# ============================================================
def create_validation_views(
connection: pyodbc.Connection,
) -> int:
if not CREATE_SQL_VALIDATION_VIEWS:
return 0
print_header("STEP 2 — CREATE POWER BI VALIDATION VIEWS")
cursor = connection.cursor()
for view_name in VALIDATION_VIEW_NAMES:
cursor.execute(
f"""
IF OBJECT_ID(
N'{MART_SCHEMA}.{view_name}',
N'V'
) IS NOT NULL
BEGIN
DROP VIEW {full_name(MART_SCHEMA, view_name)};
END;
"""
)
fact = full_name(DW_SCHEMA, FACT_TABLE)
cursor.execute(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwPowerBIModelRowCounts")}
AS
SELECT N'{FACT_TABLE}' AS ObjectName,
COUNT_BIG(*) AS RowCount
FROM {fact}
UNION ALL
SELECT N'DimDate',
COUNT_BIG(*)
FROM {full_name(DW_SCHEMA, "DimDate")}
UNION ALL
SELECT N'DimInspectionType',
COUNT_BIG(*)
FROM {full_name(DW_SCHEMA, "DimInspectionType")}
UNION ALL
SELECT N'DimInspectionStatus',
COUNT_BIG(*)
FROM {full_name(DW_SCHEMA, "DimInspectionStatus")}
UNION ALL
SELECT N'DimReferralAgency',
COUNT_BIG(*)
FROM {full_name(DW_SCHEMA, "DimReferralAgency")}
UNION ALL
SELECT N'DimFireOrganization',
COUNT_BIG(*)
FROM {full_name(DW_SCHEMA, "DimFireOrganization")}
UNION ALL
SELECT N'DimLocation',
COUNT_BIG(*)
FROM {full_name(DW_SCHEMA, "DimLocation")};
"""
)
cursor.execute(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwPowerBIModelRelationshipQuality")}
AS
SELECT
COUNT_BIG(*) AS FactRows,
SUM(
CASE
WHEN InspectionTypeKey = -1
THEN 1 ELSE 0
END
) AS UnknownInspectionTypeRows,
SUM(
CASE
WHEN InspectionStatusKey = -1
THEN 1 ELSE 0
END
) AS UnknownInspectionStatusRows,
SUM(
CASE
WHEN ReferralAgencyKey = -1
THEN 1 ELSE 0
END
) AS UnknownReferralAgencyRows,
SUM(
CASE
WHEN FireOrganizationKey = -1
THEN 1 ELSE 0
END
) AS UnknownFireOrganizationRows,
SUM(
CASE
WHEN LocationKey = -1
THEN 1 ELSE 0
END
) AS UnknownLocationRows
FROM {fact};
"""
)
cursor.execute(
f"""
CREATE VIEW {full_name(MART_SCHEMA, "vwPowerBIModelFinancialValidation")}
AS
SELECT
SUM(InvoiceAmount) AS TotalInvoiceAmount,
SUM(Fee) AS TotalFee,
SUM(PenaltyAmount) AS TotalPenaltyAmount,
SUM(PostingFee) AS TotalPostingFee,
SUM(InterestAmount) AS TotalInterestAmount,
SUM(PaidAmount) AS TotalPaidAmount,
SUM(OutstandingBalance) AS TotalOutstandingBalance
FROM {fact};
"""
)
connection.commit()
print(
f"Created {len(VALIDATION_VIEW_NAMES)} "
"Power BI validation views."
)
return len(VALIDATION_VIEW_NAMES)
# ============================================================
# EXPORT HELPERS
# ============================================================
def write_csv_file(
path: Path,
rows: list[dict[str, Any]],
) -> None:
if not rows:
path.write_text("", encoding="utf-8")
return
with open(
path,
"w",
newline="",
encoding="utf-8-sig",
) as file:
writer = csv.DictWriter(
file,
fieldnames=list(rows[0].keys()),
)
writer.writeheader()
writer.writerows(rows)
def write_dax_measures_file(
path: Path,
measures: list[dict[str, str]],
) -> None:
lines = [
"// ============================================================",
"// AKPE POWER BI MEASURES",
f"// Model: {POWER_BI_MODEL_NAME}",
"// Copy each measure into Power BI Desktop.",
"// ============================================================",
"",
]
for measure in measures:
lines.extend(
[
f"// Display folder: {measure['display_folder']}",
f"// Format: {measure['format_string']}",
f"// {measure['description']}",
f"{measure['measure_name']} =",
measure["dax_expression"],
"",
]
)
path.write_text(
"\n".join(lines),
encoding="utf-8",
)
def write_power_bi_checklist(
path: Path,
) -> None:
checklist = f"""AKPE POWER BI MODEL IMPLEMENTATION CHECKLIST
============================================================
MODEL
-----
[ ] Connect Power BI Desktop to SQL Server: {SQL_SERVER}
[ ] Select database: {DATABASE_NAME}
[ ] Import the tables from schema: {DW_SCHEMA}
[ ] Load {FACT_TABLE} and all six dimensions
[ ] Use Import mode initially
[ ] Confirm row counts using mart.vwPowerBIModelRowCounts
RELATIONSHIPS
-------------
[ ] Confirm many-to-one relationships from FactFireInspection to each dimension
[ ] Use single-direction filtering from dimensions to fact
[ ] Keep only the InspectionStartDate relationship active for DimDate
[ ] Create inactive role-playing date relationships as required
[ ] Validate unknown-key counts using mart.vwPowerBIModelRelationshipQuality
DATE TABLE
----------
[ ] Mark DimDate as the date table using DimDate[FullDate]
[ ] Create Calendar Hierarchy
[ ] Sort MonthName by CalendarMonth
[ ] Sort QuarterName by CalendarQuarter
[ ] Sort YearMonthLabel by YearMonthNumber
[ ] Sort DayOfWeekName by DayOfWeekNumber
MODEL CLEANUP
-------------
[ ] Hide surrogate keys
[ ] Hide technical batch and source columns
[ ] Disable summarization for identifiers and keys
[ ] Set currency formats for financial fields
[ ] Set percentage format for Collection Rate
[ ] Set geographic data categories
[ ] Create display folders for measures
MEASURES
--------
[ ] Add measures from power_bi_measures.dax
[ ] Place all measures in FactFireInspection
[ ] Validate totals against mart.vwPowerBIModelFinancialValidation
REPORT PAGES
------------
[ ] Executive Summary
[ ] Inspection Trends
[ ] Geographic Analysis
[ ] Operational Performance
[ ] Collections
[ ] Data Quality
FINAL VALIDATION
----------------
[ ] Compare Total Inspections to SQL fact row count
[ ] Compare financial totals to SQL validation view
[ ] Test slicers across all pages
[ ] Test drill-through by Inspection Number
[ ] Confirm no ambiguous relationship paths
[ ] Document refresh ownership and schedule
"""
path.write_text(
checklist,
encoding="utf-8",
)
def write_html_documentation(
path: Path,
model_package: dict[str, Any],
) -> None:
table_rows = "".join(
f"""
<tr>
<td>{item['table_name']}</td>
<td>{item['table_type']}</td>
<td>{item['storage_mode']}</td>
<td>{item['description']}</td>
</tr>
"""
for item in model_package["tables"]
)
relationship_rows = "".join(
f"""
<tr>
<td>{item['from_table']}.{item['from_column']}</td>
<td>{item['to_table']}.{item['to_column']}</td>
<td>{item['cardinality']}</td>
<td>{item['cross_filter_direction']}</td>
</tr>
"""
for item in model_package["relationships"]
)
measure_rows = "".join(
f"""
<tr>
<td>{item['measure_name']}</td>
<td>{item['display_folder']}</td>
<td><code>{item['dax_expression']}</code></td>
<td>{item['format_string']}</td>
</tr>
"""
for item in model_package["measures"]
)
hierarchy_cards = "".join(
f"""
<section class="card">
<h3>{item['hierarchy_name']}</h3>
<p><strong>Table:</strong> {item['table_name']}</p>
<p><strong>Levels:</strong> {" → ".join(item['levels'])}</p>
</section>
"""
for item in model_package["hierarchies"]
)
page_cards = "".join(
f"""
<section class="card">
<h3>{item['page_name']}</h3>
<p><strong>Purpose:</strong> {item['page_purpose']}</p>
<p><strong>Visuals:</strong> {item['recommended_visuals']}</p>
<p><strong>Slicers:</strong> {item['recommended_slicers']}</p>
</section>
"""
for item in model_package["dashboard_pages"]
)
html = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AKPE Power BI Model Documentation</title>
<style>
body {{
margin: 0;
font-family: Arial, Helvetica, sans-serif;
background: #f4f7f5;
color: #1f2d27;
line-height: 1.55;
}}
header {{
background: #173f35;
color: white;
padding: 34px 24px;
}}
main {{
max-width: 1250px;
margin: auto;
padding: 24px;
}}
.card {{
background: white;
border: 1px solid #d9e3de;
border-radius: 12px;
padding: 20px;
margin-bottom: 18px;
box-shadow: 0 4px 14px rgba(0,0,0,.05);
}}
table {{
width: 100%;
border-collapse: collapse;
background: white;
margin-bottom: 28px;
}}
th, td {{
border: 1px solid #d9e3de;
padding: 10px;
text-align: left;
vertical-align: top;
}}
th {{
background: #e8f1ed;
}}
code {{
white-space: pre-wrap;
word-break: break-word;
}}
footer {{
text-align: center;
padding: 24px;
color: #51665d;
}}
</style>
</head>
<body>
<header>
<h1>AKPE Power BI Model Documentation</h1>
<p><strong>Model:</strong> {model_package['model_name']}</p>
<p><strong>Database:</strong> {model_package['database']}</p>
<p><strong>Generated:</strong> {model_package['generated_at']}</p>
</header>
<main>
<section class="card">
<h2>Model Strategy</h2>
<p>
Import-mode star schema with one fact table, six dimensions,
single-direction relationships, role-playing date keys,
hidden technical fields, and centralized DAX measures.
</p>
<p>
This package prepares the implementation artifacts and
documentation. It does not create a PBIX file automatically.
</p>
</section>
<h2>Tables</h2>
<table>
<thead>
<tr>
<th>Table</th>
<th>Type</th>
<th>Storage</th>
<th>Description</th>
</tr>
</thead>
<tbody>{table_rows}</tbody>
</table>
<h2>Relationships</h2>
<table>
<thead>
<tr>
<th>From</th>
<th>To</th>
<th>Cardinality</th>
<th>Filter Direction</th>
</tr>
</thead>
<tbody>{relationship_rows}</tbody>
</table>
<h2>Recommended Measures</h2>
<table>
<thead>
<tr>
<th>Measure</th>
<th>Display Folder</th>
<th>DAX</th>
<th>Format</th>
</tr>
</thead>
<tbody>{measure_rows}</tbody>
</table>
<h2>Hierarchies</h2>
{hierarchy_cards}
<h2>Recommended Report Pages</h2>
{page_cards}
</main>
<footer>
AKPE Power BI Model Generator
</footer>
</body>
</html>
"""
path.write_text(
html,
encoding="utf-8",
)
# ============================================================
# EXPORT MODEL PACKAGE
# ============================================================
def export_model_package(
columns: list[dict[str, Any]],
relationships: list[dict[str, Any]],
) -> dict[str, int]:
print_header("STEP 3 — EXPORT POWER BI MODEL PACKAGE")
OUTPUT_FOLDER.mkdir(
parents=True,
exist_ok=True,
)
tables = table_configuration()
measures = measure_definitions()
hierarchies = hierarchy_definitions()
data_categories = data_category_definitions()
sort_by_columns = sort_by_column_definitions()
hidden_columns = hidden_column_definitions()
pages = dashboard_pages()
model_package = {
"model_name": POWER_BI_MODEL_NAME,
"server": SQL_SERVER,
"database": DATABASE_NAME,
"generated_at": datetime.now().isoformat(
timespec="seconds"
),
"implementation_note": (
"This package contains model-definition artifacts and "
"documentation. It does not generate a PBIX file."
),
"recommended_storage_mode": "Import",
"fact_table": FACT_TABLE,
"date_table": DATE_TABLE,
"tables": tables,
"columns": columns,
"relationships": relationships,
"measures": measures,
"hierarchies": hierarchies,
"data_categories": data_categories,
"sort_by_columns": sort_by_columns,
"hidden_columns": hidden_columns,
"dashboard_pages": pages,
"validation_views": [
f"{MART_SCHEMA}.{name}"
for name in VALIDATION_VIEW_NAMES
],
}
files = {
"model_json":
OUTPUT_FOLDER / "power_bi_model_definition.json",
"tables_csv":
OUTPUT_FOLDER / "power_bi_tables.csv",
"columns_csv":
OUTPUT_FOLDER / "power_bi_columns.csv",
"relationships_csv":
OUTPUT_FOLDER / "power_bi_relationships.csv",
"measures_csv":
OUTPUT_FOLDER / "power_bi_measures.csv",
"measures_dax":
OUTPUT_FOLDER / "power_bi_measures.dax",
"hierarchies_csv":
OUTPUT_FOLDER / "power_bi_hierarchies.csv",
"data_categories_csv":
OUTPUT_FOLDER / "power_bi_data_categories.csv",
"sort_by_csv":
OUTPUT_FOLDER / "power_bi_sort_by_columns.csv",
"hidden_columns_csv":
OUTPUT_FOLDER / "power_bi_hidden_columns.csv",
"dashboard_pages_csv":
OUTPUT_FOLDER / "power_bi_dashboard_pages.csv",
"checklist_txt":
OUTPUT_FOLDER / "power_bi_implementation_checklist.txt",
"documentation_html":
OUTPUT_FOLDER / "power_bi_model_documentation.html",
}
with open(
files["model_json"],
"w",
encoding="utf-8",
) as file:
json.dump(
model_package,
file,
indent=2,
ensure_ascii=False,
default=str,
)
write_csv_file(
files["tables_csv"],
tables,
)
write_csv_file(
files["columns_csv"],
columns,
)
write_csv_file(
files["relationships_csv"],
relationships,
)
write_csv_file(
files["measures_csv"],
measures,
)
write_dax_measures_file(
files["measures_dax"],
measures,
)
flattened_hierarchies = [
{
"hierarchy_name": item["hierarchy_name"],
"table_name": item["table_name"],
"levels": " | ".join(item["levels"]),
}
for item in hierarchies
]
write_csv_file(
files["hierarchies_csv"],
flattened_hierarchies,
)
write_csv_file(
files["data_categories_csv"],
data_categories,
)
write_csv_file(
files["sort_by_csv"],
sort_by_columns,
)
write_csv_file(
files["hidden_columns_csv"],
hidden_columns,
)
write_csv_file(
files["dashboard_pages_csv"],
pages,
)
write_power_bi_checklist(
files["checklist_txt"],
)
write_html_documentation(
files["documentation_html"],
model_package,
)
print("Generated files:")
for position, path in enumerate(
files.values(),
start=1,
):
print(f"{position}. {path}")
return {
"tables": len(tables),
"relationships": len(relationships),
"measures": len(measures),
"hierarchies": len(hierarchies),
"data_categories": len(data_categories),
}
# ============================================================
# MAIN
# ============================================================
def main() -> None:
validate_environment()
print_header(
"AKPE POWER BI MODEL GENERATOR"
)
print(f"Server: {SQL_SERVER}")
print(f"Database: {DATABASE_NAME}")
print(f"Data Warehouse schema: {DW_SCHEMA}")
print(f"Model name: {POWER_BI_MODEL_NAME}")
print(f"ODBC driver: {ODBC_DRIVER}")
batch_id = str(uuid.uuid4())
started_at = datetime.now().replace(
microsecond=0
)
connection = connect_to_sql()
try:
validate_required_tables(connection)
create_audit_table(connection)
insert_audit_start(
connection,
batch_id,
started_at,
)
try:
validation_view_count = (
create_validation_views(
connection
)
)
print_header(
"STEP 3 — READ SQL MODEL METADATA"
)
columns = read_model_columns(
connection
)
relationships = read_relationships(
connection
)
print(
f"Columns documented: "
f"{len(columns):,}"
)
print(
f"Relationships documented: "
f"{len(relationships):,}"
)
counts = export_model_package(
columns,
relationships,
)
counts["validation_views"] = (
validation_view_count
)
update_audit_success(
connection,
batch_id,
counts,
)
except Exception as processing_error:
connection.rollback()
update_audit_failure(
connection,
batch_id,
str(processing_error),
)
raise
finally:
connection.close()
print_header(
"PROCESS COMPLETED SUCCESSFULLY"
)
print(
f"Power BI model build batch ID: "
f"{batch_id}"
)
print(
f"Tables documented: "
f"{counts['tables']}"
)
print(
f"Relationships documented: "
f"{counts['relationships']}"
)
print(
f"Measures generated: "
f"{counts['measures']}"
)
print(
f"Hierarchies generated: "
f"{counts['hierarchies']}"
)
print(
f"Data categories generated: "
f"{counts['data_categories']}"
)
print(
f"Validation views created: "
f"{counts['validation_views']}"
)
print(
"\nMain model definition:\n"
f"{OUTPUT_FOLDER / 'power_bi_model_definition.json'}"
)
print(
"\nDAX measures file:\n"
f"{OUTPUT_FOLDER / 'power_bi_measures.dax'}"
)
print(
"\nImplementation checklist:\n"
f"{OUTPUT_FOLDER / 'power_bi_implementation_checklist.txt'}"
)
print(
"\nImportant note:\n"
"This generator creates the Power BI model package, "
"DAX, metadata, validation views, and documentation. "
"It does not create a PBIX file directly."
)
print(
"\nNext SQL validation queries:\n"
)
print(
f"""
USE {quote_identifier(DATABASE_NAME)};
GO
SELECT *
FROM {full_name(MART_SCHEMA, "vwPowerBIModelRowCounts")};
GO
SELECT *
FROM {full_name(MART_SCHEMA, "vwPowerBIModelRelationshipQuality")};
GO
SELECT *
FROM {full_name(MART_SCHEMA, "vwPowerBIModelFinancialValidation")};
GO
SELECT *
FROM {full_name(AUDIT_SCHEMA, AUDIT_TABLE)}
ORDER BY PowerBIModelBuildLogId DESC;
GO
""".strip()
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(
"\nProcess cancelled by the user."
)
sys.exit(1)
except Exception as error:
print_header(
"PROCESS FAILED"
)
print(type(error).__name__)
print(error)
print(
"\nVerify SQL Server, database, Data Warehouse tables, "
"permissions, object names, and ODBC driver."
)
sys.exit(1)
Validated result: a complete Power BI implementation package was generated, including 17 recommended DAX measures and three SQL validation views. These outputs are documentation; the analytical data source remains SQL Server.
Resultado validado: se generó un paquete completo de implementación Power BI, incluyendo 17 medidas DAX recomendadas y tres vistas SQL de validación. Estas salidas son documentación; la fuente analítica continúa siendo SQL Server.
The eight validated modules produced the following reusable architecture:
Los ocho módulos validados produjeron la siguiente arquitectura reutilizable: