Containerization 3.0

.gitignore

@@ -16,9 +16,14 @@ venv/
 backend/Data/Parcels/
 
 #DuckLake files
+
+# DuckLake catalogue
 backend/Data/lake
 backend/Data/lake.files/
 
+# DuckDB Database with CLEAN tables
+backend/Data/warehouse.duckdb
+
 # Local planning/notes files
 claude_todo.md
 claude-work-done.md

backend/ETL/dockerfile.clean

@@ -22,7 +22,7 @@ COPY --chown=app:app ./api /code/api
 COPY --chown=app:app ./logger /code/logger
 # TODO: Remove the dependency on app_utils
 COPY --chown=app:app ./app_utils /code/app_utils
-COPY --chown=app:app ./datastore /code/datastore
+COPY --chown=app:app ./lake_build.py /code/lake_build.py
 
 
 ENV DATA_DIR=/data

backend/ETL/dockerfile.collect

@@ -21,7 +21,7 @@ COPY --chown=app:app ./run_data_collection.py /code/run_data_collection.py
 # TODO: Remove the dependency on app_utils
 COPY --chown=app:app ./app_utils /code/app_utils
 
-COPY --chown=app:app ./datastore /code/datastore
+COPY --chown=app:app ./lake_build.py /code/lake_build.py
 
 
 ENV DATA_DIR=/data

backend/ETL/dockerfile.lake

@@ -0,0 +1,24 @@
+FROM python:3.12-slim-trixie
+COPY --from=ghcr.io/astral-sh/uv:0.11.21 /uv /uvx /bin/
+
+# Set working directory
+WORKDIR /code
+RUN useradd -m -u 1000 app && chown app:app /code /home/app
+USER app
+
+# Dependencies
+COPY --chown=app:app pyproject.toml uv.lock /code/
+RUN --mount=type=cache,target=/home/app/.cache/uv,uid=1000,gid=1000 \
+    uv sync --frozen --no-dev --no-install-project
+
+# Spatial extension installation
+RUN uv run python -c "import duckdb; duckdb.connect().execute('INSTALL spatial')"
+
+# The source data collection scripts (source --> destination)
+COPY --chown=app:app ./lake_build.py /code/lake_build.py
+
+ENV DATA_DIR=/data
+ENV PYTHONPATH=/code
+
+# Command to run the DuckLake builder script (container entrypoint) 
+ENTRYPOINT ["uv", "run", "python", "-u", "-m", "lake_build"]

backend/ETL/dockerfile.load

@@ -19,7 +19,7 @@ COPY --chown=app:app ./run_data_loading.py /code/run_data_loading.py
 # TODO: Remove the dependency on app_utils
 COPY --chown=app:app ./app_utils /code/app_utils
 # For lake connection import
-COPY --chown=app:app ./datastore /code/datastore
+COPY --chown=app:app ./lake_build.py /code/lake_build.py
 
 
 ENV DATA_DIR=/data

backend/data_cleaning/clean_cdc.py

@@ -14,7 +14,7 @@
 from sklearn.decomposition import PCA
 
 from build.core_functions import bin_measures
-from datastore.lake_build import con
+from lake_build import con
 
 # Columns we'd like excluded from the cleaned tables, IF they exist on that
 # particular RAW table. Tract- and county-level releases don't always share

backend/data_cleaning/clean_demographics.py

@@ -12,7 +12,7 @@
 
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_dependency_ratio.py

@@ -18,7 +18,7 @@
 import numpy as np
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_derived_time_series.py

@@ -29,7 +29,7 @@
 import numpy as np
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 UNAVAILABLE_SENTINEL = -666666666.0
 

backend/data_cleaning/clean_economic.py

@@ -12,7 +12,7 @@
 
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_education.py

@@ -12,7 +12,7 @@
 
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_flood.py

@@ -9,7 +9,7 @@
 python -m ETL.data_cleaning.clean_flood
 """
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 ## LOAD SPATIAL EXTENSION FUNCTION --------------------

backend/data_cleaning/clean_health_insurance_coverage.py

@@ -13,7 +13,7 @@
 import numpy as np
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_historic_population.py

@@ -11,7 +11,7 @@
 
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_housing.py

@@ -12,7 +12,7 @@
 
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_qcew.py

@@ -12,7 +12,7 @@
 
 import pandas as pd
 
-from datastore.lake_build import con
+from lake_build import con
 
 
 def read_raw_data() -> pd.DataFrame:

backend/data_cleaning/clean_wastewater.py

@@ -10,7 +10,7 @@
 python -m data_cleaning.clean_wastewater
 """
 
-from datastore.lake_build import con
+from lake_build import con
 
 # TODO: Path is useful when sql files are created!
 # from build import BACKEND

backend/data_cleaning/clean_zoning.py

@@ -14,7 +14,7 @@
 import pandas as pd
 
 from app_utils.sql_render import render_sql
-from datastore.lake_build import con
+from lake_build import con
 
 SQL_PATH = Path(__file__).resolve().parent / "sql"
 # Town and zoning-district boundaries were digitised separately, so subtracting

backend/data_collection/acs5.py

@@ -11,6 +11,7 @@
 Use --append to merge new rows into existing files instead of overwriting.
 """
 
+import os
 import time
 
 import pandas as pd
@@ -19,9 +20,10 @@
 from app_utils.census import tidy_census
 from data_collection.base import ALL_GEOS
 
-API_KEY = (
-    "29af5488bbdb8c7d9f67b7f4ff9c9151e8c2bd0a"  # TODO: Get this as a .env variable!!!
-)
+# Define API key through the .env file
+API_KEY = os.environ.get("CENSUS_API_KEY")
+
+
 BASE_URL = "https://api.census.gov/data/{year}/acs/acs5/profile"
 STATE_FIPS = "50"  # Vermont
 TABLES = {
@@ -30,10 +32,11 @@
     "DP04": "Housing",
     "DP05": "Demographic",
 }
-YEARS = list(range(2009, 2025))
 STORAGE_LOCATION = "Data/Census/ACS_5"
 ID_VARS = ["year", "geo_type", "table", "NAME", "state", "county"]
 
+YEARS = range(2009, 2025)
+
 # Default geos list in (label, for_clause, in_clause) format
 GEOS = [(k, *v) for k, v in ALL_GEOS.items()]
 
@@ -70,21 +73,25 @@ def fetch_table(year, table, for_clause, in_clause):
         return None
 
 
-def run_acs5_scrape(year: int = 2024, geos: list = GEOS, append: bool = False):
+def run_acs5_scrape(years: range = YEARS, geos: list = GEOS, append: bool = False):
+    """
+    Collect ACS 5-year profile tables for multiple years.
+    """
     # Collect raw frames per table
     all_frames = {table: [] for table in TABLES}
 
-    print(f"\n=== {year} ===")
-    for geo_label, for_clause, in_clause in geos:
-        for table in TABLES:
-            print(f"  {table} / {geo_label}...")
-            df = fetch_table(year, table, for_clause, in_clause)
-            if df is not None:
-                df["geo_type"] = geo_label
-                all_frames[table].append(df)
-            time.sleep(0.1)
-
-    # Save wide + tidy per table
+    for year in years:
+        print(f"\n=== {year} ===")
+        for geo_label, for_clause, in_clause in geos:
+            for table in TABLES:
+                print(f"  {table} / {geo_label}...")
+                df = fetch_table(year, table, for_clause, in_clause)
+                if df is not None:
+                    df["geo_type"] = geo_label
+                    all_frames[table].append(df)
+                time.sleep(0.1)
+
+        # Save wide + tidy per table
     results = {}
     for table, frames in all_frames.items():
         if not frames:
@@ -106,7 +113,7 @@ def run_acs5_scrape(year: int = 2024, geos: list = GEOS, append: bool = False):
         # wide_csv_path = f"{STORAGE_LOCATION}/{title}.csv"
 
         if append:
-            new_names = set(combined["NAME"].unique())
+            new_names = set(combined["year", "geo_type", "NAME"].unique())
             # --- Wide ---
             try:
                 existing_wide = pd.read_parquet(wide_parquet_path)
@@ -123,31 +130,41 @@ def run_acs5_scrape(year: int = 2024, geos: list = GEOS, append: bool = False):
         # print(f"Saved wide: {title} ({len(combined):,} rows)")
 
         # Tidy: run per-year so column labels are year-accurate
-        tidy_frames = []
+        # Tidy: run per-year so column labels are year-accurate
+    tidy_frames = []
+
+    for year in sorted(combined["year"].unique()):
         year_df = combined[combined["year"] == year]
-        if not year_df.empty:
-            try:
-                tidy_year = tidy_census(year_df, year=year, id_vars=ID_VARS)
-                tidy_year["table"] = table
-                tidy_frames.append(tidy_year)
-            except Exception as e:
-                print(f"  SKIP tidy {year} / {table}: {e}")
 
-        if tidy_frames:
-            tidy = pd.concat(tidy_frames, ignore_index=True)
+        if year_df.empty:
+            continue
+
+        try:
+            tidy_year = tidy_census(
+                year_df,
+                year=year,
+                id_vars=ID_VARS,
+            )
+            tidy_year["table"] = table
+            tidy_frames.append(tidy_year)
+
+        except Exception as e:
+            print(f"  SKIP tidy {year} / {table}: {e}")
 
-            label = TABLES[table]
+    if tidy_frames:
+        tidy = pd.concat(tidy_frames, ignore_index=True)
 
-            results[f"acs5_{label.lower()}"] = tidy
-            # tidy_parquet_path = f"{STORAGE_LOCATION}/{title}_tidy.parquet"
-            # tidy_csv_path = f"{STORAGE_LOCATION}/{title}_tidy.csv"
+        label = TABLES[table]
+        results[f"acs5_{label.lower()}"] = tidy
+        # tidy_parquet_path = f"{STORAGE_LOCATION}/{title}_tidy.parquet"
+        # tidy_csv_path = f"{STORAGE_LOCATION}/{title}_tidy.csv"
 
-            # No separate append needed for tidy: it's derived from the
-            # already-merged wide frame, so it naturally contains all geos.
+        # No separate append needed for tidy: it's derived from the
+        # already-merged wide frame, so it naturally contains all geos.
 
-            # tidy.to_csv(tidy_csv_path, index=False)
-            # tidy.to_parquet(tidy_parquet_path, index=False)
-            # print(f"Saved tidy: {title}_tidy ({len(tidy):,} rows)")
+        # tidy.to_csv(tidy_csv_path, index=False)
+        # tidy.to_parquet(tidy_parquet_path, index=False)
+        # print(f"Saved tidy: {title}_tidy ({len(tidy):,} rows)")
 
     return results
 
@@ -171,13 +188,13 @@ def merge_tidy_tables():
     return
 
 
-def collect(year: int = 2024, geos=GEOS, append=False):
+def collect(years: range = YEARS, geos=GEOS, append=False):
     """
     Collect ACS profile tables and return tidy datasets.
     """
 
     tables = run_acs5_scrape(
-        year=year,
+        years=years,
         geos=geos,
         append=append,
     )

backend/data_collection/base.py

@@ -21,6 +21,7 @@
 replaces rather than duplicates those rows), then writes the merged result.
 """
 
+import os
 import time
 from dataclasses import dataclass
 
@@ -29,10 +30,10 @@
 
 from app_utils.census import split_name_col
 
-API_KEY = "29af5488bbdb8c7d9f67b7f4ff9c9151e8c2bd0a"
+# Define API key through the .env file
+API_KEY = os.environ.get("CENSUS_API_KEY")
 BASE_URL = "https://api.census.gov/data/{year}/acs/acs5"
 STATE_FIPS = "50"
-# YEARS = list(range(2009, 2025))
 STORAGE_LOCATION = "Data/Census/ACS_5"
 
 # ---------------------------------------------------------------------------