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

Add missing function body import in client.find_automation #16824

Merged
merged 1 commit into from
Jan 23, 2025
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
4 changes: 4 additions & 0 deletions src/prefect/client/orchestration/_automations/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ def read_automations(self) -> list["Automation"]:
return Automation.model_validate_list(response.json())

def find_automation(self, id_or_name: "str | UUID") -> "Automation | None":
from uuid import UUID

if isinstance(id_or_name, str):
name = id_or_name
try:
Expand Down Expand Up @@ -202,6 +204,8 @@ async def read_automations(self) -> list["Automation"]:
return Automation.model_validate_list(response.json())

async def find_automation(self, id_or_name: "str | UUID") -> "Automation | None":
from uuid import UUID

if isinstance(id_or_name, str):
name = id_or_name
try:
Expand Down
31 changes: 31 additions & 0 deletions tests/test_automations.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from prefect.automations import Automation, DoNothing
from prefect.events import ResourceSpecification
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.exceptions import ObjectNotFound
from prefect.settings import PREFECT_API_SERVICES_TRIGGERS_ENABLED, temporary_settings


Expand Down Expand Up @@ -206,3 +207,33 @@ def test_disabled_automation_can_be_enabled_sync(automation: Automation):

updated_automation = Automation.read(id=automation.id)
assert updated_automation.enabled is True


async def test_find_automation(automation: Automation):
from prefect.client.orchestration import get_client

client = get_client()

# Test finding by UUID
found = await client.find_automation(automation.id)
assert found == automation

# Test finding by UUID string
found = await client.find_automation(str(automation.id))
assert found == automation

# Test finding by exact name
found = await client.find_automation(automation.name)
assert found == automation

# Test finding by case-insensitive name
found = await client.find_automation(automation.name.upper())
assert found == automation

# Test finding nonexistent UUID raises ObjectNotFound
with pytest.raises(ObjectNotFound):
await client.find_automation(UUID("6d222a09-3c68-42d4-b019-bd331a3abb88"))

# Test finding nonexistent name returns None
found = await client.find_automation("nonexistent_name")
assert found is None
Loading