feat: rsc metadata local field with url, content guid

pins/boards.py

@@ -158,9 +158,11 @@ def pin_meta(self, name, version: str = None) -> Meta:
         meta_name = self.meta_factory.get_meta_name(*components)
 
         path_meta = self.construct_path([*components, meta_name])
-        f = self._open_pin_meta(path_meta)
+        f, local = self._open_pin_meta(path_meta)
 
-        meta = self.meta_factory.read_pin_yaml(f, pin_name, selected_version)
+        meta = self.meta_factory.read_pin_yaml(
+            f, pin_name, selected_version, local=local
+        )
 
         return meta
 
@@ -605,7 +607,10 @@ def _open_pin_meta(self, path):
         f = self.fs.open(path)
         self._touch_cache(path)
 
-        return f
+        # optional additional data to put in Meta.local
+        local = {}
+
+        return f, local
 
     def _get_cache_path(self, pin_name, version=None, fname=None):
         version_part = [version] if version is not None else []
@@ -691,8 +696,8 @@ def pin_meta(self, name, version=None):
         # note that pins on this board should point to versions, so we use an
         # empty string to mark version (it ultimately is ignored)
         path_meta = self.construct_path([pin_name, "", meta_name])
-        f = self._open_pin_meta(path_meta)
-        meta = self.meta_factory.read_pin_yaml(f, pin_name, VersionRaw(""))
+        f, local = self._open_pin_meta(path_meta)
+        meta = self.meta_factory.read_pin_yaml(f, pin_name, VersionRaw(""), local=local)
 
         # TODO(#59,#83): handle caching, and then re-enable pin_read.
         # self._touch_cache(path_meta)
@@ -866,6 +871,23 @@ def pin_versions_prune(self, *args, **kwargs):
             )
         super().pin_versions_prune(*args, **kwargs)
 
+    def _open_pin_meta(self, path):
+        f = self.fs.open(path)
+        self._touch_cache(path)
+
+        # optional additional data to put in Meta.local
+        user_name, content_name, bundle_id = str(path).split("/")[:3]
+        user_guid = self.fs._user_name_cache[user_name]
+        content_guid = self.fs._content_name_cache[(user_guid, content_name)]
+
+        local = {
+            "content_id": content_guid,
+            "version": bundle_id,
+            "url": f"{self.fs.api.server_url}/content/{content_guid}/",
+        }
+
+        return f, local
+
     def validate_pin_name(self, name) -> None:
         # this should be the default behavior, expecting a full pin name.
         # but because the tests use short names, we allow it to be disabled via config

pins/meta.py

@@ -57,6 +57,9 @@ class Meta:
         TODO - where is this in R pins?
     user:
         A dictionary of additional metadata that may be specified by the user.
+    local:
+        A dictionary of additional metadata that may be added by the board, depending
+        on the backend used. E.g. RStudio Connect content id, url, etc..
 
     """
 
@@ -81,6 +84,7 @@ class Meta:
 
     name: Optional[str] = None
     user: Mapping = field(default_factory=dict)
+    local: Mapping = field(default_factory=dict)
 
     def to_dict(self) -> Mapping:
         data = asdict(self)
@@ -89,18 +93,22 @@ def to_dict(self) -> Mapping:
 
     def to_pin_dict(self):
         d = self.to_dict()
+
         del d["name"]
         del d["version"]
+        del d["local"]
+
         return d
 
     @classmethod
-    def from_pin_dict(cls, data, pin_name, version) -> "Meta":
+    def from_pin_dict(cls, data, pin_name, version, local=None) -> "Meta":
 
         # TODO: re-arrange Meta argument positions to reflect what's been
         # learned about default arguments. e.g. title was not used at some
         # point in api_version 1
         extra = {"title": None} if "title" not in data else {}
-        return cls(**data, **extra, name=pin_name, version=version)
+        local = {} if local is None else local
+        return cls(**data, **extra, name=pin_name, version=version, local=local)
 
     def to_pin_yaml(self, f: Optional[IOBase] = None) -> "str | None":
         data = self.to_pin_dict()
@@ -121,6 +129,7 @@ class MetaV0:
     # holds raw data.txt contents
     original_fields: dict = field(default_factory=dict)
     user: dict = field(default_factory=dict, init=False)
+    local: Mapping = field(default_factory=dict)
 
     title: ClassVar[None] = None
     created: ClassVar[None] = None
@@ -132,15 +141,22 @@ def to_dict(self):
         return asdict(self)
 
     @classmethod
-    def from_pin_dict(cls, data, pin_name, version) -> "MetaV0":
+    def from_pin_dict(cls, data, pin_name, version, local=None) -> "MetaV0":
         # could infer from dataclasses.fields(), but seems excessive.
         req_fields = {"type", "description"}
 
         # Note that we need to .get(), since fields may not be in metadata
         req_inputs = {k: data.get(k) for k in req_fields}
         req_inputs["file"] = data["path"]
 
-        return cls(**req_inputs, name=pin_name, original_fields=data, version=version)
+        local = {} if local is None else local
+        return cls(
+            **req_inputs,
+            name=pin_name,
+            original_fields=data,
+            version=version,
+            local=local,
+        )
 
     def to_pin_dict(self):
         raise NotImplementedError("v0 pins metadata are read only.")
@@ -216,7 +232,11 @@ def create_raw(self, files: Sequence[StrOrFile], type: str, name: str) -> MetaRa
         return MetaRaw(files, type, name)
 
     def read_pin_yaml(
-        self, f: IOBase, pin_name: str, version: "str | VersionRaw"
+        self,
+        f: IOBase,
+        pin_name: str,
+        version: "str | VersionRaw",
+        local=None,
     ) -> Meta:
         if isinstance(version, str):
             version_obj = guess_version(version)
@@ -235,4 +255,4 @@ def read_pin_yaml(
         else:
             cls_meta = Meta
 
-        return cls_meta.from_pin_dict(data, pin_name, version=version_obj)
+        return cls_meta.from_pin_dict(data, pin_name, version=version_obj, local=local)

pins/rsconnect/fs.py

@@ -115,6 +115,9 @@ def __init__(self, server_url, **kwargs):
         else:
             self.api = RsConnectApi(server_url, **kwargs)
 
+        self._user_name_cache = {}
+        self._content_name_cache = {}
+
     def ls(
         self, path, details=False, **kwargs
     ) -> "Sequence[BaseEntity] | Sequence[str]":
@@ -390,7 +393,11 @@ def _get_content_from_name(self, user_guid, content_name):
             raise err(
                 f"Expecting 1 content entry, but found {len(contents)}: {contents}"
             )
-        return contents[0]
+
+        res = contents[0]
+
+        self._content_name_cache[(user_guid, content_name)] = res["guid"]
+        return res
 
     def _get_content_bundle(self, content_guid, bundle_id):
         """Fetch a content bundle."""
@@ -410,7 +417,10 @@ def _get_user_from_name(self, name):
         """Fetch a single user entity from user name."""
         users = self.api.get_users(prefix=name)
         try:
-            user_guid = next(iter([x for x in users if x["username"] == name]))
-            return user_guid
+            user = next(iter([x for x in users if x["username"] == name]))
+
+            self._user_name_cache[user["username"]] = user["guid"]
+
+            return user
         except StopIteration:
             raise ValueError(f"No user named {name} found.")

setup.cfg

@@ -50,7 +50,7 @@ doc =
 
 test =
     pip-tools
-    pytest
+    pytest==7.1.3
     pytest-cases
     pytest-dotenv
     pytest-parallel