diff --git a/.vscode/settings.json b/.vscode/settings.json index 4169db45..6fbbf1ba 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -4,7 +4,4 @@ ], "python.testing.unittestEnabled": false, "python.testing.pytestEnabled": true, - "cSpell.words": [ - "mypy" - ] } diff --git a/docs/_quarto.yml b/docs/_quarto.yml index a0555602..4cacdfce 100644 --- a/docs/_quarto.yml +++ b/docs/_quarto.yml @@ -104,6 +104,7 @@ quartodoc: - connect.permissions - connect.tasks - connect.users + - connect.vanities - title: Posit Connect Metrics package: posit contents: diff --git a/integration/tests/posit/connect/test_vanities.py b/integration/tests/posit/connect/test_vanities.py new file mode 100644 index 00000000..b79d43cc --- /dev/null +++ b/integration/tests/posit/connect/test_vanities.py @@ -0,0 +1,73 @@ +from posit import connect + + +class TestVanities: + @classmethod + def setup_class(cls): + cls.client = connect.Client() + + @classmethod + def teardown_class(cls): + assert cls.client.content.count() == 0 + + def test_all(self): + content = self.client.content.create(name="example") + + # None by default + vanities = self.client.vanities.all() + assert len(vanities) == 0 + + # Set + content.vanity = "example" + + # Get + vanities = self.client.vanities.all() + assert len(vanities) == 1 + + # Cleanup + content.delete() + + vanities = self.client.vanities.all() + assert len(vanities) == 0 + + def test_property(self): + content = self.client.content.create(name="example") + + # None by default + assert content.vanity is None + + # Set + content.vanity = "example" + + # Get + vanity = content.vanity + assert vanity == "/example/" + + # Delete + del content.vanity + assert content.vanity is None + + # Cleanup + content.delete() + + def test_destroy(self): + content = self.client.content.create(name="example") + + # None by default + assert content.vanity is None + + # Set + content.vanity = "example" + + # Get + vanity = content.find_vanity() + assert vanity + assert vanity["path"] == "/example/" + + # Delete + vanity.destroy() + content.reset_vanity() + assert content.vanity is None + + # Cleanup + content.delete() diff --git a/src/posit/connect/client.py b/src/posit/connect/client.py index 0f42f475..5c46dd85 100644 --- a/src/posit/connect/client.py +++ b/src/posit/connect/client.py @@ -17,6 +17,7 @@ from .resources import ResourceParameters from .tasks import Tasks from .users import User, Users +from .vanities import Vanities class Client(ContextManager): @@ -271,6 +272,10 @@ def oauth(self) -> OAuth: """ return OAuth(self.resource_params, self.cfg.api_key) + @property + def vanities(self) -> Vanities: + return Vanities(self.resource_params) + def __del__(self): """Close the session when the Client instance is deleted.""" if hasattr(self, "session") and self.session is not None: diff --git a/src/posit/connect/content.py b/src/posit/connect/content.py index d90377d1..fdf2cc95 100644 --- a/src/posit/connect/content.py +++ b/src/posit/connect/content.py @@ -7,14 +7,14 @@ from posixpath import dirname from typing import Any, List, Literal, Optional, overload -from posit.connect.oauth.associations import ContentItemAssociations - from . import tasks from .bundles import Bundles from .env import EnvVars +from .oauth.associations import ContentItemAssociations from .permissions import Permissions from .resources import Resource, ResourceParameters, Resources from .tasks import Task +from .vanities import VanityMixin from .variants import Variants @@ -32,7 +32,7 @@ class ContentItemOwner(Resource): pass -class ContentItem(Resource): +class ContentItem(VanityMixin, Resource): def __getitem__(self, key: Any) -> Any: v = super().__getitem__(key) if key == "owner" and isinstance(v, dict): diff --git a/src/posit/connect/context.py b/src/posit/connect/context.py index 14b7e330..c93fe7b0 100644 --- a/src/posit/connect/context.py +++ b/src/posit/connect/context.py @@ -1,6 +1,7 @@ import functools from typing import Optional, Protocol +import requests from packaging.version import Version @@ -21,7 +22,7 @@ def wrapper(instance: ContextManager, *args, **kwargs): class Context(dict): - def __init__(self, session, url): + def __init__(self, session: requests.Session, url: str): self.session = session self.url = url diff --git a/src/posit/connect/vanities.py b/src/posit/connect/vanities.py new file mode 100644 index 00000000..a13d0282 --- /dev/null +++ b/src/posit/connect/vanities.py @@ -0,0 +1,236 @@ +from typing import Callable, List, Optional, TypedDict + +from typing_extensions import NotRequired, Required, Unpack + +from .errors import ClientError +from .resources import Resource, ResourceParameters, Resources + + +class Vanity(Resource): + """A vanity resource. + + Vanities maintain custom URL paths assigned to content. + + Warnings + -------- + Vanity paths may only contain alphanumeric characters, hyphens, underscores, and slashes. + + Vanities cannot have children. For example, if the vanity path "/finance/" exists, the vanity path "/finance/budget/" cannot. But, if "/finance" does not exist, both "/finance/budget/" and "/finance/report" are allowed. + + The following vanities are reserved by Connect: + - `/__` + - `/favicon.ico` + - `/connect` + - `/apps` + - `/users` + - `/groups` + - `/setpassword` + - `/user-completion` + - `/confirm` + - `/recent` + - `/reports` + - `/plots` + - `/unpublished` + - `/settings` + - `/metrics` + - `/tokens` + - `/help` + - `/login` + - `/welcome` + - `/register` + - `/resetpassword` + - `/content` + """ + + AfterDestroyCallback = Callable[[], None] + + class VanityAttributes(TypedDict): + """Vanity attributes.""" + + path: Required[str] + content_guid: Required[str] + created_time: Required[str] + + def __init__( + self, + /, + params: ResourceParameters, + *, + after_destroy: Optional[AfterDestroyCallback] = None, + **kwargs: Unpack[VanityAttributes], + ): + """Initialize a Vanity. + + Parameters + ---------- + params : ResourceParameters + after_destroy : AfterDestroyCallback, optional + Called after the Vanity is successfully destroyed, by default None + """ + super().__init__(params, **kwargs) + self._after_destroy = after_destroy + self._content_guid = kwargs["content_guid"] + + @property + def _endpoint(self): + return self.params.url + f"v1/content/{self._content_guid}/vanity" + + def destroy(self) -> None: + """Destroy the vanity. + + Raises + ------ + ValueError + If the foreign unique identifier is missing or its value is `None`. + + Warnings + -------- + This operation is irreversible. + + Note + ---- + This action requires administrator privileges. + """ + self.params.session.delete(self._endpoint) + + if self._after_destroy: + self._after_destroy() + + +class Vanities(Resources): + """Manages a collection of vanities.""" + + def all(self) -> List[Vanity]: + """Retrieve all vanities. + + Returns + ------- + List[Vanity] + + Notes + ----- + This action requires administrator privileges. + """ + endpoint = self.params.url + "v1/vanities" + response = self.params.session.get(endpoint) + results = response.json() + return [Vanity(self.params, **result) for result in results] + + +class VanityMixin(Resource): + """Mixin class to add a vanity attribute to a resource.""" + + class HasGuid(TypedDict): + """Has a guid.""" + + guid: Required[str] + + def __init__(self, params: ResourceParameters, **kwargs: Unpack[HasGuid]): + super().__init__(params, **kwargs) + self._content_guid = kwargs["guid"] + self._vanity: Optional[Vanity] = None + + @property + def _endpoint(self): + return self.params.url + f"v1/content/{self._content_guid}/vanity" + + @property + def vanity(self) -> Optional[str]: + """Get the vanity.""" + if self._vanity: + return self._vanity["path"] + + try: + self._vanity = self.find_vanity() + self._vanity._after_destroy = self.reset_vanity + return self._vanity["path"] + except ClientError as e: + if e.http_status == 404: + return None + raise e + + @vanity.setter + def vanity(self, value: str) -> None: + """Set the vanity. + + Parameters + ---------- + value : str + The vanity path. + + Note + ---- + This action requires owner or administrator privileges. + + See Also + -------- + create_vanity + """ + self._vanity = self.create_vanity(path=value) + self._vanity._after_destroy = self.reset_vanity + + @vanity.deleter + def vanity(self) -> None: + """Destroy the vanity. + + Warnings + -------- + This operation is irreversible. + + Note + ---- + This action requires owner or administrator privileges. + + See Also + -------- + reset_vanity + """ + self.vanity + if self._vanity: + self._vanity.destroy() + self.reset_vanity() + + def reset_vanity(self) -> None: + """Unload the cached vanity. + + Forces the next access, if any, to query the vanity from the Connect server. + """ + self._vanity = None + + class CreateVanityRequest(TypedDict, total=False): + """A request schema for creating a vanity.""" + + path: Required[str] + """The vanity path (.e.g, 'my-dashboard')""" + + force: NotRequired[bool] + """Whether to force creation of the vanity""" + + def create_vanity(self, **kwargs: Unpack[CreateVanityRequest]) -> Vanity: + """Create a vanity. + + Parameters + ---------- + path : str, required + The path for the vanity. + force : bool, not required + Whether to force the creation of the vanity. When True, any other vanity with the same path will be deleted. + + Warnings + -------- + If setting force=True, the destroy operation performed on the other vanity is irreversible. + """ + response = self.params.session.put(self._endpoint, json=kwargs) + result = response.json() + return Vanity(self.params, **result) + + def find_vanity(self) -> Vanity: + """Find the vanity. + + Returns + ------- + Vanity + """ + response = self.params.session.get(self._endpoint) + result = response.json() + return Vanity(self.params, **result) diff --git a/tests/posit/connect/test_vanities.py b/tests/posit/connect/test_vanities.py new file mode 100644 index 00000000..e77bfc54 --- /dev/null +++ b/tests/posit/connect/test_vanities.py @@ -0,0 +1,116 @@ +from unittest.mock import Mock + +import requests +import responses +from responses.matchers import json_params_matcher + +from posit.connect.resources import ResourceParameters +from posit.connect.urls import Url +from posit.connect.vanities import Vanities, Vanity, VanityMixin + + +class TestVanityDestroy: + @responses.activate + def test_destroy_sends_delete_request(self): + content_guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" + base_url = "http://connect.example/__api__" + endpoint = f"{base_url}/v1/content/{content_guid}/vanity" + mock_delete = responses.delete(endpoint) + + session = requests.Session() + url = Url(base_url) + params = ResourceParameters(session, url) + vanity = Vanity(params, content_guid=content_guid, path=Mock(), created_time=Mock()) + + vanity.destroy() + + assert mock_delete.call_count == 1 + + @responses.activate + def test_destroy_calls_after_destroy_callback(self): + content_guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" + base_url = "http://connect.example/__api__" + endpoint = f"{base_url}/v1/content/{content_guid}/vanity" + responses.delete(endpoint) + + session = requests.Session() + url = Url(base_url) + after_destroy = Mock() + params = ResourceParameters(session, url) + vanity = Vanity(params, after_destroy=after_destroy, content_guid=content_guid) + + vanity.destroy() + + assert after_destroy.call_count == 1 + + +class TestVanitiesAll: + @responses.activate + def test_all_sends_get_request(self): + base_url = "http://connect.example/__api__" + endpoint = f"{base_url}/v1/vanities" + mock_get = responses.get(endpoint, json=[]) + + session = requests.Session() + url = Url(base_url) + params = ResourceParameters(session, url) + vanities = Vanities(params) + + vanities.all() + + assert mock_get.call_count == 1 + + +class TestVanityMixin: + @responses.activate + def test_vanity_getter_returns_vanity(self): + guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" + base_url = "http://connect.example/__api__" + endpoint = f"{base_url}/v1/content/{guid}/vanity" + mock_get = responses.get(endpoint, json={"content_guid": guid, "path": "my-dashboard"}) + + session = requests.Session() + url = Url(base_url) + params = ResourceParameters(session, url) + content = VanityMixin(params, guid=guid) + + assert content.vanity == "my-dashboard" + assert mock_get.call_count == 1 + + @responses.activate + def test_vanity_setter_with_string(self): + guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" + base_url = "http://connect.example/__api__" + endpoint = f"{base_url}/v1/content/{guid}/vanity" + path = "example" + mock_put = responses.put( + endpoint, + json={"content_guid": guid, "path": path}, + match=[json_params_matcher({"path": path})], + ) + + session = requests.Session() + url = Url(base_url) + params = ResourceParameters(session, url) + content = VanityMixin(params, guid=guid) + content.vanity = path + assert content.vanity == path + + assert mock_put.call_count == 1 + + @responses.activate + def test_vanity_deleter(self): + guid = "8ce6eaca-60af-4c2f-93a0-f5f3cddf5ee5" + base_url = "http://connect.example/__api__" + endpoint = f"{base_url}/v1/content/{guid}/vanity" + mock_delete = responses.delete(endpoint) + + session = requests.Session() + url = Url(base_url) + params = ResourceParameters(session, url) + content = VanityMixin(params, guid=guid) + content._vanity = Vanity(params, path=Mock(), content_guid=guid, created_time=Mock()) + del content.vanity + + assert content._vanity is None + assert mock_delete.call_count == 1