Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/events/jinja_filters.py: 31%

1from __future__ import annotations 

2 

3import re 

4import urllib.parse 

5from typing import Any, Callable, Mapping 

6 

7from jinja2 import pass_context 

8 

9from prefect.server.schemas.core import ( 

10 Deployment, 

11 Flow, 

12 FlowRun, 

13 TaskRun, 

14 WorkPool, 

15 WorkQueue, 

16) 

17from prefect.server.schemas.responses import WorkQueueWithStatus 

18from prefect.server.utilities.schemas import ORMBaseModel 

19from prefect.settings import PREFECT_UI_URL 

20from prefect.utilities.urls import url_for 

21 

22model_to_kind: dict[type[ORMBaseModel], str] = { 

23 Deployment: "prefect.deployment", 

24 Flow: "prefect.flow", 

25 FlowRun: "prefect.flow-run", 

26 TaskRun: "prefect.task-run", 

27 WorkQueue: "prefect.work-queue", 

28 WorkQueueWithStatus: "prefect.work-queue", 

29 WorkPool: "prefect.work-pool", 

30} 

31 

32 

33@pass_context 

34def ui_url(ctx: Mapping[str, Any], obj: Any) -> str | None: 

35 """Return the UI URL for the given object.""" 

36 return url_for(obj, url_type="ui") 

37 

38 

39@pass_context 

40def ui_resource_events_url(ctx: Mapping[str, Any], obj: Any) -> str | None: 

41 """Given a Resource or Model, return a UI link to the events page 

42 filtered for that resource. If an unsupported object is provided, 

43 return `None`. 

44 

45 Currently supports Automation, Resource, Deployment, Flow, FlowRun, TaskRun, and 

46 WorkQueue objects. Within a Resource, deployment, flow, flow-run, task-run, 

47 and work-queue are supported.""" 

48 from prefect.server.events.schemas.automations import Automation 

49 from prefect.server.events.schemas.events import Resource 

50 

51 url = None 

52 url_format = "events?resource={resource_id}" 

53 

54 if isinstance(obj, Automation): 

55 url = url_format.format(resource_id=f"prefect.automation.{obj.id}") 

56 elif isinstance(obj, Resource): 

57 kind, _, id = obj.id.rpartition(".") 

58 url = url_format.format(resource_id=f"{kind}.{id}") 

59 elif isinstance(obj, ORMBaseModel): 

60 kind = model_to_kind.get(type(obj)) # type: ignore 

61 

62 if kind: 

63 url = url_format.format(resource_id=f"{kind}.{obj.id}") 

64 if url: 

65 return urllib.parse.urljoin(PREFECT_UI_URL.value(), url) 

66 else: 

67 return None 

68 

69 

70def flow_run_id(text: str | None) -> str | None: 

71 """Extract a flow run ID from a string, such as a PR body.""" 

72 if not text: 

73 return None 

74 match = re.search( 

75 r"flow-run/([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})", 

76 text, 

77 ) 

78 if match: 

79 return match.group(1) 

80 return None 

81 

82 

83all_filters: dict[str, Callable[[Mapping[str, Any], Any], str | None]] = { 

84 "ui_url": ui_url, 

85 "ui_resource_events_url": ui_resource_events_url, 

86 "flow_run_id": flow_run_id, # type: ignore 

87}