Skip to content

prefect.utilities.urls

convert_class_to_name(obj)

Convert CamelCase class name to dash-separated lowercase name

Source code in src/prefect/utilities/urls.py
61
62
63
64
65
66
67
def convert_class_to_name(obj: Any) -> str:
    """
    Convert CamelCase class name to dash-separated lowercase name
    """
    cls = obj if inspect.isclass(obj) else obj.__class__
    name = cls.__name__
    return "".join(["-" + i.lower() if i.isupper() else i for i in name]).lstrip("-")

url_for(obj, obj_id=None, url_type='ui', default_base_url=None)

Returns the URL for a Prefect object.

Pass in a supported object directly or provide an object name and ID.

Parameters:

Name Type Description Default
obj Union[PrefectFuture, Block, Variable, Automation, Resource, ReceivedEvent, BaseModel, str]

A Prefect object to get the URL for, or its URL name and ID.

required
obj_id Union[str, UUID]

The UUID of the object.

None
url_type Literal['ui', 'api']

Whether to return the URL for the UI (default) or API.

'ui'
default_base_url str

The default base URL to use if no URL is configured.

None

Returns:

Type Description
Optional[str]

Optional[str]: The URL for the given object or None if the object is not supported.

Examples:

url_for(my_flow_run) url_for(obj=my_flow_run) url_for("flow-run", obj_id="123e4567-e89b-12d3-a456-426614174000")

Source code in src/prefect/utilities/urls.py
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def url_for(
    obj: Union[
        PrefectFuture,
        Block,
        Variable,
        Automation,
        Resource,
        ReceivedEvent,
        BaseModel,
        str,
    ],
    obj_id: Optional[Union[str, UUID]] = None,
    url_type: URLType = "ui",
    default_base_url: Optional[str] = None,
) -> Optional[str]:
    """
    Returns the URL for a Prefect object.

    Pass in a supported object directly or provide an object name and ID.

    Args:
        obj (Union[PrefectFuture, Block, Variable, Automation, Resource, ReceivedEvent, BaseModel, str]):
            A Prefect object to get the URL for, or its URL name and ID.
        obj_id (Union[str, UUID], optional):
            The UUID of the object.
        url_type (Literal["ui", "api"], optional):
            Whether to return the URL for the UI (default) or API.
        default_base_url (str, optional):
            The default base URL to use if no URL is configured.

    Returns:
        Optional[str]: The URL for the given object or None if the object is not supported.

    Examples:
        url_for(my_flow_run)
        url_for(obj=my_flow_run)
        url_for("flow-run", obj_id="123e4567-e89b-12d3-a456-426614174000")
    """
    if isinstance(obj, PrefectFuture):
        name = "task-run"
    elif isinstance(obj, Block):
        name = "block"
    elif isinstance(obj, Automation):
        name = "automation"
    elif isinstance(obj, ReceivedEvent):
        name = "received-event"
    elif isinstance(obj, Resource):
        if obj.id.startswith("prefect."):
            name = obj.id.split(".")[1]
        else:
            logger.debug(f"No URL known for resource with ID: {obj.id}")
            return None
    elif isinstance(obj, str):
        name = obj
    else:
        name = convert_class_to_name(obj)

    # Can't do an isinstance check here because the client build
    # doesn't have access to that server schema.
    if name == "work-queue-with-status":
        name = "work-queue"

    if url_type != "ui" and url_type != "api":
        raise ValueError(f"Invalid URL type: {url_type}. Use 'ui' or 'api'.")

    if url_type == "ui" and name not in UI_URL_FORMATS:
        logger.debug("No UI URL known for this object: %s", name)
        return None
    elif url_type == "api" and name not in API_URL_FORMATS:
        logger.debug("No API URL known for this object: %s", name)
        return None

    if isinstance(obj, str) and not obj_id:
        raise ValueError(
            "If passing an object name, you must also provide an object ID."
        )

    base_url = (
        settings.PREFECT_UI_URL.value()
        if url_type == "ui"
        else settings.PREFECT_API_URL.value()
    )
    base_url = base_url or default_base_url

    if not base_url:
        logger.debug(
            f"No URL found for the Prefect {'UI' if url_type == 'ui' else 'API'}, "
            f"and no default base path provided."
        )
        return None

    if not obj_id:
        # We treat PrefectFuture as if it was the underlying task run,
        # so we need to check the object type here instead of name.
        if isinstance(obj, PrefectFuture):
            obj_id = getattr(obj, "task_run_id", None)
        elif name == "block":
            # Blocks are client-side objects whose API representation is a
            # BlockDocument.
            obj_id = obj._block_document_id
        elif name in ("variable", "work-pool"):
            obj_id = obj.name
        elif isinstance(obj, Resource):
            obj_id = obj.id.rpartition(".")[2]
        else:
            obj_id = getattr(obj, "id", None)
        if not obj_id:
            logger.debug(
                "An ID is required to build a URL, but object did not have one: %s", obj
            )
            return ""

    url_format = (
        UI_URL_FORMATS.get(name) if url_type == "ui" else API_URL_FORMATS.get(name)
    )

    if isinstance(obj, ReceivedEvent):
        url = url_format.format(
            occurred=obj.occurred.strftime("%Y-%m-%d"), obj_id=obj_id
        )
    else:
        url = url_format.format(obj_id=obj_id)

    if not base_url.endswith("/"):
        base_url += "/"
    return urllib.parse.urljoin(base_url, url)