Skip to content
Merged
Show file tree
Hide file tree
Changes from 10 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
139 changes: 139 additions & 0 deletions python/waterdata_client/scripts/build_constants.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
"""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
$ git clone git@github.com:NOAA-OWP/hydrotools.git
$ cd hydrotools/python/waterdata_client
$ python3 -m venv env
$ source env/bin/activate
(env) $ python3 -m pip install -U pip wheel
(env) $ pip install -e .[develop]
(env) $ python3 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="-") # NOTE: "-" means stdout
@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