diff --git a/examples/convert_database.py b/examples/convert_database.py new file mode 100644 index 00000000..6bc0b4aa --- /dev/null +++ b/examples/convert_database.py @@ -0,0 +1,140 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = ["pyvolca==0.8.0"] +# /// +"""Download VoLCA, import one database, and export one or more formats.""" + +import argparse +import json +import os +import re +import secrets +import socket +import tempfile +import urllib.parse +import urllib.request +import warnings +from pathlib import Path + +from volca import Client, Server, VoLCAError, download + +EXT = { + "simapro": ".simapro.csv", + "ilcd": ".ilcd.zip", + "ecospold1": ".ecospold1.xml", + "ecospold2": ".ecospold2.zip", + "brightway": ".brightway.xlsx", +} + + +def safe_slug(slug): + if not isinstance(slug, str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", slug): + raise ValueError("server returned an unsafe database slug") + return slug + + +def free_port(): + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def export_bytes(client, server, fmt): + """Use pyvolca, with a narrow 0.8.0 fallback for its missing Accept header.""" + try: + return client.export_database(fmt) + except VoLCAError as error: + if "HTTP 406" not in str(error): + raise + request = urllib.request.Request( + f"{server.base_url}/api/v1/db/{client.db}/export", + data=json.dumps({"format": fmt}).encode(), method="POST", + headers={ + "Authorization": f"Bearer {server.password}", + "Content-Type": "application/json", + "Accept": "application/octet-stream", + }, + ) + with urllib.request.urlopen(request, timeout=1800) as response: + for line in urllib.parse.unquote( + response.headers.get("X-Volca-Export-Warnings", "") + ).splitlines(): + warnings.warn(line, stacklevel=2) + return response.read() + + +def publish(data, output, force): + with tempfile.NamedTemporaryFile(dir=output.parent, delete=False) as stream: + temporary = Path(stream.name) + stream.write(data) + try: + if force: + os.replace(temporary, output) + temporary = None + else: + os.link(temporary, output) + finally: + if temporary: + temporary.unlink(missing_ok=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source", required=True, type=Path) + parser.add_argument("-f", "--format", nargs="+", required=True, choices=EXT) + parser.add_argument("-o", "--out", type=Path, default=Path.cwd()) + parser.add_argument("--engine-version", default="v0.9.1") + parser.add_argument("--force", action="store_true") + args = parser.parse_args() + + source, out = args.source.expanduser().resolve(), args.out.expanduser().resolve() + if not source.is_file(): + raise FileNotFoundError(source) + out.mkdir(parents=True, exist_ok=True) + installed = download(version=args.engine_version) + + with tempfile.TemporaryDirectory(prefix="volca-convert-") as raw: + root, port, password = Path(raw), free_port(), secrets.token_urlsafe(32) + config = root / "volca.toml" + config.write_text(f'[server]\nhost="127.0.0.1"\nport={port}\npassword="{password}"\n') + previous_data_dir = os.environ.get("VOLCA_DATA_DIR") + os.environ["VOLCA_DATA_DIR"] = raw + try: + with Server(config=str(config), binary=str(installed.binary)) as server: + client = Client(server.base_url, password=server.password) + uploaded = client.upload_database(source, source.stem) + client.db = safe_slug(uploaded["slug"]) + loaded = False + try: + setup = client.get_setup() + if not setup["isReady"]: + raise RuntimeError( + "database needs dependencies: " + + ", ".join(setup["missingSuppliers"]) + ) + client.finalize_database() + loaded = True + outputs = { + fmt: out / (client.db + EXT[fmt]) + for fmt in dict.fromkeys(args.format) + } + if not args.force: + existing = [path for path in outputs.values() if os.path.lexists(path)] + if existing: + raise FileExistsError(existing[0]) + for fmt, output in outputs.items(): + publish(export_bytes(client, server, fmt), output, args.force) + print(f"{fmt}: {output}") + finally: + if loaded: + client.unload_database(client.db) + client.delete_database() + finally: + if previous_data_dir is None: + os.environ.pop("VOLCA_DATA_DIR", None) + else: + os.environ["VOLCA_DATA_DIR"] = previous_data_dir + + +if __name__ == "__main__": + main() diff --git a/pyvolca/src/volca/client.py b/pyvolca/src/volca/client.py index d37303c8..ca6b5168 100644 --- a/pyvolca/src/volca/client.py +++ b/pyvolca/src/volca/client.py @@ -860,6 +860,7 @@ def export_database(self, fmt: str, db_name: str | None = None) -> bytes: resp = self._session.post( f"{self.base_url}/api/v1/db/{target}/export", json={"format": fmt_norm}, + headers={"Accept": "application/octet-stream"}, ) if resp.status_code >= 400: raise VoLCAError( diff --git a/pyvolca/tests/test_db_write.py b/pyvolca/tests/test_db_write.py index f184ccf4..056e5519 100644 --- a/pyvolca/tests/test_db_write.py +++ b/pyvolca/tests/test_db_write.py @@ -211,6 +211,9 @@ def test_returns_raw_bytes(self, mocked_client): url = session.post.call_args[0][0] assert url == "http://test.local/api/v1/db/testdb/export" assert session.post.call_args[1]["json"] == {"format": "ecospold2"} + assert session.post.call_args[1]["headers"] == { + "Accept": "application/octet-stream" + } def test_format_normalized_before_send(self, mocked_client): client, session = mocked_client diff --git a/pyvolca/tests/test_examples.py b/pyvolca/tests/test_examples.py new file mode 100644 index 00000000..bd9f7c15 --- /dev/null +++ b/pyvolca/tests/test_examples.py @@ -0,0 +1,97 @@ +import ast +import importlib.util +import threading +import urllib.parse +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest + + +EXAMPLE = Path(__file__).parents[2] / "examples" / "convert_database.py" + + +def load_example(): + spec = importlib.util.spec_from_file_location("convert_database", EXAMPLE) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_converter_rejects_unsafe_server_slugs(): + example = load_example() + assert example.safe_slug("database-1.0") == "database-1.0" + for slug in ("../outside", "/absolute", "nested/path", "nested\\path", ""): + with pytest.raises(ValueError): + example.safe_slug(slug) + + +def test_pyvolca_080_export_fallback_surfaces_warnings(): + example = load_example() + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + self.send_response(200) + self.send_header( + "X-Volca-Export-Warnings", urllib.parse.quote("first warning") + ) + self.end_headers() + self.wfile.write(b"export") + + class Client: + db = "database" + + def export_database(self, _fmt): + raise example.VoLCAError("export_database failed (HTTP 406)") + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + managed = type( + "Server", (), + {"base_url": f"http://127.0.0.1:{server.server_port}", "password": "token"}, + )() + try: + with pytest.warns(UserWarning, match="first warning"): + assert example.export_bytes(Client(), managed, "ilcd") == b"export" + finally: + server.shutdown() + server.server_close() + + +def test_converter_uses_only_public_pyvolca_lifecycle(): + source = EXAMPLE.read_text() + tree = ast.parse(source) + imported = { + name.name + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module == "volca" + for name in node.names + } + calls = { + node.func.attr + for node in ast.walk(tree) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + } + + assert {"Client", "Server", "download"} <= imported + assert { + "upload_database", + "get_setup", + "finalize_database", + "export_database", + "unload_database", + "delete_database", + } <= calls + assert "._session" not in source + assert "import requests" not in source + unload_calls = [ + node for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "unload_database" + ] + assert unload_calls and all(call.args for call in unload_calls)