Skip to content

Commit 7c93da2

Browse files
jirhikerjirhiker
committed
feat: enhance NGWMN backfill process with unique constraints and conditional table creation
1 parent a3ce5c9 commit 7c93da2

12 files changed

Lines changed: 910 additions & 65 deletions

‎.github/workflows/CD_staging.yml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,16 @@ jobs:
4747
run: |
4848
uv run alembic upgrade head
4949
50+
- name: Run backfill script on staging database
51+
env:
52+
DB_DRIVER: "cloudsql"
53+
CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}"
54+
CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}"
55+
CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}"
56+
CLOUD_SQL_PASSWORD: "${{ secrets.CLOUD_SQL_PASSWORD }}"
57+
run: |
58+
uv run python transfers/backfill/staging.py
59+
5060
# Uses Google Cloud Secret Manager to store secret credentials
5161
- name: Create app.yaml
5262
run: |

‎alembic/versions/2101e0b029dc_make_location_description_nullable.py‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,12 @@
1414
import sqlalchemy_utils
1515

1616

17+
def _column_exists(bind, table: str, column: str) -> bool:
18+
inspector = sa.inspect(bind)
19+
cols = [c["name"] for c in inspector.get_columns(table)]
20+
return column in cols
21+
22+
1723
# revision identifiers, used by Alembic.
1824
revision: str = "2101e0b029dc"
1925
down_revision: Union[str, Sequence[str], None] = "66ac1af4ba69"
@@ -27,11 +33,20 @@ def upgrade() -> None:
2733
Makes the location.description column nullable to accommodate
2834
legacy data from MS Access that may not have descriptions.
2935
"""
30-
op.alter_column("location", "description", existing_type=sa.String(), nullable=True)
36+
bind = op.get_bind()
37+
if _column_exists(bind, "location", "description"):
38+
op.alter_column(
39+
"location", "description", existing_type=sa.String(), nullable=True
40+
)
41+
else:
42+
# If the column is absent (non-standard schema), skip the alteration.
43+
pass
3144

3245

3346
def downgrade() -> None:
3447
"""Downgrade schema."""
35-
op.alter_column(
36-
"location", "description", existing_type=sa.String(), nullable=False
37-
)
48+
bind = op.get_bind()
49+
if _column_exists(bind, "location", "description"):
50+
op.alter_column(
51+
"location", "description", existing_type=sa.String(), nullable=False
52+
)

‎alembic/versions/7c02d9f8f412_create_nmawaterlevelscontinuouspressuredaily.py‎

Lines changed: 30 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
from alembic import op
1111
import sqlalchemy as sa
12+
from sqlalchemy import inspect
1213

1314
# revision identifiers, used by Alembic.
1415
revision: str = "7c02d9f8f412"
@@ -19,30 +20,36 @@
1920

2021
def upgrade() -> None:
2122
"""Create the legacy daily pressure table used for backfill."""
22-
op.create_table(
23-
"NMA_WaterLevelsContinuous_Pressure_Daily",
24-
sa.Column("GlobalID", sa.String(length=40), primary_key=True),
25-
sa.Column("OBJECTID", sa.Integer(), autoincrement=True, nullable=True),
26-
sa.Column("WellID", sa.String(length=40), nullable=True),
27-
sa.Column("PointID", sa.String(length=50), nullable=True),
28-
sa.Column("DateMeasured", sa.DateTime(), nullable=False),
29-
sa.Column("TemperatureWater", sa.Float(), nullable=True),
30-
sa.Column("WaterHead", sa.Float(), nullable=True),
31-
sa.Column("WaterHeadAdjusted", sa.Float(), nullable=True),
32-
sa.Column("DepthToWaterBGS", sa.Float(), nullable=True),
33-
sa.Column("MeasurementMethod", sa.String(length=2), nullable=True),
34-
sa.Column("DataSource", sa.String(length=5), nullable=True),
35-
sa.Column("MeasuringAgency", sa.String(length=50), nullable=True),
36-
sa.Column("QCed", sa.Boolean(), nullable=True),
37-
sa.Column("Notes", sa.String(length=100), nullable=True),
38-
sa.Column("Created", sa.DateTime(), nullable=False),
39-
sa.Column("Updated", sa.DateTime(), nullable=False),
40-
sa.Column("ProcessedBy", sa.String(length=4), nullable=True),
41-
sa.Column("CheckedBy", sa.String(length=4), nullable=True),
42-
sa.Column("CONDDL (mS/cm)", sa.Float(), nullable=True),
43-
)
23+
bind = op.get_bind()
24+
inspector = inspect(bind)
25+
if not inspector.has_table("NMA_WaterLevelsContinuous_Pressure_Daily"):
26+
op.create_table(
27+
"NMA_WaterLevelsContinuous_Pressure_Daily",
28+
sa.Column("GlobalID", sa.String(length=40), primary_key=True),
29+
sa.Column("OBJECTID", sa.Integer(), autoincrement=True, nullable=True),
30+
sa.Column("WellID", sa.String(length=40), nullable=True),
31+
sa.Column("PointID", sa.String(length=50), nullable=True),
32+
sa.Column("DateMeasured", sa.DateTime(), nullable=False),
33+
sa.Column("TemperatureWater", sa.Float(), nullable=True),
34+
sa.Column("WaterHead", sa.Float(), nullable=True),
35+
sa.Column("WaterHeadAdjusted", sa.Float(), nullable=True),
36+
sa.Column("DepthToWaterBGS", sa.Float(), nullable=True),
37+
sa.Column("MeasurementMethod", sa.String(length=2), nullable=True),
38+
sa.Column("DataSource", sa.String(length=5), nullable=True),
39+
sa.Column("MeasuringAgency", sa.String(length=50), nullable=True),
40+
sa.Column("QCed", sa.Boolean(), nullable=True),
41+
sa.Column("Notes", sa.String(length=100), nullable=True),
42+
sa.Column("Created", sa.DateTime(), nullable=False),
43+
sa.Column("Updated", sa.DateTime(), nullable=False),
44+
sa.Column("ProcessedBy", sa.String(length=4), nullable=True),
45+
sa.Column("CheckedBy", sa.String(length=4), nullable=True),
46+
sa.Column("CONDDL (mS/cm)", sa.Float(), nullable=True),
47+
)
4448

4549

4650
def downgrade() -> None:
4751
"""Drop the legacy daily pressure table."""
48-
op.drop_table("NMA_WaterLevelsContinuous_Pressure_Daily")
52+
bind = op.get_bind()
53+
inspector = inspect(bind)
54+
if inspector.has_table("NMA_WaterLevelsContinuous_Pressure_Daily"):
55+
op.drop_table("NMA_WaterLevelsContinuous_Pressure_Daily")

‎alembic/versions/8a1de3e3f0b3_add_ngwmn_unique_constraints.py‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
"""Add unique constraints for NGWMN backfill upserts
2+
3+
Revision ID: 8a1de3e3f0b3
4+
Revises: 9c0f061c8322
5+
Create Date: 2026-02-10 00:10:00.000000
6+
"""
7+
8+
from typing import Sequence, Union
9+
10+
from alembic import op
11+
import sqlalchemy as sa
12+
13+
# revision identifiers, used by Alembic.
14+
revision: str = "8a1de3e3f0b3"
15+
down_revision: Union[str, Sequence[str], None] = "9c0f061c8322"
16+
branch_labels: Union[str, Sequence[str], None] = None
17+
depends_on: Union[str, Sequence[str], None] = None
18+
19+
20+
def upgrade() -> None:
21+
"""Add unique constraints to support ON CONFLICT upserts."""
22+
op.create_unique_constraint(
23+
"uq_nma_view_ngwmn_waterlevels_point_date",
24+
"NMA_view_NGWMN_WaterLevels",
25+
["PointID", "DateMeasured"],
26+
)
27+
op.create_unique_constraint(
28+
"uq_nma_view_ngwmn_wellconstruction_point_casing_screen",
29+
"NMA_view_NGWMN_WellConstruction",
30+
["PointID", "CasingTop", "ScreenTop"],
31+
)
32+
op.create_unique_constraint(
33+
"uq_nma_view_ngwmn_lithology_objectid",
34+
"NMA_view_NGWMN_Lithology",
35+
["OBJECTID"],
36+
)
37+
38+
39+
def downgrade() -> None:
40+
"""Drop unique constraints."""
41+
op.drop_constraint(
42+
"uq_nma_view_ngwmn_lithology_objectid",
43+
"NMA_view_NGWMN_Lithology",
44+
type_="unique",
45+
)
46+
op.drop_constraint(
47+
"uq_nma_view_ngwmn_wellconstruction_point_casing_screen",
48+
"NMA_view_NGWMN_WellConstruction",
49+
type_="unique",
50+
)
51+
op.drop_constraint(
52+
"uq_nma_view_ngwmn_waterlevels_point_date",
53+
"NMA_view_NGWMN_WaterLevels",
54+
type_="unique",
55+
)

‎alembic/versions/9c0f061c8322_create_ngwmn_legacy_views.py‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
"""Create legacy NGWMN view tables
2+
3+
Revision ID: 9c0f061c8322
4+
Revises: 7c02d9f8f412
5+
Create Date: 2026-02-10 00:00:00.000000
6+
"""
7+
8+
from typing import Sequence, Union
9+
10+
from alembic import op
11+
import sqlalchemy as sa
12+
from sqlalchemy import inspect
13+
14+
# revision identifiers, used by Alembic.
15+
revision: str = "9c0f061c8322"
16+
down_revision: Union[str, Sequence[str], None] = "7c02d9f8f412"
17+
branch_labels: Union[str, Sequence[str], None] = None
18+
depends_on: Union[str, Sequence[str], None] = None
19+
20+
21+
def upgrade() -> None:
22+
"""Create the three NGWMN legacy view tables."""
23+
bind = op.get_bind()
24+
inspector = inspect(bind)
25+
26+
if not inspector.has_table("NMA_view_NGWMN_WellConstruction"):
27+
op.create_table(
28+
"NMA_view_NGWMN_WellConstruction",
29+
sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
30+
sa.Column("PointID", sa.String(length=50), nullable=True),
31+
sa.Column("CasingTop", sa.Float(), nullable=True),
32+
sa.Column("CasingBottom", sa.Float(), nullable=True),
33+
sa.Column("CasingDepthUnits", sa.String(length=20), nullable=True),
34+
sa.Column("ScreenTop", sa.Float(), nullable=True),
35+
sa.Column("ScreenBottom", sa.Float(), nullable=True),
36+
sa.Column("ScreenBottomUnit", sa.String(length=20), nullable=True),
37+
sa.Column("ScreenDescription", sa.String(length=250), nullable=True),
38+
sa.Column("CasingDescription", sa.String(length=250), nullable=True),
39+
)
40+
41+
if not inspector.has_table("NMA_view_NGWMN_WaterLevels"):
42+
op.create_table(
43+
"NMA_view_NGWMN_WaterLevels",
44+
sa.Column("PointID", sa.String(length=50), primary_key=True),
45+
sa.Column("DateMeasured", sa.Date(), primary_key=True),
46+
sa.Column("DepthToWaterBGS", sa.Float(), nullable=True),
47+
sa.Column("WLUnits", sa.String(length=10), nullable=True),
48+
sa.Column("MeasurementMethod", sa.String(length=50), nullable=True),
49+
sa.Column("WLAccuracy", sa.Float(), nullable=True),
50+
sa.Column("PublicRelease", sa.Boolean(), nullable=True),
51+
)
52+
53+
if not inspector.has_table("NMA_view_NGWMN_Lithology"):
54+
op.create_table(
55+
"NMA_view_NGWMN_Lithology",
56+
sa.Column("OBJECTID", sa.Integer(), primary_key=True, autoincrement=True),
57+
sa.Column("PointID", sa.String(length=50), nullable=True),
58+
sa.Column("Lithology", sa.String(length=50), nullable=True),
59+
sa.Column("TERM", sa.String(length=100), nullable=True),
60+
sa.Column("StratSource", sa.String(length=100), nullable=True),
61+
sa.Column("StratTop", sa.Float(), nullable=True),
62+
sa.Column("StratTopUnit", sa.String(length=20), nullable=True),
63+
sa.Column("StratBottom", sa.Float(), nullable=True),
64+
sa.Column("StratBottomUnit", sa.String(length=20), nullable=True),
65+
)
66+
67+
68+
def downgrade() -> None:
69+
"""Drop the NGWMN legacy view tables."""
70+
bind = op.get_bind()
71+
inspector = inspect(bind)
72+
73+
for table in (
74+
"NMA_view_NGWMN_Lithology",
75+
"NMA_view_NGWMN_WaterLevels",
76+
"NMA_view_NGWMN_WellConstruction",
77+
):
78+
if inspector.has_table(table):
79+
op.drop_table(table)

‎api/ngwmn.py‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
# ===============================================================================
2+
# Copyright 2023 Jake Ross
3+
#
4+
# Licensed under the Apache License, Version 2.0 (the "License");
5+
# you may not use this file except in compliance with the License.
6+
# You may obtain a copy of the License at
7+
#
8+
# http://www.apache.org/licenses/LICENSE-2.0
9+
#
10+
# Unless required by applicable law or agreed to in writing, software
11+
# distributed under the License is distributed on an "AS IS" BASIS,
12+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
# See the License for the specific language governing permissions and
14+
# limitations under the License.
15+
# ===============================================================================
16+
from fastapi import APIRouter
17+
from starlette.responses import Response
18+
19+
from core.dependencies import session_dependency
20+
from services.ngwmn_helper import make_waterlevels_response, make_well_construction_response, make_lithology_response
21+
22+
router = APIRouter(prefix="/ngwmn", tags=["NGWMN"])
23+
24+
25+
@router.get(
26+
"/waterlevels/{pointid}",
27+
summary="Get waterlevels for a given pointid in the NGWMN format",
28+
)
29+
async def read_ngwmn_waterlevels(pointid: str, db: session_dependency):
30+
data = make_waterlevels_response(pointid, db)
31+
return Response(content=data, media_type="application/xml")
32+
33+
34+
@router.get(
35+
"/wellconstruction/{pointid}",
36+
summary="Get wellconstruction for a given pointid in the NGWMN format",
37+
)
38+
async def read_ngwmn_wellconstruction(pointid: str, db: session_dependency):
39+
data = make_well_construction_response(pointid, db)
40+
return Response(content=data, media_type="application/xml")
41+
42+
43+
@router.get(
44+
"/lithology/{pointid}",
45+
summary="Get lithology for a given pointid in the NGWMN format",
46+
)
47+
async def read_ngwmn_lithology(pointid: str, db: session_dependency):
48+
data = make_lithology_response(pointid, db)
49+
return Response(content=data, media_type="application/xml")
50+
51+
52+
# ============= EOF =============================================

‎core/initializers.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,7 @@ def register_routes(app):
123123
from api.asset import router as asset_router
124124
from api.search import router as search_router
125125
from api.geospatial import router as geospatial_router
126+
from api.ngwmn import router as ngwmn_router
126127

127128
app.include_router(asset_router)
128129
app.include_router(author_router)
@@ -137,6 +138,7 @@ def register_routes(app):
137138
app.include_router(sensor_router)
138139
app.include_router(search_router)
139140
app.include_router(thing_router)
141+
app.include_router(ngwmn_router)
140142
add_pagination(app)
141143

142144

0 commit comments

Comments
 (0)