Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add find_by method to content #296

Merged
merged 4 commits into from
Sep 19, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions integration/tests/posit/connect/test_content.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@ def test_get(self):
def test_find(self):
assert self.client.content.find()

def test_find_by(self):
assert self.client.content.find_by(guid=self.content["guid"]) == self.content

def test_find_one(self):
assert self.client.content.find_one()

Expand Down
125 changes: 125 additions & 0 deletions src/posit/connect/content.py
Original file line number Diff line number Diff line change
Expand Up @@ -534,6 +534,131 @@ def find(self, include: Optional[str | list[Any]] = None, **conditions) -> List[
for result in response.json()
]

@overload
def find_by(
self,
*,
# Required
name: str,
# Content Metadata
title: Optional[str] = None,
description: Optional[str] = None,
access_type: Literal["all", "acl", "logged_in"] = "acl",
owner_guid: Optional[str] = None,
# Timeout Settings
connection_timeout: Optional[int] = None,
read_timeout: Optional[int] = None,
init_timeout: Optional[int] = None,
idle_timeout: Optional[int] = None,
# Process and Resource Limits
max_processes: Optional[int] = None,
min_processes: Optional[int] = None,
max_conns_per_process: Optional[int] = None,
load_factor: Optional[float] = None,
cpu_request: Optional[float] = None,
cpu_limit: Optional[float] = None,
memory_request: Optional[int] = None,
memory_limit: Optional[int] = None,
amd_gpu_limit: Optional[int] = None,
nvidia_gpu_limit: Optional[int] = None,
# Execution Settings
run_as: Optional[str] = None,
run_as_current_user: Optional[bool] = False,
default_image_name: Optional[str] = None,
default_r_environment_management: Optional[bool] = None,
default_py_environment_management: Optional[bool] = None,
service_account_name: Optional[str] = None,
) -> Optional[ContentItem]:
"""Find the first content record matching the specified attributes. There is no implied ordering so if order matters, you should find it yourself.

Parameters
----------
name : str, optional
URL-friendly identifier. Allows alphanumeric characters, hyphens ("-"), and underscores ("_").
title : str, optional
Content title. Default is None
description : str, optional
Content description.
access_type : Literal['all', 'acl', 'logged_in'], optional
How content manages viewers.
owner_guid : str, optional
The unique identifier of the user who owns this content item.
connection_timeout : int, optional
Max seconds without data exchange.
read_timeout : int, optional
Max seconds without data received.
init_timeout : int, optional
Max startup time for interactive apps.
idle_timeout : int, optional
Max idle time before process termination.
max_processes : int, optional
Max concurrent processes allowed.
min_processes : int, optional
Min concurrent processes required.
max_conns_per_process : int, optional
Max client connections per process.
load_factor : float, optional
Aggressiveness in spawning new processes (0.0 - 1.0).
cpu_request : float, optional
Min CPU units required (1 unit = 1 core).
cpu_limit : float, optional
Max CPU units allowed.
memory_request : int, optional
Min memory (bytes) required.
memory_limit : int, optional
Max memory (bytes) allowed.
amd_gpu_limit : int, optional
Number of AMD GPUs allocated.
nvidia_gpu_limit : int, optional
Number of NVIDIA GPUs allocated.
run_as : str, optional
UNIX user to execute the content.
run_as_current_user : bool, optional
Run process as the visiting user (for app content). Default is False.
default_image_name : str, optional
Default image for execution if not defined in the bundle.
default_r_environment_management : bool, optional
Manage R environment for the content.
default_py_environment_management : bool, optional
Manage Python environment for the content.
service_account_name : str, optional
Kubernetes service account name for running content.

Returns
-------
Optional[ContentItem]
"""
...

@overload
def find_by(self, **attributes) -> Optional[ContentItem]:
"""Find the first content record matching the specified attributes. There is no implied ordering so if order matters, you should find it yourself.

Returns
-------
Optional[ContentItem]
"""
...

def find_by(self, **attributes) -> Optional[ContentItem]:
"""Find the first content record matching the specified attributes. There is no implied ordering so if order matters, you should find it yourself.

Returns
-------
Optional[ContentItem]

Example
-------
>>> find_by(name="example-content-name")
"""
results = self.find()
results = (
result
for result in results
if all(item in result.items() for item in attributes.items())
)
return next(results, None)

@overload
def find_one(
self,
Expand Down
39 changes: 39 additions & 0 deletions tests/posit/connect/test_content.py
Original file line number Diff line number Diff line change
Expand Up @@ -406,6 +406,45 @@ def test_params_include_none(self):
assert mock_get.call_count == 1


class TestContentsFindBy:
@responses.activate
def test(self):
# behavior
mock_get = responses.get(
"https://connect.example/__api__/v1/content",
json=load_mock("v1/content.json"),
)

# setup
client = Client("https://connect.example", "12345")

# invoke
content = client.content.find_by(name="team-admin-dashboard")

# assert
assert mock_get.call_count == 1
assert content
assert content.name == "team-admin-dashboard"

@responses.activate
def test_miss(self):
# behavior
mock_get = responses.get(
"https://connect.example/__api__/v1/content",
json=load_mock("v1/content.json"),
)

# setup
client = Client("https://connect.example", "12345")

# invoke
content = client.content.find_by(name="does-not-exist")
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if I tried passing in an invalid keyword arg through the **attributes parameter, would anything bad happen?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nothing bad. You just won't get any results.


# assert
assert mock_get.call_count == 1
assert content is None


class TestContentsFindOne:
@responses.activate
def test(self):
Expand Down