Merge branch 'main' into patch-1

docs/api-ref/server/index.md

@@ -7,6 +7,6 @@ tags:
 
 # Server API 
 
-The Prefect Server API is used by the server to work with workflow metadata and enforce orchestration logic. This API is primarily used by Prefect developers.
+The Prefect server API is used by the server to work with workflow metadata and enforce orchestration logic. This API is primarily used by Prefect developers.
 
 Select links in the left navigation menu to explore. 

docs/concepts/variables.md

@@ -7,7 +7,7 @@ tags:
 
 # Variables
 
-Variables enable you to store and reuse non-sensitive bits of data, such as configuration information. Variables are named, mutable string values, much like environment variables. Variables are scoped to a Prefect Server instance or a single workspace in Prefect Cloud.
+Variables enable you to store and reuse non-sensitive bits of data, such as configuration information. Variables are named, mutable string values, much like environment variables. Variables are scoped to a Prefect server instance or a single workspace in Prefect Cloud.
 
 Variables can be created or modified at any time, but are intended for values with infrequent writes and frequent reads. Variable values may be cached for quicker retrieval.
 
@@ -32,7 +32,7 @@ Optionally, you can add tags to the variable.
 
 ### Via the Prefect UI
 
-You can see all the variables in your Prefect Server instance or Prefect Cloud workspace on the **Variables** page of the Prefect UI. Both the name and value of all variables are visible to anyone with access to the server or workspace.
+You can see all the variables in your Prefect server instance or Prefect Cloud workspace on the **Variables** page of the Prefect UI. Both the name and value of all variables are visible to anyone with access to the server or workspace.
 
 To create a new variable, select the **+** button next to the header of the **Variables** page. Enter the name and value of the variable.
 

docs/host/index.md

@@ -12,11 +12,11 @@ tags:
     - SQLite 
 ---
 
-# Hosting Prefect Server
+# Hosting a Prefect server
 
 After you install Prefect you have a Python SDK client that can communicate with [Prefect Cloud](https://app.prefect.cloud), the platform hosted by Prefect. You also have an [API server](/api-ref/) backed by a database and a UI. 
 
-In this section you'll learn how to host your own Prefect Server.
+In this section you'll learn how to host your own Prefect server.
 
 ![Prefect Server UI](/img/ui/flow-run-page-server.png)
 
@@ -37,7 +37,7 @@ Shut down the Prefect server with <kdb> ctrl </kbd> + <kdb> c </kbd> in the term
 
 ### Differences between Prefect Server and Cloud
 
-Prefect Server and Cloud share a base of features. Prefect Cloud also includes the following features that you can read about in the [Cloud](/cloud/) section of the docs. 
+The self-hosted Prefect server and Prefect Cloud share a base of features. Prefect Cloud also includes the following features that you can read about in the [Cloud](/cloud/) section of the docs. 
 
 - [User accounts](#user-accounts) &mdash; personal accounts for working in Prefect Cloud. 
 - [Workspaces](/cloud/workspaces/) &mdash; isolated environments to organize your flows, deployments, and flow runs.
@@ -51,7 +51,7 @@ Prefect Server and Cloud share a base of features. Prefect Cloud also includes t
 - Collaborators &mdash; invite others to work in your [workspace](/cloud/workspaces/#workspace-collaborators) or [organization](/cloud/organizations/#organization-members).
 
 
-### Configuring Prefect Server
+### Configuring a Prefect Server
 
 Go to your terminal session and run this command to set the API URL to point to a Prefect server instance:
 
@@ -63,7 +63,7 @@ $ prefect config set PREFECT_API_URL="http://127.0.0.1:4200/api"
 
 
 !!! tip "`PREFECT_API_URL` required when running Prefect inside a container"
-    You must set the API Server address to use Prefect within a container, such a a Docker container. 
+    You must set the API server address to use Prefect within a container, such as a Docker container. 
 
     You can save the API server address in a [Prefect profile](/concepts/settings/). Whenever that profile is active, the API endpoint will be be at that address.
 
@@ -229,7 +229,7 @@ See the [contributing docs](/contributing/overview/#adding-database-migrations)
 
 ## Notifications
 
-When you use [Prefect Cloud](/cloud/) you gain access to a hosted platform with Workspace & User controls, Events, and Automations. Prefect Cloud has an option for automation notifications. The more limited Notifications option is provided for the self-hosted Prefect Server.
+When you use [Prefect Cloud](/cloud/) you gain access to a hosted platform with Workspace & User controls, Events, and Automations. Prefect Cloud has an option for automation notifications. The more limited Notifications option is provided for the self-hosted Prefect server.
 
 Notifications enable you to set up alerts that are sent when a flow enters any state you specify. When your flow and task runs changes [state](/concepts/states/), Prefect notes the state change and checks whether the new state matches any notification policies. If it does, a new notification is queued.
 
@@ -247,7 +247,7 @@ Prefect supports sending notifications via:
 
 ### Configure notifications
 
-To configure a notification in Prefect Server, go to the **Notifications** page and select **Create Notification** or the **+** button. 
+To configure a notification in a Prefect server, go to the **Notifications** page and select **Create Notification** or the **+** button. 
 
 ![Creating a notification in the Prefect UI](/img/ui/create-slack-notification.png)
 

src/prefect/agent.py

@@ -378,6 +378,7 @@ async def _mark_flow_run_as_cancelled(
 
     async def get_infrastructure(self, flow_run: FlowRun) -> Infrastructure:
         deployment = await self.client.read_deployment(flow_run.deployment_id)
+
         flow = await self.client.read_flow(deployment.flow_id)
 
         # overrides only apply when configuring known infra blocks

src/prefect/artifacts.py

@@ -31,11 +31,13 @@ async def _create_artifact(
     """
     Helper function to create an artifact.
 
-    Args:
-        - type:  A string identifying the type of artifact.
-        - key: A string user-provided identifier.
-        - description: A user-specified description of the artifact.
-        - data: A JSON payload that allows for a result to be retrieved.
+    Arguments:
+        type: A string identifying the type of artifact.
+        key: A user-provided string identifier.
+          The key must only contain lowercase letters, numbers, and dashes.
+        description: A user-specified description of the artifact.
+        data: A JSON payload that allows for a result to be retrieved.
+        client: The PrefectClient
 
     Returns:
         - The table artifact ID.
@@ -74,11 +76,17 @@ async def create_link_artifact(
     """
     Create a link artifact.
 
-    Args:
-        - link: The link to create.
+    Arguments:
+        link: The link to create.
+        link_text: The link text.
+        key: A user-provided string identifier.
+          Required for the artifact to show in the Artifacts page in the UI.
+          The key must only contain lowercase letters, numbers, and dashes.
+        description: A user-specified description of the artifact.
+
 
     Returns:
-        - The table artifact ID.
+        The table artifact ID.
     """
     formatted_link = f"[{link_text}]({link})" if link_text else f"[{link}]({link})"
     artifact = await _create_artifact(
@@ -100,11 +108,15 @@ async def create_markdown_artifact(
     """
     Create a markdown artifact.
 
-    Args:
-        - markdown: The markdown to create.
+    Arguments:
+        markdown: The markdown to create.
+        key: A user-provided string identifier.
+          Required for the artifact to show in the Artifacts page in the UI.
+          The key must only contain lowercase letters, numbers, and dashes.
+        description: A user-specified description of the artifact.
 
     Returns:
-        - The table artifact ID.
+        The table artifact ID.
     """
     artifact = await _create_artifact(
         key=key,
@@ -125,11 +137,15 @@ async def create_table_artifact(
     """
     Create a table artifact.
 
-    Args:
-        - table: The table to create.
+    Arguments:
+        table: The table to create.
+        key: A user-provided string identifier.
+          Required for the artifact to show in the Artifacts page in the UI.
+          The key must only contain lowercase letters, numbers, and dashes.
+        description: A user-specified description of the artifact.
 
     Returns:
-        - The table artifact ID.
+        The table artifact ID.
     """
 
     def _sanitize_nan_values(container):

src/prefect/cli/config.py

@@ -38,6 +38,13 @@ def set_(settings: List[str]):
         if setting not in prefect.settings.SETTING_VARIABLES:
             exit_with_error(f"Unknown setting name {setting!r}.")
 
+        # Guard against changing settings that tweak config locations
+        if setting in {"PREFECT_HOME", "PREFECT_PROFILES_PATH"}:
+            exit_with_error(
+                f"Setting {setting!r} cannot be changed with this command. "
+                "Use an environment variable instead."
+            )
+
         parsed_settings[setting] = value
 
     try:

src/prefect/client/orchestration.py

@@ -1497,7 +1497,13 @@ async def read_deployment(
         Returns:
             a [Deployment model][prefect.server.schemas.core.Deployment] representation of the deployment
         """
-        response = await self._client.get(f"/deployments/{deployment_id}")
+        try:
+            response = await self._client.get(f"/deployments/{deployment_id}")
+        except httpx.HTTPStatusError as e:
+            if e.response.status_code == status.HTTP_404_NOT_FOUND:
+                raise prefect.exceptions.ObjectNotFound(http_exc=e) from e
+            else:
+                raise
         return schemas.responses.DeploymentResponse.parse_obj(response.json())
 
     async def read_deployment_by_name(

src/prefect/server/schemas/actions.py

@@ -644,7 +644,7 @@ class VariableUpdate(ActionBaseModel):
         default=None,
         description="The value of the variable",
         example="my-value",
-        max_length=schemas.core.MAX_VARIABLE_NAME_LENGTH,
+        max_length=schemas.core.MAX_VARIABLE_VALUE_LENGTH,
     )
     tags: Optional[List[str]] = FieldFrom(schemas.core.Variable)
 

src/prefect/workers/base.py

@@ -565,7 +565,13 @@ async def cancel_run(self, flow_run: "FlowRun"):
             )
             return
 
-        configuration = await self._get_configuration(flow_run)
+        try:
+            configuration = await self._get_configuration(flow_run)
+        except ObjectNotFound:
+            self._logger.warning(
+                f"Flow run {flow_run.id!r} cannot be cancelled by this worker:"
+                f" associated deployment {flow_run.deployment_id!r} does not exist."
+            )
 
         try:
             await self.kill_infrastructure(
@@ -740,7 +746,7 @@ async def _submit_run(self, flow_run: "FlowRun") -> None:
 
         try:
             await self._check_flow_run(flow_run)
-        except ValueError:
+        except (ValueError, ObjectNotFound):
             self._logger.exception(
                 (
                     "Flow run %s did not pass checks and will not be submitted for"

tests/cli/test_config.py

@@ -74,6 +74,19 @@ def test_set_with_unknown_setting():
     )
 
 
+@pytest.mark.parametrize("setting", ["PREFECT_HOME", "PREFECT_PROFILES_PATH"])
+def test_set_with_disallowed_setting(setting):
+    save_profiles(ProfilesCollection([Profile(name="foo", settings={})], active=None))
+
+    invoke_and_assert(
+        ["--profile", "foo", "config", "set", f"{setting}=BAR"],
+        expected_output=f"""                
+            Setting {setting!r} cannot be changed with this command. Use an environment variable instead.
+            """,
+        expected_code=1,
+    )
+
+
 def test_set_with_invalid_value_type():
     save_profiles(ProfilesCollection([Profile(name="foo", settings={})], active=None))
 

tests/client/test_orion_client.py

@@ -722,7 +722,7 @@ def foo():
     )
 
     await orion_client.delete_deployment(deployment_id)
-    with pytest.raises(httpx.HTTPStatusError, match="404"):
+    with pytest.raises(prefect.exceptions.ObjectNotFound):
         await orion_client.read_deployment(deployment_id)
 
 

tests/server/api/test_variables.py

@@ -534,6 +534,54 @@ async def test_name_unique(
         assert res
         assert res.status_code == 409
 
+    async def test_name_max_length(
+        self,
+        client: AsyncClient,
+        variable,
+    ):
+        max_length = 255
+
+        res = await client.patch(
+            f"/variables/{variable.id}", json={"name": "v" * max_length}
+        )
+        assert res
+        assert res.status_code == 204
+
+        max_length_plus1 = max_length + 1
+
+        res = await client.patch(
+            f"/variables/{variable.id}", json={"name": "v" * max_length_plus1}
+        )
+        assert res
+        assert res.status_code == 422
+        assert (
+            "ensure this value has at most" in res.json()["exception_detail"][0]["msg"]
+        )
+
+    async def test_value_max_length(
+        self,
+        client: AsyncClient,
+        variable,
+    ):
+        max_length = 5000
+
+        res = await client.patch(
+            f"/variables/{variable.id}", json={"value": "v" * max_length}
+        )
+        assert res
+        assert res.status_code == 204
+
+        max_length_plus1 = max_length + 1
+
+        res = await client.patch(
+            f"/variables/{variable.id}", json={"value": "v" * max_length_plus1}
+        )
+        assert res
+        assert res.status_code == 422
+        assert (
+            "ensure this value has at most" in res.json()["exception_detail"][0]["msg"]
+        )
+
 
 class TestUpdateVariableByName:
     async def test_update_variable(
@@ -584,6 +632,54 @@ async def test_name_unique(
         )
         assert res.status_code == 409
 
+    async def test_name_max_length(
+        self,
+        client: AsyncClient,
+        variable,
+    ):
+        max_length = 255
+
+        res = await client.patch(
+            f"/variables/name/{variable.name}", json={"name": "v" * max_length}
+        )
+        assert res
+        assert res.status_code == 204
+
+        max_length_plus1 = max_length + 1
+
+        res = await client.patch(
+            f"/variables/name/{variable.name}", json={"name": "v" * max_length_plus1}
+        )
+        assert res
+        assert res.status_code == 422
+        assert (
+            "ensure this value has at most" in res.json()["exception_detail"][0]["msg"]
+        )
+
+    async def test_value_max_length(
+        self,
+        client: AsyncClient,
+        variable,
+    ):
+        max_length = 5000
+
+        res = await client.patch(
+            f"/variables/name/{variable.name}", json={"value": "v" * max_length}
+        )
+        assert res
+        assert res.status_code == 204
+
+        max_length_plus1 = max_length + 1
+
+        res = await client.patch(
+            f"/variables/name/{variable.name}", json={"value": "v" * max_length_plus1}
+        )
+        assert res
+        assert res.status_code == 422
+        assert (
+            "ensure this value has at most" in res.json()["exception_detail"][0]["msg"]
+        )
+
 
 class TestDeleteVariable:
     async def test_delete_variable(