Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion python/waterdata_client/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ dynamic = ["version"]
version = { attr = "hydrotools.waterdata_client._version.__version__" }

[project.optional-dependencies]
develop = ["pytest", "pytest-aiohttp"]
develop = ["pytest", "pytest-aiohttp", "Jinja2", "click"]
env = ["dotenv"]

[project.urls]
Expand Down
136 changes: 136 additions & 0 deletions python/waterdata_client/scripts/build_constants.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
"""Run this script to generate and write the `constants.py` file using the Jinja2
template and the "items" collections found in the USGS OGC API schema.

# Usage
The following assumes you are developing in an UNIX-like environment and using a
Python virtual environment called "env".

```bash
(env) $ git clone git@github.com:NOAA-OWP/hydrotools.git
Comment thread
jarq6c marked this conversation as resolved.
Outdated
(env) $ cd hydrotools/python/waterdata_client
(env) $ pip install -e .[develop]
(env) $ python scripts/build_constants.py ./templates/ --output src/hydrotools/waterdata_client/constants.py
```
"""
import keyword
from pathlib import Path
import re
from typing import Any
from datetime import datetime, UTC
import click
from jinja2 import Environment, FileSystemLoader
from hydrotools.waterdata_client.schema import get_schema
from hydrotools.waterdata_client.client_config import SETTINGS
from hydrotools.waterdata_client._version import __version__

def get_template_data(
schema: dict[str, Any],
ignore_errors: bool = False
) -> list[dict[str, str]]:
"""Extracts collection metadata for the constants template.

Args:
schema: Deserialized dict derived from USGS OGC API schema.
ignore_errors: If True, skips collections with invalid Python identifier
characters. If False, raises.

Returns:
List of extracted mappings (dict) from enum members in screaming
SNAKE_CASE to enum values for use with Jinja2 StrEnum building
template.

Raises:
SyntaxError if unable to translate collection label to valid Python
identifier.
"""
collections = []
paths = schema.get("paths", {})

for path in paths.keys():
# Match pattern: /collections/{collectionId}/items
match = re.search(r"/collections/(?P<cid>[^/]+)/items$", path)
Comment thread
jarq6c marked this conversation as resolved.
Outdated
if match:
cid = match.group("cid")
enum_member = cid.upper().replace("-", "_")

# Validate identifier
if not enum_member.isidentifier() or keyword.iskeyword(enum_member):
if ignore_errors:
continue
raise SyntaxError(f"{enum_member} is not a valid identifier")
collections.append({
"enum_member": enum_member,
"value": cid
})

# Return sorted by value for a deterministic file
return sorted(collections, key=lambda x: x["value"])

@click.command()
Comment thread
jarq6c marked this conversation as resolved.
@click.argument("templates", type=click.Path(exists=True, file_okay=False,
dir_okay=True, path_type=Path))
@click.option("-n", "--name", nargs=1, type=str, default="constants.py.j2")
@click.option("-o", "--output", nargs=1, type=click.Path(
exists=False, file_okay=True, dir_okay=False, path_type=Path, allow_dash=True),
help="Output file path", default="-")
Comment thread
jarq6c marked this conversation as resolved.
Outdated
@click.option('--overwrite/--no-overwrite', default=False,
help="Overwrite existing file, disabled by default")
@click.option('--ignore-errors/--no-ignore-errors', default=False,
help="Overwrite existing file, disabled by default")
def write_constants_module(
templates: Path,
name: str,
output: Path,
overwrite: bool = False,
ignore_errors: bool = False
) -> None:
"""Renders the constants.py file from the OGC schema.

\b
Args:
templates: File system directory containing Jinja2 template files.
name: Template file name. Defaults to 'constants.py.j2'.
output: The location to write the resulting file. Defaults to stdout.
overwrite: If true, overwrite the file if it exists. Defaults to false.
ignore_errors: If True, skips collections with invalid Python identifier
characters. If False, raises.

\b
Raises:
FileExistsError: If the file exists and overwrite is False.
"""
# Check for file
if output.exists() and not overwrite:
raise FileExistsError(f"{output} already exists")

# Resolve schema
schema = get_schema()
template_data = get_template_data(schema, ignore_errors=ignore_errors)

# Metadata
timestamp = datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S Z")

# Setup Jinja2
env = Environment(
loader=FileSystemLoader(templates),
trim_blocks=True,
lstrip_blocks=True
)

# Render and save
template = env.get_template(name=name)
content = template.render(
timestamp=timestamp,
schema_source=str(SETTINGS.schema_url),
schema_version=schema.get("info", {}).get("version", "UNKNOWN"),
openapi_version=schema.get("openapi", "UNKNOWN"),
collections=template_data,
package_version=__version__,
script_name=Path(__file__).name
)
with click.open_file(output, "w", encoding="utf-8") as fo:
fo.write(content)

if __name__ == "__main__":
# pylint: disable=no-value-for-parameter
write_constants_module()
186 changes: 186 additions & 0 deletions python/waterdata_client/scripts/test_build_constants.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,186 @@
"""Test the build_constants stand-alone script."""
import pytest
from unittest.mock import patch
from click.testing import CliRunner
from typing import Any
from build_constants import get_template_data, write_constants_module

@pytest.fixture
def mock_template() -> str:
"""Mock test template."""
return """
from enum import StrEnum

class USGSCollection(StrEnum):
{% for item in collections %}
{{ item.enum_member }} = "{{ item.value }}"
{% endfor %}

"""

@pytest.fixture
def mock_schema() -> dict[str, Any]:
"""Mock test schema."""
return {
"paths": {
"/collections/monitoring-locations/items": {"get": {}},
"/collections/daily/items": {"get": {}},
"/conformance": {"get": {}}
}
}

@pytest.fixture
def bad_schema_special() -> dict[str, Any]:
"""Bad test schema with special characters."""
return {
"paths": {
"/collections/@@$$%%/items": {"get": {}},
"/collections/daily/items": {"get": {}},
"/conformance": {"get": {}}
}
}

@pytest.fixture
def bad_schema_digits() -> dict[str, Any]:
"""Bad test schema with initial digit."""
return {
"paths": {
"/collections/123/items": {"get": {}},
"/collections/daily/items": {"get": {}},
"/conformance": {"get": {}}
}
}

def test_get_template_data(mock_schema):
"""Verify extraction of collections."""
data = get_template_data(mock_schema)

assert len(data) == 2
assert data[0]["value"] == "daily"
assert data[1]["enum_member"] == "MONITORING_LOCATIONS"

def test_cli_output_to_stdout(mock_schema, mock_template, tmp_path):
"""Verify CLI."""
# Setup template file
template_directory = tmp_path / "templates"
template_directory.mkdir()
template_name = "constants.py.j2"
template_file = template_directory / template_name
template_file.write_text(mock_template)

# Patch and run CLI
with patch("build_constants.get_schema") as mock_get_schema:
mock_get_schema.return_value = mock_schema

runner = CliRunner()
result = runner.invoke(
write_constants_module,
[str(template_directory), "--name", template_name]
)

assert result.exit_code == 0
assert "class USGSCollection" in result.output
assert "daily" in result.output
assert "MONITORING_LOCATIONS" in result.output

def test_cli_overwrite_error(mock_schema, mock_template, tmp_path):
"""Verify CLI."""
# Setup template file
template_directory = tmp_path / "templates"
template_directory.mkdir()
template_name = "constants.py.j2"
template_file = template_directory / template_name
template_file.write_text(mock_template)
output_file = template_directory / "constant.py"
output_file.touch()

# Patch and run CLI
with patch("build_constants.get_schema") as mock_get_schema:
mock_get_schema.return_value = mock_schema

runner = CliRunner()
result = runner.invoke(
write_constants_module,
[
str(template_directory),
"--name",
template_name,
"--output",
str(output_file)
]
)

assert result.exit_code == 1
assert isinstance(result.exception, FileExistsError)

def test_cli_overwrite(mock_schema, mock_template, tmp_path):
"""Verify CLI."""
# Setup template file
template_directory = tmp_path / "templates"
template_directory.mkdir()
template_name = "constants.py.j2"
template_file = template_directory / template_name
template_file.write_text(mock_template)
output_file = template_directory / "constant.py"
output_file.touch()

# Patch and run CLI
with patch("build_constants.get_schema") as mock_get_schema:
mock_get_schema.return_value = mock_schema

runner = CliRunner()
result = runner.invoke(
write_constants_module,
[
str(template_directory),
"--name",
template_name,
"--output",
str(output_file),
"--overwrite"
]
)

assert result.exit_code == 0

def test_bad_schema_special(bad_schema_special):
"""Verify raises SyntaxError."""
with pytest.raises(SyntaxError):
data = get_template_data(bad_schema_special)

def test_bad_schema_digits(bad_schema_digits):
"""Verify raises SyntaxError."""
with pytest.raises(SyntaxError):
data = get_template_data(bad_schema_digits)

def test_ignore_bad_schema(bad_schema_special):
"""Verify extraction of collections."""
data = get_template_data(bad_schema_special, ignore_errors=True)

assert len(data) == 1
assert data[0]["value"] == "daily"
assert data[0]["enum_member"] == "DAILY"

def test_cli_ignore_errors(bad_schema_digits, mock_template, tmp_path):
"""Verify CLI."""
# Setup template file
template_directory = tmp_path / "templates"
template_directory.mkdir()
template_name = "constants.py.j2"
template_file = template_directory / template_name
template_file.write_text(mock_template)

# Patch and run CLI
with patch("build_constants.get_schema") as mock_get_schema:
mock_get_schema.return_value = bad_schema_digits

runner = CliRunner()
result = runner.invoke(
write_constants_module,
[str(template_directory), "--name", template_name, "--ignore-errors"],
)

assert result.exit_code == 0
assert "class USGSCollection" in result.output
assert "daily" in result.output
assert "DAILY" in result.output
Original file line number Diff line number Diff line change
@@ -1,4 +1,15 @@
"""Package-wide constants."""
# AUTO-GENERATED FILE. DO NOT EDIT MANUALLY.
"""This is a Jinja2 auto-generated module. This module contains enums
and constants used package-wide. The `USGSCollection` StrEnum is generated
by inspecting the USGS OGC API JSON schema and identifying all "items" endpoints.

Package version: 0.1.0
Generation script: build_constants.py
Generated: 2026-04-15 17:50:47 Z
JSON Schema source: https://api.waterdata.usgs.gov/ogcapi/v0/openapi?f=json
JSON Schema version: 0.44.0
OpenAPI version: 3.0.2
"""
from enum import StrEnum

class OGCAPI(StrEnum):
Expand All @@ -16,7 +27,7 @@ class OGCPATH(StrEnum):
SCHEMA = "schema"

class USGSCollection(StrEnum):
"""USGS OGC API Collectins."""
"""USGS OGC API Collections."""
AGENCY_CODES = "agency-codes"
ALTITUDE_DATUMS = "altitude-datums"
AQUIFER_CODES = "aquifer-codes"
Expand All @@ -28,9 +39,10 @@ class USGSCollection(StrEnum):
COORDINATE_DATUM_CODES = "coordinate-datum-codes"
COORDINATE_METHOD_CODES = "coordinate-method-codes"
COUNTIES = "counties"
COUNTRIES = "countries"
DAILY = "daily"
FIELD_MEASUREMENTS_METADATA = "field-measurements-metadata"
FIELD_MEASUREMENTS = "field-measurements"
FIELD_MEASUREMENTS_METADATA = "field-measurements-metadata"
HYDROLOGIC_UNIT_CODES = "hydrologic-unit-codes"
LATEST_CONTINUOUS = "latest-continuous"
LATEST_DAILY = "latest-daily"
Expand Down
Loading