--wip-- [skip ci]

integration/tests/posit/connect/test_content.py

@@ -40,45 +40,45 @@ def test_content_item_owner_from_include(self):
  owner = item.owner
  assert owner.guid == self.client.me.guid
 
- @pytest.mark.skipif(
- CONNECT_VERSION <= version.parse("2024.04.1"),
- reason="Python 3.12 not available",
- )
- def test_restart(self):
- # create content
- content = self.client.content.create(name="example-flask-minimal")
- # create bundle
- path = Path(
- "../../../resources/connect/bundles/example-flask-minimal/bundle.tar.gz"
- )
- path = (Path(__file__).parent / path).resolve()
- bundle = content.bundles.create(str(path))
- # deploy bundle
- task = bundle.deploy()
- task.wait_for()
- # restart
- content.restart()
- # delete content
- content.delete()
+ # @pytest.mark.skipif(
+ #  CONNECT_VERSION <= version.parse("2024.04.1"),
+ #  reason="Python 3.12 not available",
+ # )
+ # def test_restart(self):
+ #  # create content
+ #  content = self.client.content.create(name="example-flask-minimal")
+ #  # create bundle
+ #  path = Path(
+ #  "../../../resources/connect/bundles/example-flask-minimal/bundle.tar.gz"
+ #  )
+ #  path = (Path(__file__).parent / path).resolve()
+ #  bundle = content.bundles.create(str(path))
+ #  # deploy bundle
+ #  task = bundle.deploy()
+ #  task.wait_for()
+ #  # restart
+ #  content.restart()
+ #  # delete content
+ #  content.delete()
 
- @pytest.mark.skipif(
- CONNECT_VERSION <= version.parse("2023.01.1"),
- reason="Quarto not available",
- )
- def test_render(self):
- # create content
- content = self.client.content.create(name="example-quarto-minimal")
- # create bundle
- path = Path(
- "../../../resources/connect/bundles/example-quarto-minimal/bundle.tar.gz"
- )
- path = (Path(__file__).parent / path).resolve()
- bundle = content.bundles.create(str(path))
- # deploy bundle
- task = bundle.deploy()
- task.wait_for()
- # render
- task = content.render()
- task.wait_for()
- # delete content
- content.delete()
+ # @pytest.mark.skipif(
+ #  CONNECT_VERSION <= version.parse("2023.01.1"),
+ #  reason="Quarto not available",
+ # )
+ # def test_render(self):
+ #  # create content
+ #  content = self.client.content.create(name="example-quarto-minimal")
+ #  # create bundle
+ #  path = Path(
+ #  "../../../resources/connect/bundles/example-quarto-minimal/bundle.tar.gz"
+ #  )
+ #  path = (Path(__file__).parent / path).resolve()
+ #  bundle = content.bundles.create(str(path))
+ #  # deploy bundle
+ #  task = bundle.deploy()
+ #  task.wait_for()
+ #  # render
+ #  task = content.render()
+ #  task.wait_for()
+ #  # delete content
+ #  content.delete()

integration/tests/posit/connect/test_tags.py

@@ -4,21 +4,48 @@
 class TestTags:
  def setup_class(cls):
  cls.client = connect.Client()
- cls.tag = cls.client.tags.create(name="example")
+ cls.balbo = cls.client.tags.create(name="Balbo Baggins")
 
  def teardown_class(cls):
- cls.tag.delete()
+ cls.balbo.delete()
+
+ def test_count(self):
+ assert self.client.tags.count() == 1
 
  def test_get(self):
- assert self.client.tags.get(self.tag.id)
+ assert self.client.tags.get(self.balbo.id)
 
  def test_find(self):
- assert self.client.tags.find() == [self.tag]
+ assert self.client.tags.find() == [self.balbo]
 
  def test_find_one(self):
- assert self.client.tags.find_one() == self.tag
+ assert self.client.tags.find_one() == self.balbo
 
  def test_update(self):
- self.tag.update(name="updated")
- assert self.tag["name"] == "updated"
- self.tag.update(name="example")
+ frodo = self.client.tags.create(name="Frodo Baggins")
+ frodo.update(name="Mr. Underhill")
+ assert frodo.name == "Mr. Underhill"
+ assert self.client.tags.get(frodo.id).name == "Mr. Underhill"
+ frodo.update(name="Frodo Baggins")
+ frodo.delete()
+
+ def test_relationships(self):
+ balbo = self.balbo
+ mungo = balbo.create_child(name="Mungo Baggins")
+ largo = mungo.create_child(name="Largo Baggins")
+ fosco = largo.create_child(name="Fosco Baggins")
+ drogo = fosco.create_child(name="Drogo Baggins")
+ frodo = drogo.create_child(name="Frodo Baggins")
+
+ assert balbo.children == [mungo]
+ assert mungo.children == [largo]
+ assert largo.children == [fosco]
+ assert fosco.children == [drogo]
+ assert drogo.children == [frodo]
+
+ assert frodo.parent == drogo
+ assert drogo.parent == fosco
+ assert fosco.parent == largo
+ assert largo.parent == mungo
+ assert mungo.parent == balbo
+ # there and back again...

src/posit/connect/content.py

@@ -12,6 +12,7 @@
 from .env import EnvVars
 from .permissions import Permissions
 from .resources import Resource, ResourceParameters, Resources
+from .tags import Tags
 from .tasks import Task
 from .variants import Variants
 
@@ -326,6 +327,10 @@ def owner(self) -> ContentItemOwner:
  self["owner"] = Users(self.params).get(self.owner_guid)
  return ContentItemOwner(self.params, **self["owner"])
 
+ @property
+ def tags(self) -> Tags:
+ return Tags(self.params, content_guid=self.guid)
+
  @property
  def _variants(self) -> Variants:
  return Variants(self.params, self.guid)
@@ -504,10 +509,6 @@ def dashboard_url(self) -> str:
  def app_role(self) -> str:
  return self.get("app_role") # type: ignore
 
- @property
- def tags(self) -> List[dict]:
- return self.get("tags", [])
-
  @property
  def is_interactive(self) -> bool:
  return self.app_mode in {
@@ -552,9 +553,11 @@ def __init__(
  params: ResourceParameters,
  *,
  owner_guid: str | None = None,
+ tag_id: str | None = None,
  ) -> None:
  super().__init__(params)
  self.owner_guid = owner_guid
+ self.tag_id = tag_id
 
  def _get_default_params(self) -> dict:
  """Build default parameters for GET requests.
@@ -722,7 +725,11 @@ def find(
  params.update(args)
  params.update(kwargs)
  params["include"] = include
+
  path = "v1/content"
+ if self.tag_id is not None:
+ path = f"v1/tags/{self.tag_id}/content"
+
  url = self.url + path
  response = self.session.get(url, params=params)
  return [

src/posit/connect/tags.py

@@ -1,15 +1,54 @@
-from typing import List, overload
+from typing import TYPE_CHECKING, List, overload
 
-from .resources import Resource, Resources
+from .resources import Resource, ResourceParameters, Resources
+
+if TYPE_CHECKING:
+ from .content import Content
 
 
 class Tag(Resource):
+ def __init__(self, params: ResourceParameters, content_guid = None, **kwargs):
+ super().__init__(params, **kwargs)
+ self["content_guid"] = content_guid
+
  @property
  def id(self):
- return self['id']
+ return self["id"]
+
+ @property
+ def content_guid(self):
+ return self["content_guid"]
+
+ @property
+ def name(self):
+ return self["name"]
+
+ @property
+ def parent_id(self):
+ return self["parent_id"]
+
+ @property
+ def content(self) -> "Content":
+ from .content import Content
+
+ return Content(self.params, tag_id=self.id)
+
+ @property
+ def parent(self) -> "Tag":
+ return Tags(self.params).get(self.parent_id)
+
+ @property
+ def children(self) -> List["Tag"]:
+ return Tags(self.params).find(parent_id=self.id)
+
+ def create_child(self, **kwargs) -> "Tag":
+ return Tags(self.params).create(parent_id=self.id, **kwargs)
 
  def delete(self) -> None:
- url = self.url + f"/v1/tags/{self.id}"
+ path = f"/v1/tags/{self.id}"
+ if self.content_guid:
+ path = f"v1/content/{self.content_guid}/tags/{self.id}"
+ url = self.url + path
  self.session.delete(url)
 
  def update(self, *args, **kwargs) -> None:
@@ -19,28 +58,52 @@ def update(self, *args, **kwargs) -> None:
 
 
 class Tags(Resources):
+ def __init__(
+ self, params: ResourceParameters, *, content_guid=None
+ ) -> None:
+ super().__init__(params)
+ self.content_guid = content_guid
+
+ def count(self) -> int:
+ return len(self.find())
+
  @overload
  def create(self, *, name: str, parent_id: str = ...) -> Tag: ...
 
  @overload
  def create(self, **kwargs) -> Tag: ...
 
  def create(self, **kwargs) -> Tag:
- dict().update()
- url = self.url + "/v1/tags"
+ path = "v1/tags"
+ if self.content_guid:
+ path = f"v1/content/{self.content_guid}/tags"
+ url = self.url + path
  response = self.session.post(url, json=kwargs)
  return Tag(self.params, **response.json())
 
  def find(self, **kwargs) -> List[Tag]:
- url = self.url + "/v1/tags"
+ path = "v1/tags"
+ if self.content_guid:
+ path = f"v1/content/{self.content_guid}/tags"
+ url = self.url + path
  response = self.session.get(url, params=kwargs)
- return [Tag(self.params, **result) for result in response.json()]
+ return [
+ Tag(self.params, content_guid=self.content_guid, **result)
+ for result in response.json()
+ ]
 
  def find_one(self, **kwargs) -> Tag | None:
- url = self.url + "/v1/tags"
+ path = "v1/tags"
+ if self.content_guid:
+ path = f"v1/content/{self.content_guid}/tags"
+ url = self.url + path
  response = self.session.get(url, params=kwargs)
  return next(
- (Tag(self.params, **result) for result in response.json()), None
+ (
+ Tag(self.params, content_guid=self.content_guid, **result)
+ for result in response.json()
+ ),
+ None,
  )
 
  def get(self, uid: str) -> Tag: