Merge branch 'main' into jean/oss-5403-create-and-synchronize-a-globa…

…l-concurrency-limit-for

docs/3.0/manage/cloud/manage-users/secure-access-by-ip-address.mdx

@@ -3,10 +3,11 @@ title: Secure access by IP address
 description: Manage network access to Prefect Cloud accounts with IP Allowlists.
 ---
 
-Prefect Cloud's [Enterprise plan](https://www.prefect.io/pricing) offers IP allowlisting to restrict access to
-Prefect Cloud APIs and UI at the network level by giving users the ability to manage an explicit list of allowed IP addresses for an account.
+IP allowlisting is an available upgrade to certain Enterprise plans. 
+IP allowlisting enables account administrators to restrict access to Prefect Cloud APIs and the UI at the network level. 
+To learn more, please contact your account manager or the Prefect team at [email protected].
 
-To get started, use the Prefect CLI to add an IP address to the allowlist:
+Once the feature has been enabled for your team's Enterprise account, use the Prefect CLI to add an IP address to the allowlist:
 
 <Note>To help prevent accidental account lockouts, an update to an allowlist requires the requestor's current IP address to be on the list.</Note>
 
@@ -39,4 +40,4 @@ For other related commands, see the CLI help documentation with:
 
 ```bash
 prefect cloud ip-allowlist --help
-```
+```

src/integrations/prefect-azure/tests/test_credentials.py

@@ -1,13 +1,14 @@
 from unittest.mock import MagicMock
 
 import pytest
-from azure.storage.blob import BlobClient, BlobServiceClient, ContainerClient
+from azure.storage.blob.aio import BlobClient, BlobServiceClient, ContainerClient
 from conftest import CosmosClientMock
 from prefect_azure.credentials import (
     AzureBlobStorageCredentials,
     AzureCosmosDbCredentials,
     AzureMlCredentials,
 )
+from pydantic import SecretStr
 
 from prefect import flow
 
@@ -34,7 +35,7 @@ def test_flow():
 
     client = test_flow()
     assert isinstance(client, ContainerClient)
-    client.container_name == "container"
+    assert client.container_name == "container"
 
 
 def test_get_blob_client(blob_connection_string):
@@ -47,8 +48,8 @@ def test_flow():
 
     client = test_flow()
     assert isinstance(client, BlobClient)
-    client.container_name == "container"
-    client.blob_name == "blob"
+    assert client.container_name == "container"
+    assert client.blob_name == "blob"
 
 
 def test_get_service_client_either_error():
@@ -73,16 +74,16 @@ def test_get_blob_container_client_no_conn_str(account_url):
         "container"
     )
     assert isinstance(client, ContainerClient)
-    client.container_name == "container"
+    assert client.container_name == "container"
 
 
 def test_get_blob_client_no_conn_str(account_url):
     client = AzureBlobStorageCredentials(account_url=account_url).get_blob_client(
         "container", "blob"
     )
     assert isinstance(client, BlobClient)
-    client.container_name == "container"
-    client.blob_name == "blob"
+    assert client.container_name == "container"
+    assert client.blob_name == "blob"
 
 
 def test_get_cosmos_client(cosmos_connection_string):
@@ -126,7 +127,7 @@ def test_flow():
         workspace = AzureMlCredentials(
             tenant_id="tenant_id",
             service_principal_id="service_principal_id",
-            service_principal_password="service_principal_password",
+            service_principal_password=SecretStr("service_principal_password"),
             subscription_id="subscription_id",
             resource_group="resource_group",
             workspace_name="workspace_name",

src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py

@@ -181,6 +181,14 @@ async def get_client(
                 context=context,
                 client_configuration=client_configuration,
             )
+        else:
+            try:
+                config.load_incluster_config(client_configuration=client_configuration)
+            except config.ConfigException:
+                # If in-cluster config fails, load the local kubeconfig
+                config.load_kube_config(
+                    client_configuration=client_configuration,
+                )
         async with ApiClient(configuration=client_configuration) as api_client:
             try:
                 yield await self.get_resource_specific_client(

src/integrations/prefect-kubernetes/tests/test_credentials.py

@@ -2,6 +2,7 @@
 import tempfile
 from pathlib import Path
 from typing import Dict
+from unittest.mock import AsyncMock, MagicMock
 
 import pydantic
 import pytest
@@ -13,9 +14,13 @@
     CoreV1Api,
     CustomObjectsApi,
 )
+from kubernetes_asyncio.config import ConfigException
 from kubernetes_asyncio.config.kube_config import list_kube_config_contexts
 from OpenSSL import crypto
-from prefect_kubernetes.credentials import KubernetesClusterConfig
+from prefect_kubernetes.credentials import (
+    KubernetesClusterConfig,
+    KubernetesCredentials,
+)
 
 
 @pytest.fixture
@@ -107,6 +112,19 @@ def config_file(tmp_path, config_context) -> Path:
     return config_file
 
 
+@pytest.fixture
+def mock_cluster_config(monkeypatch):
+    mock = MagicMock()
+    # We cannot mock this or the `except` clause will complain
+    mock.ConfigException.return_value = ConfigException
+    mock.load_kube_config = AsyncMock()
+    monkeypatch.setattr("prefect_kubernetes.credentials.config", mock)
+    monkeypatch.setattr(
+        "prefect_kubernetes.credentials.config.ConfigException", ConfigException
+    )
+    return mock
+
+
 class TestCredentials:
     @pytest.mark.parametrize(
         "resource_type,client_type",
@@ -130,6 +148,21 @@ async def test_client_bad_resource_type(self, kubernetes_credentials):
             async with kubernetes_credentials.get_client("shoo-ba-daba-doo"):
                 pass
 
+    async def test_incluster_config(self, mock_cluster_config):
+        kubernetes_credentials = KubernetesCredentials()
+        mock_cluster_config.load_incluster_config.return_value = None
+        async with kubernetes_credentials.get_client("batch"):
+            assert mock_cluster_config.load_incluster_config.called
+            assert not mock_cluster_config.load_kube_config.called
+
+    async def test_load_kube_config(self, mock_cluster_config):
+        kubernetes_credentials = KubernetesCredentials()
+        mock_cluster_config.load_incluster_config.side_effect = ConfigException()
+        mock_cluster_config.load_kube_config.return_value = None
+        async with kubernetes_credentials.get_client("batch"):
+            assert mock_cluster_config.load_incluster_config.called
+            assert mock_cluster_config.load_kube_config.called
+
 
 class TestKubernetesClusterConfig:
     async def test_instantiation_from_file(self, config_file, config_context):

src/prefect/cli/_prompts.py

@@ -14,7 +14,6 @@
 from rich.progress import Progress, SpinnerColumn, TextColumn
 from rich.prompt import Confirm, InvalidResponse, Prompt, PromptBase
 from rich.table import Table
-from rich.text import Text
 
 from prefect.cli._utilities import exit_with_error
 from prefect.client.collections import get_collections_metadata_client
@@ -96,45 +95,40 @@ def prompt_select_from_table(
     current_idx = 0
     selected_row = None
     table_kwargs = table_kwargs or {}
+    visible_rows = min(10, console.height - 4)  # Adjust number of visible rows
+    scroll_offset = 0
+    total_options = len(data) + (1 if opt_out_message else 0)
 
     def build_table() -> Union[Table, Group]:
-        """
-        Generate a table of options. The `current_idx` will be highlighted.
-        """
-
+        nonlocal scroll_offset
         table = Table(**table_kwargs)
         table.add_column()
         for column in columns:
             table.add_column(column.get("header", ""))
 
-        rows = []
-        max_length = 250
-        for item in data:
-            rows.append(
-                tuple(
-                    (
-                        value[:max_length] + "...\n"
-                        if isinstance(value := item.get(column.get("key")), str)
-                        and len(value) > max_length
-                        else value
-                    )
-                    for column in columns
+        # Adjust scroll_offset if necessary
+        if current_idx < scroll_offset:
+            scroll_offset = current_idx
+        elif current_idx >= scroll_offset + visible_rows:
+            scroll_offset = current_idx - visible_rows + 1
+
+        for i, item in enumerate(data[scroll_offset : scroll_offset + visible_rows]):
+            row = [item.get(column.get("key", "")) for column in columns]
+            if i + scroll_offset == current_idx:
+                table.add_row(
+                    "[bold][blue]>", *[f"[bold][blue]{cell}[/]" for cell in row]
                 )
-            )
-
-        for i, row in enumerate(rows):
-            if i == current_idx:
-                # Use blue for selected options
-                table.add_row("[bold][blue]>", f"[bold][blue]{row[0]}[/]", *row[1:])
             else:
                 table.add_row("  ", *row)
 
         if opt_out_message:
-            prefix = "  > " if current_idx == len(data) else " " * 4
-            bottom_text = Text(prefix + opt_out_message)
+            opt_out_row = [""] * (len(columns) - 1) + [opt_out_message]
             if current_idx == len(data):
-                bottom_text.stylize("bold blue")
-            return Group(table, bottom_text)
+                table.add_row(
+                    "[bold][blue]>", *[f"[bold][blue]{cell}[/]" for cell in opt_out_row]
+                )
+            else:
+                table.add_row("  ", *opt_out_row)
 
         return table
 
@@ -151,24 +145,14 @@ def build_table() -> Union[Table, Group]:
             key = readchar.readkey()
 
             if key == readchar.key.UP:
-                current_idx = current_idx - 1
-                # wrap to bottom if at the top
-                if opt_out_message and current_idx < 0:
-                    current_idx = len(data)
-                elif not opt_out_message and current_idx < 0:
-                    current_idx = len(data) - 1
+                current_idx = (current_idx - 1) % total_options
             elif key == readchar.key.DOWN:
-                current_idx = current_idx + 1
-                # wrap to top if at the bottom
-                if opt_out_message and current_idx >= len(data) + 1:
-                    current_idx = 0
-                elif not opt_out_message and current_idx >= len(data):
-                    current_idx = 0
+                current_idx = (current_idx + 1) % total_options
             elif key == readchar.key.CTRL_C:
                 # gracefully exit with no message
                 exit_with_error("")
             elif key == readchar.key.ENTER or key == readchar.key.CR:
-                if current_idx >= len(data):
+                if current_idx == len(data):
                     return opt_out_response
                 else:
                     selected_row = data[current_idx]
@@ -181,8 +165,6 @@ def build_table() -> Union[Table, Group]:
 
 
 # Interval schedule prompting utilities
-
-
 class IntervalValuePrompt(PromptBase[timedelta]):
     response_type = timedelta
     validate_error_message = (
@@ -858,6 +840,7 @@ async def prompt_select_blob_storage_credentials(
     url = urls.url_for(new_block_document)
     if url:
         console.print(
-            "\nView/Edit your new credentials block in the UI:" f"\n[blue]{url}[/]\n"
+            "\nView/Edit your new credentials block in the UI:" f"\n[blue]{url}[/]\n",
+            soft_wrap=True,
         )
     return f"{{{{ prefect.blocks.{creds_block_type_slug}.{new_block_document.name} }}}}"

src/prefect/flows.py

@@ -1032,8 +1032,6 @@ def my_flow(name: str = "world"):
             if not isinstance(storage, LocalStorage):
                 storage.set_base_path(Path(tmpdir))
                 await storage.pull_code()
-            storage.set_base_path(Path(tmpdir))
-            await storage.pull_code()
 
             full_entrypoint = str(storage.destination / entrypoint)
             flow: Flow = await from_async.wait_for_call_in_new_thread(