diff --git a/.agents/skills/code-review/SKILL.md b/.agents/skills/code-review/SKILL.md new file mode 100644 index 0000000..ef9612d --- /dev/null +++ b/.agents/skills/code-review/SKILL.md @@ -0,0 +1,101 @@ +--- +name: code-review +description: Only use this skill when explicitly asked to conduct a code review for a piece of code or a pull request. +--- + +Below is a set of different specific code review rules that you should check when asked for a review. Only check these rules and nothing else. Open-ended code reviews are not covered by this skill. + +When conducting a code review, read the diff and check it against each rule below. Read relevant pieces of code if needed for understanding the context around the diff. Each rule consists of the slug of the rule and textual description of what specifically should be checked by the rule. + +Your response should be a structured response in JSON form that you write in `.vscode/code-review.json` file in this repository. + +JSON should have strictly the following form: + +```json +[ + { + "name": "typos", + "file": "path/from/the/root/of/the/repo/file.py", + "line_from": 10, + "line_to": 20, + "description": "Word 'asembly' is written with a typo, you likely meant 'assembly'" + }, + { + "name": "incorrect-comments", + "file": "path/from/the/root/of/the/repo/another.py", + "line_from": 15, + "line_to": 21, + "description": "The comment says that this algorithm works in O(N) but most common path according to numpy documentation is O(N^2)" + } +] +``` + +You MUST adhere to this form because it will be later used by automation to create a user-friendly UI. + +If the rule asks for a citation, put it in the `description` field. + +Below are rules you should check when reviewing the code. + +### duplicate-code + +The code written is an obvious copy-pasta of the other part of the code in the diff or in other parts of the codebase. This rule only includes code that is an actual copy and paste of another part of the codebase or is very trivially extractable into functions. + +### incorrect-comments + +The comment used for a function, variable, or expression contradicts the content of the token it describes. This also includes docstrings or any other kind of documentation directly inside the code. Token names are not included in this rule. + +### misleading-name + +The name of a function, method, variable, parameter, or class contradicts what it actually does. Examples of contradictions: a get_* or fetch_* function that mutates state, an is_*/has_* name that does not return a boolean, a singular name bound to a collection, a boolean flag whose name implies the opposite polarity of the behavior it controls, or a verb that names a different operation than the one performed. Only flag when the mismatch is visible in the body of the token in the diff or in the code the diff calls. Do not flag names that are merely vague, short, or abbreviated. + +### typos + +A comment, token, or any other part of the code contains a typo. + +### obvious-comment + +A new or changed comment only restates what the next line or block already makes obvious, and adds no non-obvious rationale, constraint, or decision. Do not flag comments that explain why, trade-offs, invariants, or external constraints. + +### stale-reference + +The diff renames, moves, or deletes a symbol, module, file, CLI flag, config key, environment variable, or API field, but references to the old name survive elsewhere in the repository, in places the type checker does not see: comments, docstrings, error and log message strings, README or docs pages, YAML or TOML config, makefile targets, or test names. Only flag when you have located the surviving old reference and can cite its file and line. + +### caller-impact + +The diff changes the observable contract of a function used outside the diff, without updating those callers. Contract changes that count: the meaning or unit of a return value, None versus empty collection, ordering guarantees, which exception type is raised, whether an input argument is mutated, or the value of a default parameter. Only flag when at least one caller outside the diff is identified by file and line and would behave differently under the new contract. Do not flag changes the type checker would already reject. + +### unit-mismatch + +A quantity is produced in one unit, scale, or coordinate convention and consumed as if it were in another: degrees versus radians, seconds versus milliseconds, bytes versus kilobytes, zero-based versus one-based indices, or a value converted to a target unit in one branch but not another. Only flag when the source unit is determinable from the code, a column definition, a constant name, or a unit-bearing type in the diff or its immediate context. + +### swallowed-error + +An except clause catches an exception and then neither re-raises it, nor logs it, nor forwards it to the caller or user through the project's reporting mechanism, so the failure becomes invisible. Only flag when the handler body discards the exception entirely (pass, bare return, or assigning a fallback with no record of the failure). Do not flag handlers at documented process boundaries that log or report the exception, and do not flag handlers whose enclosing function's contract is explicitly to tolerate the failure. + +### inconsistent-duplicate + +The diff modifies one instance of code that is duplicated elsewhere in the repository while leaving the other instances unchanged, so the copies now disagree. Only flag when you have located the near-identical counterpart and can cite its file and line, and when the divergence affects behavior rather than only formatting. + +### generic-exception + +The code raises a built-in exception such as Exception, RuntimeError, or ValueError in a situation for which the project already defines a more specific exception class, or raises an exception belonging to a different layer than the one it is raised from. Only flag when the more appropriate exception class exists in the repository and is used for comparable situations elsewhere, cited by file and line. + +### layer-violation + +The diff introduces an import that inverts the dependency direction already established between two packages: module A imports package B, while B already imports A's package. Only flag when you can cite an existing import establishing the opposite direction. Also flag imports of code the project treats as generated or vendored being edited or wrapped in a way that a regeneration step would overwrite. + +### test-name-mismatch + +The name of a test describes a behavior, input, or outcome different from what the test body actually sets up and asserts. Only flag when the discrepancy is in the substance of the test, not in wording or abbreviation. + +### misleading-user-text + +Text that reaches an end user contradicts the code that produces or consumes it: a form field title or description, CLI help text, a task description, or an error message. + +### unmanaged-resource + +A resource that requires deterministic release — database connection or cursor, transaction, file, thread, executor, temporary directory, network client — is acquired in the diff without a with block or an equivalent guaranteed-cleanup path, so it leaks when an exception is raised. Only flag when the project provides a context manager or cleanup helper for that resource, cited by file and line, or when the same resource is acquired with a with block elsewhere in the codebase. + +### call-in-loop + +A database query, HTTP request, or other heavy I/O call is issued once per item inside a loop where the project already provides a batched or bulk equivalent. Only flag when the batched helper exists, cited by file and line, and the loop bound is data-dependent rather than a small fixed number. diff --git a/makefile b/makefile index e892802..7a86d9b 100644 --- a/makefile +++ b/makefile @@ -118,6 +118,7 @@ frontend: dev: @set -e; \ + rm -rf uploader/static; \ trap 'kill $$backend_pid $$frontend_pid 2>/dev/null || true; wait $$backend_pid $$frontend_pid 2>/dev/null || true' INT TERM EXIT; \ $(MAKE) --no-print-directory serve & \ backend_pid=$$!; \ diff --git a/tests/test_designation_upload.py b/tests/test_designation_upload.py index 659677b..cbf4396 100644 --- a/tests/test_designation_upload.py +++ b/tests/test_designation_upload.py @@ -18,7 +18,7 @@ def _mock_client() -> Mock: @patch("uploader.app.structured.designations.upload.rawdata_batches") -@patch("uploader.app.structured.designations.upload._fetch_column_units") +@patch("uploader.app.structured.designations.upload.fetch_column_units") def test_simple_column_expression( mock_fetch_column_units: Mock, mock_rawdata_batches: Mock, @@ -43,7 +43,7 @@ def test_simple_column_expression( @patch("uploader.app.structured.designations.upload.rawdata_batches") -@patch("uploader.app.structured.designations.upload._fetch_column_units") +@patch("uploader.app.structured.designations.upload.fetch_column_units") def test_composed_string_expression( mock_fetch_column_units: Mock, mock_rawdata_batches: Mock, @@ -65,7 +65,7 @@ def test_composed_string_expression( assert total == 1 -@patch("uploader.app.structured.designations.upload._fetch_column_units") +@patch("uploader.app.structured.designations.upload.fetch_column_units") def test_missing_referenced_columns(mock_fetch_column_units: Mock) -> None: mock_fetch_column_units.return_value = ({"other"}, {}) @@ -81,7 +81,7 @@ def test_missing_referenced_columns(mock_fetch_column_units: Mock) -> None: @patch("uploader.app.structured.designations.upload.rawdata_batches") -@patch("uploader.app.structured.designations.upload._fetch_column_units") +@patch("uploader.app.structured.designations.upload.fetch_column_units") def test_null_referenced_values_counted_as_unmatched( mock_fetch_column_units: Mock, mock_rawdata_batches: Mock, @@ -104,7 +104,7 @@ def test_null_referenced_values_counted_as_unmatched( @patch("uploader.app.structured.designations.upload.rawdata_batches") -@patch("uploader.app.structured.designations.upload._fetch_column_units") +@patch("uploader.app.structured.designations.upload.fetch_column_units") def test_expression_evaluation_error_becomes_runtime_error( mock_fetch_column_units: Mock, mock_rawdata_batches: Mock, @@ -126,7 +126,7 @@ def test_expression_evaluation_error_becomes_runtime_error( @patch("uploader.app.structured.designations.upload.rawdata_batches") -@patch("uploader.app.structured.designations.upload._fetch_column_units") +@patch("uploader.app.structured.designations.upload.fetch_column_units") def test_output_file_writes_id_designation_pairs( mock_fetch_column_units: Mock, mock_rawdata_batches: Mock, diff --git a/tests/test_geometry_upload.py b/tests/test_geometry_upload.py index 1504736..f4d75a2 100644 --- a/tests/test_geometry_upload.py +++ b/tests/test_geometry_upload.py @@ -30,7 +30,7 @@ def _base_expressions() -> dict[str, str]: @patch("uploader.app.structured.geometry.upload.rawdata_batches") -@patch("uploader.app.structured.geometry.upload._fetch_column_units") +@patch("uploader.app.structured.geometry.upload.fetch_column_units") def test_isophote_unit_conversion_error_includes_field_details( mock_fetch_column_units: Mock, mock_rawdata_batches: Mock, @@ -79,7 +79,7 @@ def test_isophote_unit_conversion_error_includes_field_details( @patch("uploader.app.structured.geometry.upload.rawdata_batches") -@patch("uploader.app.structured.geometry.upload._fetch_column_units") +@patch("uploader.app.structured.geometry.upload.fetch_column_units") def test_constant_isophote_unit_error_omits_empty_columns( mock_fetch_column_units: Mock, mock_rawdata_batches: Mock, diff --git a/tests/test_structured_catalog.py b/tests/test_structured_catalog.py new file mode 100644 index 0000000..80a28a9 --- /dev/null +++ b/tests/test_structured_catalog.py @@ -0,0 +1,188 @@ +from unittest.mock import Mock, patch + +import pytest + +from uploader.app.lib import expression, formula +from uploader.app.structured.generic.upload import is_numeric_datatype, upload_catalog_columns +from uploader.clients.gen.client import adminapi +from uploader.clients.gen.client.adminapi.models.catalog_field import CatalogField +from uploader.clients.gen.client.adminapi.models.catalog_schema import CatalogSchema +from uploader.clients.gen.client.adminapi.models.datatype_enum import DatatypeEnum +from uploader.forms.structured_catalog import ( + _task_description, + build_catalog_form, + register_structured_catalog_tasks, +) +from uploader.tasks import TASKS + + +def _mock_storage(total: int = 1) -> Mock: + storage = Mock() + storage.query.return_value = [{"cnt": total}] + return storage + + +def _mock_client() -> Mock: + return Mock(spec=adminapi.AuthenticatedClient) + + +def _sample_schema() -> CatalogSchema: + return CatalogSchema( + catalog="demo", + title="Demo catalog", + description="Demo structured catalog.", + fields=[ + CatalogField(name="label", data_type=DatatypeEnum.STRING, description="Object label"), + CatalogField(name="mag", data_type=DatatypeEnum.FLOAT, unit="mag", description="Magnitude"), + CatalogField(name="n", data_type=DatatypeEnum.INTEGER, required=False), + ], + ) + + +def test_is_numeric_datatype() -> None: + assert is_numeric_datatype(DatatypeEnum.FLOAT) + assert is_numeric_datatype(DatatypeEnum.INTEGER) + assert not is_numeric_datatype(DatatypeEnum.STRING) + assert not is_numeric_datatype(DatatypeEnum.TIMESTAMP_WITHOUT_TIME_ZONE) + + +def test_build_catalog_form_numeric_fields_are_expressions() -> None: + form_model = build_catalog_form(_sample_schema()) + fields = form_model.model_fields + + assert fields["label"].title == "label column" + assert fields["label"].description == "Object label" + + assert fields["mag"].title == "mag" + assert fields["mag"].description == "Expression. Magnitude" + + assert fields["n"].title == "n" + assert fields["n"].description == "Expression. Value for n." + assert fields["n"].is_required() is False + + +def test_task_description_matches_upload_expression_engine() -> None: + description = _task_description(_sample_schema()) + assert "Demo structured catalog." in description + assert expression.expression_syntax_help() in description + assert formula.expression_syntax_help() not in description + assert "str(x)" not in description + assert "String literals and + concatenation are supported." not in description + assert "Modulo divisors must carry units" not in description + + +@patch("uploader.forms.structured_catalog.log.logger.warning") +@patch("uploader.forms.structured_catalog.fetch_catalogs", side_effect=RuntimeError("api down")) +def test_register_structured_catalog_tasks_logs_fetch_failure( + _mock_fetch_catalogs: Mock, + mock_warning: Mock, +) -> None: + before = set(TASKS) + register_structured_catalog_tasks() + assert set(TASKS) == before + mock_warning.assert_called_once() + assert mock_warning.call_args.args[0] == "failed to fetch catalogs for structured catalog tasks" + assert mock_warning.call_args.kwargs["error"] == "api down" + + +@patch("uploader.app.structured.generic.upload.handle_call") +@patch("uploader.app.structured.generic.upload.save_structured_data.sync_detailed") +@patch("uploader.app.structured.generic.upload.rawdata_batches") +@patch("uploader.app.structured.generic.upload.fetch_column_units") +def test_upload_evaluates_numeric_expressions( + mock_fetch_column_units: Mock, + mock_rawdata_batches: Mock, + mock_sync_detailed: Mock, + mock_handle_call: Mock, +) -> None: + mock_fetch_column_units.return_value = ( + {"name", "e_ra", "count"}, + {"name": "", "e_ra": "arcsec", "count": ""}, + ) + mock_rawdata_batches.return_value = iter( + [ + [ + { + "hyperleda_internal_id": "000079ce-5ffd-82c6-3f75-3a083f0fde80", + "name": "NGC 1", + "e_ra": 0.5, + "count": 3, + }, + ], + ], + ) + + total = upload_catalog_columns( + _mock_storage(), + "test_table", + "demo", + column_map={"label": "name"}, + expressions={ + "mag": 'col("e_ra")', + "n": 'col("count") * 2', + }, + field_types={ + "label": DatatypeEnum.STRING, + "mag": DatatypeEnum.FLOAT, + "n": DatatypeEnum.INTEGER, + }, + field_units={"mag": "arcsec"}, + columns=["label", "mag", "n"], + batch_size=100, + client=_mock_client(), + write=True, + report_func=lambda _: None, + ) + + assert total == 1 + mock_rawdata_batches.assert_called_once() + fetch_columns = mock_rawdata_batches.call_args.args[2] + assert set(fetch_columns) == {"name", "e_ra", "count"} + body = mock_sync_detailed.call_args.kwargs["body"] + assert body.catalog == "demo" + assert body.columns == ["label", "mag", "n"] + assert body.ids == ["000079ce-5ffd-82c6-3f75-3a083f0fde80"] + assert body.data == [["NGC 1", 0.5, 6]] + + +@patch("uploader.app.structured.generic.upload.rawdata_batches") +@patch("uploader.app.structured.generic.upload.fetch_column_units") +def test_upload_unit_conversion_error_includes_field_details( + mock_fetch_column_units: Mock, + mock_rawdata_batches: Mock, +) -> None: + mock_fetch_column_units.return_value = ( + {"bt"}, + {"bt": ""}, + ) + mock_rawdata_batches.return_value = iter( + [ + [ + { + "hyperleda_internal_id": "000079ce-5ffd-82c6-3f75-3a083f0fde80", + "bt": 12.5, + }, + ], + ], + ) + + with pytest.raises(RuntimeError, match="failed to evaluate expressions for row") as exc_info: + upload_catalog_columns( + _mock_storage(), + "test_table", + "demo", + column_map={}, + expressions={"mag": 'col("bt")'}, + field_types={"mag": DatatypeEnum.FLOAT}, + field_units={"mag": "arcsec"}, + columns=["mag"], + batch_size=100, + client=_mock_client(), + write=False, + report_func=lambda _: None, + ) + + message = str(exc_info.value) + assert "mag" in message + assert 'col("bt")' in message + assert "arcsec" in message diff --git a/uploader/app/catalogs.py b/uploader/app/catalogs.py new file mode 100644 index 0000000..804811f --- /dev/null +++ b/uploader/app/catalogs.py @@ -0,0 +1,11 @@ +from uploader.app.endpoints import env_map +from uploader.app.upload import handle_call +from uploader.clients.gen.client import adminapi +from uploader.clients.gen.client.adminapi.api.default import get_catalogs +from uploader.clients.gen.client.adminapi.models.catalog_schema import CatalogSchema + + +def fetch_catalogs(*, base_url: str | None = None) -> list[CatalogSchema]: + client = adminapi.Client(base_url=base_url or env_map["prod"]) + response = handle_call(get_catalogs.sync_detailed(client=client)) + return list(response.data.catalogs) diff --git a/uploader/app/lib/expression.py b/uploader/app/lib/expression.py index ac65b15..1185959 100644 --- a/uploader/app/lib/expression.py +++ b/uploader/app/lib/expression.py @@ -111,6 +111,44 @@ def evaluate(self, values: dict[str, float], units: dict[str, str]) -> u.Quantit return _Evaluator(values, units).visit(self._tree.body) +def format_unit(unit: u.UnitBase) -> str: + text = f"{unit:s}".strip() + return text if text else "dimensionless" + + +def eval_context_suffix(expr: Expression, column_units: dict[str, str]) -> str: + if not expr.referenced_columns: + return "" + parts = [f"{col}={column_units.get(col, '')!r}" for col in sorted(expr.referenced_columns)] + return f"; columns: {', '.join(parts)}" + + +def evaluate_to_float( + expr: Expression, + values: dict[str, float], + column_units: dict[str, str], + field: str, + source: str, + target_unit: str, +) -> float: + try: + quantity = expr.evaluate(values, column_units) + except (ValueError, u.UnitConversionError, u.UnitTypeError) as e: + raise RuntimeError( + f"failed to evaluate {field!r} ({source!r}){eval_context_suffix(expr, column_units)}: {e}", + ) from e + try: + if target_unit: + return float(quantity.to(u.Unit(target_unit)).value) + return float(quantity.to(u.dimensionless_unscaled).value) + except (u.UnitConversionError, u.UnitTypeError) as e: + raise RuntimeError( + f"failed to convert {field!r} ({source!r}) " + f"from {format_unit(quantity.unit)} to {target_unit or 'dimensionless'}" + f"{eval_context_suffix(expr, column_units)}: {e}", + ) from e + + def parse(source: str) -> Expression: tree = ast.parse(source.strip(), mode="eval") referenced_columns = _collect_columns(tree.body) diff --git a/uploader/app/lib/table.py b/uploader/app/lib/table.py new file mode 100644 index 0000000..e7d7a41 --- /dev/null +++ b/uploader/app/lib/table.py @@ -0,0 +1,27 @@ +from uploader.app.upload import handle_call +from uploader.clients.gen.client import adminapi +from uploader.clients.gen.client.adminapi.api.default import get_table + + +def fetch_column_units( + client: adminapi.AuthenticatedClient, + table_name: str, +) -> tuple[set[str], dict[str, str]]: + resp = handle_call(get_table.sync_detailed(client=client, table_name=table_name)) + column_names: set[str] = set() + column_units: dict[str, str] = {} + for col in resp.data.column_info: + column_names.add(col.name) + if isinstance(col.unit, str): + column_units[col.name] = col.unit + return column_names, column_units + + +def validate_columns( + table_name: str, + needed_cols: set[str], + column_names: set[str], +) -> None: + missing = sorted(col for col in needed_cols if col not in column_names) + if missing: + raise RuntimeError(f"Table {table_name} has no column(s): {missing}") diff --git a/uploader/app/structured/designations/upload.py b/uploader/app/structured/designations/upload.py index 0c99ee7..1cf265b 100644 --- a/uploader/app/structured/designations/upload.py +++ b/uploader/app/structured/designations/upload.py @@ -21,11 +21,12 @@ parse, ) from uploader.app.lib.rawdata import rawdata_batches +from uploader.app.lib.table import fetch_column_units, validate_columns from uploader.app.storage import PgStorage from uploader.app.structured.designations.rules import RULES, match from uploader.app.upload import handle_call from uploader.clients.gen.client import adminapi -from uploader.clients.gen.client.adminapi.api.default import get_table, save_structured_data +from uploader.clients.gen.client.adminapi.api.default import save_structured_data from uploader.clients.gen.client.adminapi.models.save_structured_data_request import ( SaveStructuredDataRequest, ) @@ -127,30 +128,6 @@ def pct(n: int) -> float: report_func(report.DoneEvent(message=summary)) -def _fetch_column_units( - client: adminapi.AuthenticatedClient, - table_name: str, -) -> tuple[set[str], dict[str, str]]: - resp = handle_call(get_table.sync_detailed(client=client, table_name=table_name)) - column_names: set[str] = set() - column_units: dict[str, str] = {} - for col in resp.data.column_info: - column_names.add(col.name) - if isinstance(col.unit, str): - column_units[col.name] = col.unit - return column_names, column_units - - -def _validate_columns( - table_name: str, - needed_cols: set[str], - column_names: set[str], -) -> None: - missing = sorted(col for col in needed_cols if col not in column_names) - if missing: - raise RuntimeError(f"Table {table_name} has no column(s): {missing}") - - def _build_column_values( row: dict[str, Any], referenced_columns: frozenset[str], @@ -214,8 +191,8 @@ def upload_designations( ) -> int: parsed = parse(expression) needed_cols = set(parsed.referenced_columns) - column_names, column_units = _fetch_column_units(client, table_name) - _validate_columns(table_name, needed_cols, column_names) + column_names, column_units = fetch_column_units(client, table_name) + validate_columns(table_name, needed_cols, column_names) rule_counts: dict[str, int] = {r.name: 0 for r in RULES} unmatched = 0 diff --git a/uploader/app/structured/generic/__init__.py b/uploader/app/structured/generic/__init__.py new file mode 100644 index 0000000..866e3d4 --- /dev/null +++ b/uploader/app/structured/generic/__init__.py @@ -0,0 +1,3 @@ +from uploader.app.structured.generic.upload import is_numeric_datatype, upload_catalog_columns + +__all__ = ["is_numeric_datatype", "upload_catalog_columns"] diff --git a/uploader/app/structured/generic/upload.py b/uploader/app/structured/generic/upload.py new file mode 100644 index 0000000..d03645d --- /dev/null +++ b/uploader/app/structured/generic/upload.py @@ -0,0 +1,192 @@ +from collections.abc import Callable, Mapping, Sequence +from typing import Any + +from psycopg import sql + +import uploader.app.action_description as action_description +import uploader.app.report as report +from uploader.app.display import format_table +from uploader.app.lib.expression import Expression, evaluate_to_float, parse +from uploader.app.lib.rawdata import rawdata_batches +from uploader.app.lib.table import fetch_column_units, validate_columns +from uploader.app.storage import PgStorage +from uploader.app.upload import handle_call +from uploader.clients.gen.client import adminapi +from uploader.clients.gen.client.adminapi.api.default import save_structured_data +from uploader.clients.gen.client.adminapi.models.datatype_enum import DatatypeEnum +from uploader.clients.gen.client.adminapi.models.save_structured_data_request import ( + SaveStructuredDataRequest, +) +from uploader.clients.gen.client.adminapi.models.save_structured_data_request_units import ( + SaveStructuredDataRequestUnits, +) +from uploader.clients.gen.client.adminapi.types import UNSET + +_FLOAT_TYPES = frozenset( + { + DatatypeEnum.DOUBLE, + DatatypeEnum.DOUBLE_PRECISION, + DatatypeEnum.FLOAT, + DatatypeEnum.NUMBER, + DatatypeEnum.REAL, + } +) +_INT_TYPES = frozenset( + { + DatatypeEnum.INT, + DatatypeEnum.INTEGER, + DatatypeEnum.LONG, + DatatypeEnum.POSITIVEINTEGER, + DatatypeEnum.SHORT, + DatatypeEnum.SMALLINT, + DatatypeEnum.UNSIGNEDBYTE, + DatatypeEnum.UNSIGNEDINT, + DatatypeEnum.UNSIGNEDLONG, + DatatypeEnum.UNSIGNEDSHORT, + } +) +_NUMERIC_TYPES = _FLOAT_TYPES | _INT_TYPES + + +def is_numeric_datatype(data_type: DatatypeEnum) -> bool: + return data_type in _NUMERIC_TYPES + + +def _evaluate_numeric_field( + expr: Expression, + values: dict[str, float], + column_units: dict[str, str], + field: str, + source: str, + target_unit: str | None, + data_type: DatatypeEnum, +) -> float | int: + value = evaluate_to_float( + expr, + values, + column_units, + field, + source, + target_unit if target_unit is not None else "", + ) + if data_type in _INT_TYPES: + return int(value) + return value + + +def upload_catalog_columns( + storage: PgStorage, + table_name: str, + catalog: str, + column_map: Mapping[str, str], + expressions: Mapping[str, str], + field_types: Mapping[str, DatatypeEnum], + field_units: Mapping[str, str], + columns: Sequence[str], + batch_size: int, + client: adminapi.AuthenticatedClient, + *, + write: bool = False, + report_func: Callable[[report.Event], None], +) -> int: + if not columns: + raise RuntimeError("No catalog columns selected for upload") + + parsed = {field: parse(source) for field, source in expressions.items()} + expr_cols = set().union(*(expr.referenced_columns for expr in parsed.values())) if parsed else set() + mapped_cols = set(column_map.values()) + all_needed_cols = expr_cols | mapped_cols + + column_names, column_units = fetch_column_units(client, table_name) + validate_columns(table_name, all_needed_cols, column_names) + + units_payload: dict[str, str] = {name: field_units[name] for name in columns if name in field_units} + units = SaveStructuredDataRequestUnits.from_dict(units_payload) if units_payload else UNSET + + uploaded = 0 + skipped = 0 + cnt = storage.query( + sql.SQL("SELECT COUNT(*) AS cnt FROM rawdata.{}").format(sql.Identifier(table_name)), + (), + ) + total_count = int(cnt[0]["cnt"]) if cnt else 0 + processed_rows = 0 + + fetch_columns = sorted(all_needed_cols) + for rows in rawdata_batches(storage, table_name, fetch_columns, batch_size): + batch_ids: list[str] = [] + batch_data: list[list[Any]] = [] + + for row in rows: + if any(row[col] is None for col in all_needed_cols): + skipped += 1 + continue + + values = {col: float(row[col]) for col in expr_cols} + row_values: list[Any] = [] + try: + for name in columns: + if name in parsed: + row_values.append( + _evaluate_numeric_field( + parsed[name], + values, + column_units, + name, + expressions[name], + field_units.get(name), + field_types[name], + ) + ) + else: + row_values.append(str(row[column_map[name]])) + except RuntimeError as e: + raise RuntimeError( + f"failed to evaluate expressions for row {row['hyperleda_internal_id']}: {e}", + ) from e + + batch_ids.append(row["hyperleda_internal_id"]) + batch_data.append(row_values) + uploaded += 1 + + if write and batch_ids: + handle_call( + save_structured_data.sync_detailed( + client=client, + body=action_description.apply( + SaveStructuredDataRequest( + catalog=catalog, + columns=list(columns), + ids=batch_ids, + data=batch_data, + units=units, + ), + ), + ) + ) + + processed_rows += len(rows) + row_pct = int(100 * processed_rows / total_count) if total_count else 0 + report_func(report.ProgressEvent(percent=min(99, row_pct))) + report_func( + report.LogEvent( + message=f"batch: rows_read={len(rows)} uploaded={uploaded} skipped={skipped}", + ), + ) + + total = uploaded + skipped + + def row_pct_label(n: int) -> float: + return (100.0 * n / total) if total else 0.0 + + report_func(report.ProgressEvent(percent=100)) + summary = format_table( + ("Status", "Count", "%"), + [ + ("Uploaded", uploaded, row_pct_label(uploaded)), + ("Skipped (null)", skipped, row_pct_label(skipped)), + ], + title=f"Total rows: {total}\n", + ) + report_func(report.DoneEvent(message=summary)) + return total diff --git a/uploader/app/structured/geometry/upload.py b/uploader/app/structured/geometry/upload.py index 0c81ac3..1edc37f 100644 --- a/uploader/app/structured/geometry/upload.py +++ b/uploader/app/structured/geometry/upload.py @@ -1,6 +1,5 @@ from collections.abc import Callable -import astropy.units as u import matplotlib.pyplot as plt import numpy as np from psycopg import sql @@ -8,12 +7,13 @@ import uploader.app.action_description as action_description import uploader.app.report as report from uploader.app.display import format_table -from uploader.app.lib.expression import Expression, parse +from uploader.app.lib.expression import Expression, evaluate_to_float, parse from uploader.app.lib.rawdata import rawdata_batches +from uploader.app.lib.table import fetch_column_units, validate_columns from uploader.app.storage import PgStorage from uploader.app.upload import handle_call from uploader.clients.gen.client import adminapi -from uploader.clients.gen.client.adminapi.api.default import get_table, save_structured_data +from uploader.clients.gen.client.adminapi.api.default import save_structured_data from uploader.clients.gen.client.adminapi.models.save_structured_data_request import ( SaveStructuredDataRequest, ) @@ -128,70 +128,10 @@ def emit_image( report_func(report.image_event_from_figure(fig, caption=caption)) -def _fetch_column_units( - client: adminapi.AuthenticatedClient, - table_name: str, -) -> tuple[set[str], dict[str, str]]: - resp = handle_call(get_table.sync_detailed(client=client, table_name=table_name)) - column_names: set[str] = set() - column_units: dict[str, str] = {} - for col in resp.data.column_info: - column_names.add(col.name) - if isinstance(col.unit, str): - column_units[col.name] = col.unit - return column_names, column_units - - def _parse_expressions(expressions: dict[str, str]) -> dict[str, Expression]: return {field: parse(source) for field, source in expressions.items()} -def _validate_columns( - table_name: str, - needed_cols: set[str], - column_names: set[str], -) -> None: - missing = sorted(col for col in needed_cols if col not in column_names) - if missing: - raise RuntimeError(f"Table {table_name} has no column(s): {missing}") - - -def _format_unit(unit: u.UnitBase) -> str: - text = f"{unit:s}".strip() - return text if text else "dimensionless" - - -def _eval_context_suffix(expr: Expression, column_units: dict[str, str]) -> str: - if not expr.referenced_columns: - return "" - parts = [f"{col}={column_units.get(col, '')!r}" for col in sorted(expr.referenced_columns)] - return f"; columns: {', '.join(parts)}" - - -def _evaluate_field( - expr: Expression, - values: dict[str, float], - column_units: dict[str, str], - field: str, - source: str, -) -> float: - target = TARGET_UNITS[field] - try: - quantity = expr.evaluate(values, column_units) - except (ValueError, u.UnitConversionError, u.UnitTypeError) as e: - raise RuntimeError( - f"failed to evaluate {field!r} ({source!r}){_eval_context_suffix(expr, column_units)}: {e}", - ) from e - try: - return float(quantity.to(u.Unit(target)).value) - except (u.UnitConversionError, u.UnitTypeError) as e: - raise RuntimeError( - f"failed to convert {field!r} ({source!r}) " - f"from {_format_unit(quantity.unit)} to {target}" - f"{_eval_context_suffix(expr, column_units)}: {e}", - ) from e - - def upload_geometry_isophotal( storage: PgStorage, table_name: str, @@ -209,8 +149,8 @@ def upload_geometry_isophotal( {col: TARGET_UNITS[col] for col in geometry_columns if col in TARGET_UNITS}, ) needed_cols = set().union(*(expr.referenced_columns for expr in parsed.values())) - column_names, column_units = _fetch_column_units(client, table_name) - _validate_columns(table_name, needed_cols, column_names) + column_names, column_units = fetch_column_units(client, table_name) + validate_columns(table_name, needed_cols, column_names) uploaded = 0 skipped = 0 @@ -242,7 +182,14 @@ def upload_geometry_isophotal( values = {col: float(row[col]) for col in needed_cols} try: evaluated = { - field: _evaluate_field(expr, values, column_units, field, expressions[field]) + field: evaluate_to_float( + expr, + values, + column_units, + field, + expressions[field], + TARGET_UNITS[field], + ) for field, expr in parsed.items() } except RuntimeError as e: diff --git a/uploader/app/structured/icrs/upload.py b/uploader/app/structured/icrs/upload.py index ea8fb42..00e788b 100644 --- a/uploader/app/structured/icrs/upload.py +++ b/uploader/app/structured/icrs/upload.py @@ -1,6 +1,5 @@ from collections.abc import Callable -import astropy.units as u import matplotlib.pyplot as plt import numpy as np from psycopg import sql @@ -8,12 +7,13 @@ import uploader.app.action_description as action_description import uploader.app.report as report from uploader.app.display import format_table -from uploader.app.lib.expression import Expression, parse +from uploader.app.lib.expression import Expression, evaluate_to_float, parse from uploader.app.lib.rawdata import rawdata_batches +from uploader.app.lib.table import fetch_column_units, validate_columns from uploader.app.storage import PgStorage from uploader.app.upload import handle_call from uploader.clients.gen.client import adminapi -from uploader.clients.gen.client.adminapi.api.default import get_table, save_structured_data +from uploader.clients.gen.client.adminapi.api.default import save_structured_data from uploader.clients.gen.client.adminapi.models.save_structured_data_request import ( SaveStructuredDataRequest, ) @@ -79,56 +79,6 @@ def _parse_expressions(expressions: dict[str, str]) -> dict[str, Expression]: return {field: parse(source) for field, source in expressions.items()} -def _format_unit(unit: u.UnitBase) -> str: - text = f"{unit:s}".strip() - return text if text else "dimensionless" - - -def _eval_context_suffix(expr: Expression, column_units: dict[str, str]) -> str: - if not expr.referenced_columns: - return "" - parts = [f"{col}={column_units.get(col, '')!r}" for col in sorted(expr.referenced_columns)] - return f"; columns: {', '.join(parts)}" - - -def _evaluate_error_field( - expr: Expression, - values: dict[str, float], - column_units: dict[str, str], - field: str, - source: str, -) -> float: - target = TARGET_ERROR_UNITS[field] - try: - quantity = expr.evaluate(values, column_units) - except (ValueError, u.UnitConversionError, u.UnitTypeError) as e: - raise RuntimeError( - f"failed to evaluate {field!r} ({source!r}){_eval_context_suffix(expr, column_units)}: {e}", - ) from e - try: - return float(quantity.to(u.Unit(target)).value) - except (u.UnitConversionError, u.UnitTypeError) as e: - raise RuntimeError( - f"failed to convert {field!r} ({source!r}) " - f"from {_format_unit(quantity.unit)} to {target}" - f"{_eval_context_suffix(expr, column_units)}: {e}", - ) from e - - -def _fetch_column_units( - client: adminapi.AuthenticatedClient, - table_name: str, -) -> tuple[set[str], dict[str, str]]: - resp = handle_call(get_table.sync_detailed(client=client, table_name=table_name)) - column_names: set[str] = set() - column_units: dict[str, str] = {} - for col in resp.data.column_info: - column_names.add(col.name) - if isinstance(col.unit, str): - column_units[col.name] = col.unit - return column_names, column_units - - def upload_icrs( storage: PgStorage, table_name: str, @@ -142,13 +92,11 @@ def upload_icrs( report_func: Callable[[report.Event], None], ) -> int: parsed = _parse_expressions(expressions) - column_names, column_units = _fetch_column_units(client, table_name) + column_names, column_units = fetch_column_units(client, table_name) error_cols = set().union(*(expr.referenced_columns for expr in parsed.values())) all_needed_cols = {ra_column, dec_column} | error_cols - missing = sorted(col for col in all_needed_cols if col not in column_names) - if missing: - raise RuntimeError(f"Table {table_name} has no column(s): {missing}") + validate_columns(table_name, all_needed_cols, column_names) missing_units = [c for c in (ra_column, dec_column) if c not in column_units] if missing_units: @@ -195,19 +143,21 @@ def upload_icrs( values = {col: float(row[col]) for col in error_cols} try: - e_ra_val = _evaluate_error_field( + e_ra_val = evaluate_to_float( parsed["e_ra"], values, column_units, "e_ra", expressions["e_ra"], + TARGET_ERROR_UNITS["e_ra"], ) - e_dec_val = _evaluate_error_field( + e_dec_val = evaluate_to_float( parsed["e_dec"], values, column_units, "e_dec", expressions["e_dec"], + TARGET_ERROR_UNITS["e_dec"], ) except RuntimeError as e: raise RuntimeError( diff --git a/uploader/cli.py b/uploader/cli.py index 0049e35..3196dd0 100644 --- a/uploader/cli.py +++ b/uploader/cli.py @@ -102,13 +102,18 @@ def cancel_task_run(run_id: str) -> dict[str, str]: if STATIC_DIR.is_dir(): + _NO_CACHE = {"Cache-Control": "no-cache"} + _IMMUTABLE = {"Cache-Control": "public, max-age=31536000, immutable"} @app.get("/{full_path:path}") def serve_frontend(full_path: str) -> FileResponse: file_path = STATIC_DIR / full_path if file_path.is_file() and file_path.resolve().is_relative_to(STATIC_DIR.resolve()): - return FileResponse(path=file_path) - return FileResponse(path=STATIC_DIR / "index.html") + headers = _IMMUTABLE if file_path.relative_to(STATIC_DIR).parts[0] == "assets" else _NO_CACHE + return FileResponse(path=file_path, headers=headers) + if full_path.startswith("assets/") or pathlib.Path(full_path).suffix: + raise HTTPException(status_code=404, detail="Not found") + return FileResponse(path=STATIC_DIR / "index.html", headers=_NO_CACHE) @click.group() diff --git a/uploader/forms/structured_catalog.py b/uploader/forms/structured_catalog.py new file mode 100644 index 0000000..5431785 --- /dev/null +++ b/uploader/forms/structured_catalog.py @@ -0,0 +1,186 @@ +from collections.abc import Callable +from typing import Any, Literal +from urllib.parse import quote_plus + +from psycopg import connect +from pydantic import BaseModel, Field, create_model + +import uploader.app.report as report +from uploader.app import log +from uploader.app.catalogs import fetch_catalogs +from uploader.app.endpoints import db_dsn_map, env_map +from uploader.app.lib.expression import expression_syntax_help +from uploader.app.storage import PgStorage +from uploader.app.structured.generic import is_numeric_datatype, upload_catalog_columns +from uploader.clients.gen.client import adminapi +from uploader.clients.gen.client.adminapi.models.catalog_field import CatalogField +from uploader.clients.gen.client.adminapi.models.catalog_schema import CatalogSchema +from uploader.clients.gen.client.adminapi.models.datatype_enum import DatatypeEnum +from uploader.clients.gen.client.adminapi.types import Unset +from uploader.credentials import load_credentials, load_token +from uploader.tasks import TaskDefinition, register_task + +GROUP = "Structured catalogs" + +_RESERVED_FORM_FIELDS = frozenset({"table_name", "write", "advanced"}) + + +class StructuredCatalogAdvancedSettings(BaseModel): + endpoint: Literal["dev", "test", "prod"] = Field(default="prod", title="API endpoint") + batch_size: int = Field(default=10000, title="Batch size", ge=1, le=500_000) + + +def _field_required(field: CatalogField) -> bool: + if isinstance(field.required, Unset): + return True + return bool(field.required) + + +def _field_description(field: CatalogField, *, numeric: bool) -> str: + if numeric: + if isinstance(field.description, str) and field.description: + return f"Expression. {field.description}" + return f"Expression. Value for {field.name}." + if isinstance(field.description, str) and field.description: + return field.description + return f"Source column for {field.name}." + + +def _field_unit(field: CatalogField) -> str | None: + if isinstance(field.unit, str): + return field.unit + return None + + +def build_catalog_form(schema: CatalogSchema) -> type[BaseModel]: + field_definitions: dict[str, Any] = { + "table_name": (str, Field(..., title="Table name")), + } + for field in schema.fields: + if field.name in _RESERVED_FORM_FIELDS: + raise RuntimeError(f"Catalog {schema.catalog!r} field {field.name!r} conflicts with reserved form field") + numeric = is_numeric_datatype(field.data_type) + title = field.name if numeric else f"{field.name} column" + description = _field_description(field, numeric=numeric) + if _field_required(field): + field_definitions[field.name] = ( + str, + Field(..., title=title, description=description), + ) + else: + field_definitions[field.name] = ( + str, + Field(default="", title=title, description=description), + ) + field_definitions["write"] = ( + bool, + Field( + default=False, + title="Write to API", + description="If enabled, upload results; otherwise dry-run (statistics only).", + ), + ) + field_definitions["advanced"] = ( + StructuredCatalogAdvancedSettings, + Field( + default_factory=StructuredCatalogAdvancedSettings, + title="Advanced settings", + ), + ) + model_name = f"StructuredCatalog{schema.catalog.title().replace('_', '')}Form" + return create_model(model_name, **field_definitions) + + +def _make_handler( + schema: CatalogSchema, +) -> Callable[[BaseModel, Callable[[report.Event], None]], None]: + catalog_fields = list(schema.fields) + catalog_name = schema.catalog + required_names = {f.name for f in catalog_fields if _field_required(f)} + field_types: dict[str, DatatypeEnum] = {f.name: f.data_type for f in catalog_fields} + field_units: dict[str, str] = {f.name: unit for f in catalog_fields if (unit := _field_unit(f)) is not None} + field_order = [f.name for f in catalog_fields] + numeric_names = {f.name for f in catalog_fields if is_numeric_datatype(f.data_type)} + + def handler( + form: BaseModel, + report_func: Callable[[report.Event], None], + ) -> None: + values = form.model_dump() + advanced = values["advanced"] + endpoint = str(advanced["endpoint"]) + batch_size = int(advanced["batch_size"]) + table_name = str(values["table_name"]).strip() + column_map: dict[str, str] = {} + expressions: dict[str, str] = {} + provided: set[str] = set() + for name in field_order: + raw = str(values.get(name, "") or "").strip() + if not raw: + continue + provided.add(name) + if name in numeric_names: + expressions[name] = raw + else: + column_map[name] = raw + missing = sorted(name for name in required_names if name not in provided) + if missing: + raise RuntimeError(f"Missing required field(s): {missing}") + columns = [name for name in field_order if name in provided] + + db_user, db_password = load_credentials() + dsn = db_dsn_map[endpoint].format( + user=quote_plus(db_user), + password=quote_plus(db_password), + ) + client = adminapi.AuthenticatedClient( + base_url=env_map[endpoint], + token=load_token(), + ) + with connect(dsn) as conn: + storage = PgStorage(conn) + upload_catalog_columns( + storage, + table_name, + catalog_name, + column_map, + expressions, + field_types, + field_units, + columns, + batch_size, + client, + write=bool(values["write"]), + report_func=report_func, + ) + + return handler + + +def _task_description(schema: CatalogSchema) -> str: + base = schema.description.strip() + help_text = expression_syntax_help() + if base: + return f"{base}\n\n{help_text}" + return help_text + + +def register_structured_catalog_tasks( + catalogs: list[CatalogSchema] | None = None, +) -> None: + try: + schemas = catalogs if catalogs is not None else fetch_catalogs() + except Exception as e: + log.logger.warning("failed to fetch catalogs for structured catalog tasks", error=str(e)) + return + for schema in schemas: + register_task( + TaskDefinition( + id=f"upload-{schema.catalog}", + title=schema.title, + description=_task_description(schema), + form_model=build_catalog_form(schema), + handler=_make_handler(schema), + group=GROUP, + ), + ) diff --git a/uploader/task_registry.py b/uploader/task_registry.py index c4fdba7..fad11f4 100644 --- a/uploader/task_registry.py +++ b/uploader/task_registry.py @@ -1,6 +1,7 @@ from uploader.app.lib.formula import expression_syntax_help from uploader.forms.authenticate import AuthenticateForm, handle_authenticate from uploader.forms.crossmatch_layered import CrossmatchLayeredForm, handle_crossmatch_layered +from uploader.forms.structured_catalog import register_structured_catalog_tasks from uploader.forms.structured_designation import ( StructuredDesignationForm, handle_structured_designation, @@ -163,3 +164,4 @@ def register_all_tasks() -> None: group="Crossmatch", ), ) + register_structured_catalog_tasks()