Merge pull request #63 from mrc-ide/mrc-5015

Add support for pushing to http locations.

Author: plietar

example/src/data/data.py

@@ -0,0 +1,5 @@
+import random
+
+d = [random.random() for _ in range(10)]
+with open("result.txt", "w") as f:
+    f.writelines(f"{x}\n" for x in d)

example/src/data/orderly.R

@@ -0,0 +1,12 @@
+import pyorderly
+
+pyorderly.dependency(
+    None, "latest(name == 'data')", {"input.txt": "result.txt"}
+)
+pyorderly.artefact("Summary", "summary.txt")
+
+with open("input.txt") as f:
+    d = [float(x.strip()) for x in f.readlines()]
+total = sum(d)
+with open("summary.txt", "w") as f:
+    f.write(f"{total}\n")

example/src/depends/depends.py

@@ -193,6 +193,7 @@ exclude_lines = [
   "@abstractmethod",
   "@abc.abstractmethod",
   "@overload",
+  "raise NotImplementedError",
 ]
 
 [tool.pytest.ini_options]

example/src/depends/orderly.R

@@ -1,6 +1,7 @@
 from pathlib import Path
 
 from pyorderly.outpack.config import Config, read_config, write_config
+from pyorderly.outpack.root import OutpackRoot
 
 
 def outpack_init(
@@ -31,16 +32,15 @@ def outpack_init(
         _validate_same_core_configuration(config.core, read_config(path).core)
     else:
         path_outpack.mkdir(parents=True, exist_ok=True)
-        path_outpack.joinpath("metadata").mkdir(parents=True, exist_ok=True)
-        path_outpack.joinpath("location").mkdir(parents=True, exist_ok=True)
-        path_outpack.joinpath("location/local").mkdir(
-            parents=True, exist_ok=True
-        )
+        path_outpack.joinpath("metadata").mkdir(exist_ok=True)
+        path_outpack.joinpath("location").mkdir(exist_ok=True)
+        path_outpack.joinpath("location/local").mkdir(exist_ok=True)
         if path_archive is not None:
             path.joinpath(path_archive).mkdir(exist_ok=True)
+
         write_config(config, path)
 
-    return path
+    return OutpackRoot(path).path
 
 
 def _validate_same_core_configuration(now, then):

pyproject.toml

@@ -1,20 +1,24 @@
 import collections
 import shutil
 from pathlib import PurePath
+from typing import Union
 
 from pyorderly.outpack.config import Location, update_config
 from pyorderly.outpack.location_driver import LocationDriver
 from pyorderly.outpack.location_http import OutpackLocationHTTP
 from pyorderly.outpack.location_packit import outpack_location_packit
 from pyorderly.outpack.location_path import OutpackLocationPath
 from pyorderly.outpack.location_ssh import OutpackLocationSSH, parse_ssh_url
+from pyorderly.outpack.metadata import MetadataCore
 from pyorderly.outpack.root import OutpackRoot, root_open
 from pyorderly.outpack.static import (
     LOCATION_LOCAL,
     LOCATION_ORPHAN,
     LOCATION_RESERVED_NAME,
 )
 
+LocationSelector = Union[None, str, list[str]]
+
 
 def outpack_location_list(root=None, *, locate=True):
     root = root_open(root, locate=locate)
@@ -94,15 +98,20 @@ def outpack_location_rename(old, new, root=None, *, locate=True):
 
 
 def location_resolve_valid(
-    location, root, *, include_local, include_orphan, allow_no_locations
-):
+    location: LocationSelector,
+    root: OutpackRoot,
+    *,
+    include_local: bool,
+    include_orphan: bool,
+    allow_no_locations: bool,
+) -> list[str]:
     if location is None:
-        location = outpack_location_list(root)
+        result = outpack_location_list(root)
     elif isinstance(location, str):
         if location not in outpack_location_list(root):
             msg = f"Unknown location: '{location}'"
             raise Exception(msg)
-        location = [location]
+        result = [location]
     elif isinstance(location, collections.abc.Iterable) and all(
         isinstance(item, str) for item in location
     ):
@@ -111,24 +120,24 @@ def location_resolve_valid(
             unknown_text = "', '".join(unknown)
             msg = f"Unknown location: '{unknown_text}'"
             raise Exception(msg)
-        location = list(location)
+        result = list(location)
     else:
         msg = (
             "Invalid input for 'location'; expected None or a list of "
             "strings"
         )
         raise Exception(msg)
 
-    if not include_local and LOCATION_LOCAL in location:
-        location.remove(LOCATION_LOCAL)
-    if not include_orphan and LOCATION_ORPHAN in location:  # pragma: no cover
-        location.remove(LOCATION_ORPHAN)
+    if not include_local and LOCATION_LOCAL in result:
+        result.remove(LOCATION_LOCAL)
+    if not include_orphan and LOCATION_ORPHAN in result:  # pragma: no cover
+        result.remove(LOCATION_ORPHAN)
 
-    if len(location) == 0 and not allow_no_locations:
+    if len(result) == 0 and not allow_no_locations:
         msg = "No suitable location found"
         raise Exception(msg)
 
-    return location
+    return result
 
 
 def _location_check_new_name(root, name):
@@ -169,3 +178,39 @@ def _location_driver(location_name, root) -> LocationDriver:
 
     msg = "invalid location type"
     raise Exception(msg)
+
+
+def _find_all_dependencies(
+    packet_ids: list[str],
+    metadata: dict[str, MetadataCore],
+    *,
+    allow_missing_packets: bool = False,
+) -> list[str]:
+    result = []
+
+    # This is a standard depth first search through the packet graph.
+    seen = set(packet_ids)
+    todo = list(packet_ids)
+    while todo:
+        packet_id = todo.pop()
+        result.append(packet_id)
+
+        m = metadata.get(packet_id)
+        if m is not None:
+            for dep in m.depends:
+                if dep.packet not in seen:
+                    seen.add(dep.packet)
+                    todo.append(dep.packet)
+        elif not allow_missing_packets:
+            msg = f"Unknown packet {packet_id}"
+            raise Exception(msg)
+
+    # We want the result to be reverse-topologically sorted, such that
+    # dependencies come before their dependents. In principle, we could get
+    # such an order directly from the graph traversal, but doing this with
+    # multiple start points is not as easy as it seems.
+    #
+    # Using a lexicographic sort on the packet IDs is a reasonable alternative
+    # because the IDs start with their timestamp, and dependencies have smaller
+    # timestamps than dependents.
+    return sorted(result)

src/pyorderly/outpack/init.py

@@ -1,6 +1,6 @@
-import builtins
 from abc import abstractmethod
 from contextlib import AbstractContextManager
+from pathlib import Path
 
 from pyorderly.outpack.metadata import MetadataCore, PacketFile, PacketLocation
 
@@ -14,12 +14,24 @@ class LocationDriver(AbstractContextManager):
     """
 
     @abstractmethod
-    def list(self) -> dict[str, PacketLocation]: ...
+    def list_packets(self) -> dict[str, PacketLocation]: ...
 
     @abstractmethod
-    def metadata(self, packet_ids: builtins.list[str]) -> dict[str, str]: ...
+    def metadata(self, packet_ids: list[str]) -> dict[str, str]: ...
 
     @abstractmethod
     def fetch_file(
         self, packet: MetadataCore, file: PacketFile, dest: str
     ) -> None: ...
+
+    @abstractmethod
+    def list_unknown_packets(self, ids: list[str]) -> list[str]: ...
+
+    @abstractmethod
+    def list_unknown_files(self, hashes: list[str]) -> list[str]: ...
+
+    @abstractmethod
+    def push_file(self, src: Path, hash: str): ...
+
+    @abstractmethod
+    def push_metadata(self, src: Path, hash: str): ...

src/pyorderly/outpack/location.py

@@ -1,5 +1,5 @@
-import builtins
 import shutil
+from pathlib import Path
 from urllib.parse import urljoin
 
 import requests
@@ -58,15 +58,15 @@ def __exit__(self, *args):
         self._client.__exit__(*args)
 
     @override
-    def list(self) -> dict[str, PacketLocation]:
+    def list_packets(self) -> dict[str, PacketLocation]:
         response = self._client.get("metadata/list").json()
         data = response["data"]
         return {
             entry["packet"]: PacketLocation.from_dict(entry) for entry in data
         }
 
     @override
-    def metadata(self, ids: builtins.list[str]) -> dict[str, str]:
+    def metadata(self, ids: list[str]) -> dict[str, str]:
         result = {}
         for i in ids:
             result[i] = self._client.get(f"metadata/{i}/text").text
@@ -78,3 +78,34 @@ def fetch_file(self, packet: MetadataCore, file: PacketFile, dest: str):
         response = self._client.get(f"file/{file.hash}", stream=True)
         with open(dest, "wb") as f:
             shutil.copyfileobj(response.raw, f)
+
+    @override
+    def list_unknown_packets(self, ids: list[str]) -> list[str]:
+        response = self._client.post(
+            "packets/missing",
+            json={
+                "ids": ids,
+                "unpacked": True,
+            },
+        ).json()
+        return response["data"]
+
+    @override
+    def list_unknown_files(self, hashes: list[str]) -> list[str]:
+        response = self._client.post(
+            "files/missing",
+            json={
+                "hashes": hashes,
+            },
+        ).json()
+        return response["data"]
+
+    @override
+    def push_file(self, src: Path, hash: str):
+        with open(src, "rb") as f:
+            self._client.post(f"file/{hash}", stream=True, data=f)
+
+    @override
+    def push_metadata(self, src: Path, hash: str):
+        with open(src, "rb") as f:
+            self._client.post(f"packet/{hash}", stream=True, data=f)

src/pyorderly/outpack/location_driver.py

@@ -1,6 +1,6 @@
-import builtins
 import os
 import shutil
+from pathlib import Path
 
 from typing_extensions import override
 
@@ -24,11 +24,11 @@ def __exit__(self, exc_type, exc_value, exc_tb):
         pass
 
     @override
-    def list(self) -> dict[str, PacketLocation]:
+    def list_packets(self) -> dict[str, PacketLocation]:
         return self.__root.index.location(LOCATION_LOCAL)
 
     @override
-    def metadata(self, packet_ids: builtins.list[str]) -> dict[str, str]:
+    def metadata(self, packet_ids: list[str]) -> dict[str, str]:
         all_ids = self.__root.index.location(LOCATION_LOCAL).keys()
         missing_ids = set(packet_ids).difference(all_ids)
         if missing_ids:
@@ -54,3 +54,19 @@ def fetch_file(self, _packet: MetadataCore, file: PacketFile, dest: str):
                 msg = f"Hash '{file.hash}' not found at location"
                 raise Exception(msg)
         shutil.copyfile(path, dest)
+
+    @override
+    def list_unknown_packets(self, ids: list[str]) -> list[str]:
+        raise NotImplementedError()
+
+    @override
+    def list_unknown_files(self, hashes: list[str]) -> list[str]:
+        raise NotImplementedError()
+
+    @override
+    def push_file(self, src: Path, hash: str):
+        raise NotImplementedError()
+
+    @override
+    def push_metadata(self, src: Path, hash: str):
+        raise NotImplementedError()