feat: add environments

integration/tests/posit/connect/test_environments.py

@@ -0,0 +1,36 @@
+import pytest
+from packaging import version
+
+from posit import connect
+
+from . import CONNECT_VERSION
+
+
+@pytest.mark.skipif(
+    CONNECT_VERSION <= version.parse("2023.01.1"),
+    reason="Environments API not available",
+)
+class TestContent:
+    @classmethod
+    def setup_class(cls):
+        cls.client = connect.Client()
+        cls.environment = cls.client.environments.create(
+            title="Title", cluster_name="Kubernetes", name="Name"
+        )
+
+    @classmethod
+    def teardown_class(cls):
+        cls.environment.destroy()
+
+    def test_find(self):
+        uid = self.environment["guid"]
+        environment = self.client.environments.find(uid)
+        assert environment == self.environment
+
+    def test_find_by(self):
+        environment = self.client.environments.find_by(name="Name")
+        assert environment == self.environment
+
+    def test_update(self):
+        self.environment.update(title="New Title")
+        assert self.environment["title"] == "New Title"

integration/tests/posit/connect/test_jobs.py

@@ -0,0 +1,38 @@
+from pathlib import Path
+
+import pytest
+from packaging import version
+
+from posit import connect
+
+from . import CONNECT_VERSION
+
+
+class TestJobs:
+    @classmethod
+    def setup_class(cls):
+        cls.client = connect.Client()
+        cls.content = cls.client.content.create(name="example-quarto-minimal")
+
+    @classmethod
+    def teardown_class(cls):
+        cls.content.delete()
+        assert cls.client.content.count() == 0
+
+    @pytest.mark.skipif(
+        CONNECT_VERSION <= version.parse("2023.01.1"),
+        reason="Quarto not available",
+    )
+    def test(self):
+        content = self.content
+
+        path = Path("../../../resources/connect/bundles/example-quarto-minimal/bundle.tar.gz")
+        path = Path(__file__).parent / path
+        path = path.resolve()
+        path = str(path)
+
+        bundle = content.bundles.create(path)
+        bundle.deploy()
+
+        jobs = content.jobs
+        assert len(jobs) == 1

src/posit/connect/client.py

@@ -6,6 +6,8 @@
 
 from requests import Response, Session
 
+from posit.connect.environments import Environments
+
 from . import hooks, me
 from .auth import Auth
 from .config import Config
@@ -184,6 +186,11 @@ def me(self) -> User:
         """
         return me.get(self.resource_params)
 
+    @property
+    @requires(version="2023.05.0")
+    def environments(self) -> Environments:
+        return Environments(self.ctx, "v1")
+
     @property
     def groups(self) -> Groups:
         """The groups resource interface.

src/posit/connect/content.py

@@ -9,7 +9,9 @@
 
 from . import tasks
 from .bundles import Bundles
+from .context import Context
 from .env import EnvVars
+from .jobs import JobsMixin
 from .oauth.associations import ContentItemAssociations
 from .permissions import Permissions
 from .resources import Resource, ResourceParameters, Resources
@@ -32,7 +34,13 @@ class ContentItemOwner(Resource):
     pass
 
 
-class ContentItem(VanityMixin, Resource):
+class ContentItem(JobsMixin, VanityMixin, Resource):
+    def __init__(self, /, params: ResourceParameters, **kwargs):
+        ctx = Context(params.session, params.url)
+        path = f"v1/content"
+        pathinfo = kwargs["guid"]
+        super().__init__(ctx, path, pathinfo, **kwargs)
+
     def __getitem__(self, key: Any) -> Any:
         v = super().__getitem__(key)
         if key == "owner" and isinstance(v, dict):

src/posit/connect/context.py

@@ -4,6 +4,8 @@
 import requests
 from packaging.version import Version
 
+from .urls import Url
+
 
 def requires(version: str):
     def decorator(func):
@@ -22,7 +24,7 @@ def wrapper(instance: ContextManager, *args, **kwargs):
 
 
 class Context(dict):
-    def __init__(self, session: requests.Session, url: str):
+    def __init__(self, session: requests.Session, url: Url):
         self.session = session
         self.url = url
 
@@ -38,7 +40,7 @@ def version(self) -> Optional[str]:
         return value
 
     @version.setter
-    def version(self, value: str):
+    def version(self, value):
         self["version"] = value
 
 

src/posit/connect/environments.py

@@ -0,0 +1,86 @@
+from typing import Dict, Literal, Optional, TypedDict, overload
+
+from typing_extensions import NotRequired, Required, Unpack
+
+from .resources import (
+    ActiveCreatorMethods,
+    ActiveDestroyerMethods,
+    ActiveFinderMethods,
+    ActiveUpdaterMethods,
+)
+
+_Kubernetes = Literal["Kubernetes"]
+_Matching = Literal["any", "exact", "none"]
+_Installation = TypedDict("_Installation", { 'path': Required[str], 'version': Required[str] })
+_Installations = Dict[Literal["installations"], _Installation]
+
+
+class Environment(ActiveUpdaterMethods, ActiveDestroyerMethods):
+    class _UpdateEnvironment(TypedDict, total=False):
+        title: Required[str]
+        description: NotRequired[Optional[str]]
+        matching: NotRequired[Optional[_Matching]]
+        supervisor: NotRequired[Optional[str]]
+        python: NotRequired[_Installations]
+        quarto: NotRequired[_Installations]
+        r: NotRequired[_Installations]
+        tensorflow: NotRequired[_Installations]
+
+    @overload
+    def update(self, /, **attributes: Unpack[_UpdateEnvironment]): ...
+
+    @overload
+    def update(self, /, **attributes): ...
+
+    def update(self, /, **attributes):
+        return super().update(**attributes)
+
+
+class Environments(ActiveFinderMethods[Environment], ActiveCreatorMethods[Environment]):
+    def __init__(self, ctx, path, pathinfo="environments", uid="guid"):
+        super().__init__(ctx, path, pathinfo, uid)
+
+    class _CreateEnvironment(TypedDict):
+        title: Required[str]
+        description: NotRequired[Optional[str]]
+        cluster_name: Required[_Kubernetes]
+        name: Required[str]
+        matching: NotRequired[Optional[_Matching]]
+        supervisor: NotRequired[Optional[str]]
+        python: NotRequired[_Installations]
+        quarto: NotRequired[_Installations]
+        r: NotRequired[_Installations]
+        tensorflow: NotRequired[_Installations]
+
+    @overload
+    def create(self, **attributes: Unpack[_CreateEnvironment]) -> Environment: ...
+
+    @overload
+    def create(self, **attributes) -> Environment: ...
+
+    def create(self, **attributes) -> Environment:
+        return super().create(**attributes)
+
+    class _FindByEnvironment(TypedDict):
+        title: NotRequired[str]
+        description: NotRequired[Optional[str]]
+        cluster_name: NotRequired[_Kubernetes]
+        name: NotRequired[str]
+        matching: NotRequired[Optional[_Matching]]
+        supervisor: NotRequired[Optional[str]]
+        python: NotRequired[_Installations]
+        quarto: NotRequired[_Installations]
+        r: NotRequired[_Installations]
+        tensorflow: NotRequired[_Installations]
+
+    @overload
+    def find_by(self, **conditions: Unpack[_FindByEnvironment]): ...
+
+    @overload
+    def find_by(self, **conditions): ...
+
+    def find_by(self, **conditions):
+        return super().find_by(**conditions)
+
+    def _create_instance(self, path, pathinfo, /, **attributes):
+        return Environment(self._ctx, path, pathinfo, **attributes)