-
- 
diff --git a/benches/bench_flows.py b/benches/bench_flows.py
index 9494fc44a9c9..83c196d183d2 100644
--- a/benches/bench_flows.py
+++ b/benches/bench_flows.py
@@ -2,12 +2,16 @@
TODO: Add benches for higher number of tasks; blocked by engine deadlocks in CI.
"""
+from typing import TYPE_CHECKING
+
import anyio
import pytest
-from pytest_benchmark.fixture import BenchmarkFixture
from prefect import flow, task
+if TYPE_CHECKING:
+ from pytest_benchmark.fixture import BenchmarkFixture
+
def noop_function():
pass
@@ -17,12 +21,12 @@ async def anoop_function():
pass
-def bench_flow_decorator(benchmark: BenchmarkFixture):
+def bench_flow_decorator(benchmark: "BenchmarkFixture"):
benchmark(flow, noop_function)
@pytest.mark.parametrize("options", [{}, {"timeout_seconds": 10}])
-def bench_flow_call(benchmark: BenchmarkFixture, options):
+def bench_flow_call(benchmark: "BenchmarkFixture", options):
noop_flow = flow(**options)(noop_function)
benchmark(noop_flow)
@@ -35,7 +39,7 @@ def bench_flow_call(benchmark: BenchmarkFixture, options):
@pytest.mark.parametrize("num_tasks", [10, 50, 100])
-def bench_flow_with_submitted_tasks(benchmark: BenchmarkFixture, num_tasks: int):
+def bench_flow_with_submitted_tasks(benchmark: "BenchmarkFixture", num_tasks: int):
test_task = task(noop_function)
@flow
@@ -47,7 +51,7 @@ def benchmark_flow():
@pytest.mark.parametrize("num_tasks", [10, 50, 100, 250])
-def bench_flow_with_called_tasks(benchmark: BenchmarkFixture, num_tasks: int):
+def bench_flow_with_called_tasks(benchmark: "BenchmarkFixture", num_tasks: int):
test_task = task(noop_function)
@flow
@@ -62,7 +66,7 @@ def benchmark_flow():
@pytest.mark.parametrize("num_tasks", [10, 50, 100, 250])
-def bench_async_flow_with_async_tasks(benchmark: BenchmarkFixture, num_tasks: int):
+def bench_async_flow_with_async_tasks(benchmark: "BenchmarkFixture", num_tasks: int):
test_task = task(anoop_function)
@flow
@@ -78,7 +82,7 @@ async def benchmark_flow():
@pytest.mark.parametrize("num_flows", [5, 10, 20])
-def bench_flow_with_subflows(benchmark: BenchmarkFixture, num_flows: int):
+def bench_flow_with_subflows(benchmark: "BenchmarkFixture", num_flows: int):
test_flow = flow(noop_function)
@flow
@@ -91,7 +95,7 @@ def benchmark_flow():
@pytest.mark.parametrize("num_flows", [5, 10, 20])
def bench_async_flow_with_sequential_subflows(
- benchmark: BenchmarkFixture, num_flows: int
+ benchmark: "BenchmarkFixture", num_flows: int
):
test_flow = flow(anoop_function)
@@ -105,7 +109,7 @@ async def benchmark_flow():
@pytest.mark.parametrize("num_flows", [5, 10, 20])
def bench_async_flow_with_concurrent_subflows(
- benchmark: BenchmarkFixture, num_flows: int
+ benchmark: "BenchmarkFixture", num_flows: int
):
test_flow = flow(anoop_function)
diff --git a/benches/bench_import.py b/benches/bench_import.py
index 0398ec523ce0..9d3142fc8729 100644
--- a/benches/bench_import.py
+++ b/benches/bench_import.py
@@ -1,9 +1,12 @@
import importlib
import sys
+from typing import TYPE_CHECKING
import pytest
from prometheus_client import REGISTRY
-from pytest_benchmark.fixture import BenchmarkFixture
+
+if TYPE_CHECKING:
+ from pytest_benchmark.fixture import BenchmarkFixture
def reset_imports():
@@ -21,7 +24,7 @@ def reset_imports():
@pytest.mark.benchmark(group="imports")
-def bench_import_prefect(benchmark: BenchmarkFixture):
+def bench_import_prefect(benchmark: "BenchmarkFixture"):
def import_prefect():
reset_imports()
@@ -32,7 +35,7 @@ def import_prefect():
@pytest.mark.timeout(180)
@pytest.mark.benchmark(group="imports")
-def bench_import_prefect_flow(benchmark: BenchmarkFixture):
+def bench_import_prefect_flow(benchmark: "BenchmarkFixture"):
def import_prefect_flow():
reset_imports()
diff --git a/benches/bench_tasks.py b/benches/bench_tasks.py
index e5317beeecda..88c151a6d584 100644
--- a/benches/bench_tasks.py
+++ b/benches/bench_tasks.py
@@ -1,4 +1,7 @@
-from pytest_benchmark.fixture import BenchmarkFixture
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from pytest_benchmark.fixture import BenchmarkFixture
from prefect import flow, task
@@ -7,11 +10,11 @@ def noop_function():
pass
-def bench_task_decorator(benchmark: BenchmarkFixture):
+def bench_task_decorator(benchmark: "BenchmarkFixture"):
benchmark(task, noop_function)
-def bench_task_call(benchmark: BenchmarkFixture):
+def bench_task_call(benchmark: "BenchmarkFixture"):
noop_task = task(noop_function)
@flow
@@ -21,7 +24,7 @@ def benchmark_flow():
benchmark_flow()
-def bench_task_submit(benchmark: BenchmarkFixture):
+def bench_task_submit(benchmark: "BenchmarkFixture"):
noop_task = task(noop_function)
# The benchmark occurs within the flow to measure _submission_ time without
diff --git a/benches/conftest.py b/benches/conftest.py
deleted file mode 100644
index a8f8313c0128..000000000000
--- a/benches/conftest.py
+++ /dev/null
@@ -1,22 +0,0 @@
-import traceback
-
-import pytest
-import pytest_benchmark.plugin
-
-_handle_saving = pytest_benchmark.session.BenchmarkSession.handle_saving
-
-
-@pytest.hookimpl(hookwrapper=True)
-def handle_saving(*args, **kwargs):
- """
- Patches pytest-benchmark's save handler to avoid raising exceptions on failure.
- An upstream bug causes failures to generate the benchmark JSON when tests fail.
- """
- try:
- return _handle_saving(*args, **kwargs)
- except Exception:
- print("Failed to save benchmark results:")
- traceback.print_exc()
-
-
-pytest_benchmark.session.BenchmarkSession.handle_saving = handle_saving
diff --git a/compat-tests b/compat-tests
index 3c5ec0111e2a..9b5fc44426b6 160000
--- a/compat-tests
+++ b/compat-tests
@@ -1 +1 @@
-Subproject commit 3c5ec0111e2aa7b160f2b21cfd383d19448dfe13
+Subproject commit 9b5fc44426b6a98a05408106fd6b5453ae9a0c76
diff --git a/docs/3.0/api-ref/index.mdx b/docs/3.0/api-ref/index.mdx
deleted file mode 100644
index 683883188a6e..000000000000
--- a/docs/3.0/api-ref/index.mdx
+++ /dev/null
@@ -1,12 +0,0 @@
----
-title: API & SDK References
-sidebarTitle: Overview
-description: Explore Prefect's auto-generated API & SDK reference documentation.
----
-
-Prefect auto-generates reference documentation for the following components:
-
-- **[Prefect Python SDK](https://prefect-python-sdk-docs.netlify.app/)**: used to build, test, and execute workflows.
-- **[Prefect REST API](/3.0/api-ref/rest-api)**: used by workflow clients and the Prefect UI for orchestration and data retrieval.
- - Prefect Cloud REST API documentation: https://app.prefect.cloud/api/docs.
- - Self-hosted Prefect server [REST API documentation](/3.0/api-ref/rest-api/server/). Additionally, if self-hosting a Prefect server instance, you can access REST API documentation at the `/docs` endpoint of your [`PREFECT_API_URL`](/3.0/manage/settings-and-profiles/). For example, if you run `prefect server start` with no additional configuration you can find this reference at http://localhost:4200/docs.
diff --git a/docs/3.0/automate/index.mdx b/docs/3.0/automate/index.mdx
deleted file mode 100644
index 3d5f0704d6c7..000000000000
--- a/docs/3.0/automate/index.mdx
+++ /dev/null
@@ -1,14 +0,0 @@
----
-title: Automate overview
-sidebarTitle: Overview
-description: Learn how to automate workflows with Prefect.
----
-
-This **Automate** section explains how create workflows that run automatically and respond to events.
-
-- [Schedule flow runs](/3.0/automate/add-schedules/) explains how to schedule flow runs for specific times or intervals.
-- [Track activity through events](/3.0/automate/events/events/) discusses how to observe events that create a record of activity.
-- [Trigger actions on events](/3.0/automate/events/automations-triggers/) shows how to use events to trigger actions.
-- [Define custom event triggers](/3.0/automate/events/custom-triggers/) discusses advanced trigger options.
-- [Receive events with webhooks](/3.0/automate/events/webhook-triggers/) explains how to use webhooks to receive events from external systems with Prefect Cloud.
-- [Manage incidents](/3.0/automate/incidents/) show how Prefect Cloud can help identify, resolve, and document issues in mission-critical workflows.
diff --git a/docs/3.0/develop/index.mdx b/docs/3.0/develop/index.mdx
deleted file mode 100644
index 12717ded5db4..000000000000
--- a/docs/3.0/develop/index.mdx
+++ /dev/null
@@ -1,38 +0,0 @@
----
-title: Develop overview
-sidebarTitle: Overview
-description: Learn how to write code with Prefect objects for workflows you can trust.
----
-
-These pages explain how to develop Prefect workflows.
-
-The **Write and run workflows** section introduces and explains Prefect's core concepts.
-
-- [Write and run flows](/3.0/develop/write-flows/) introduces the most central Prefect object, the `flow`.
-- [Write and run tasks](/3.0/develop/write-tasks/) shows you how to create `task` decorated functions that represent a discrete unit of work in a Prefect workflow.
-- [Run tasks in the background](/3.0/develop/task-runners/) explains how to use tasks to execute small, discrete units of work quickly.
-- [Log run activity](/3.0/develop/logging/) shows you how to capture fine-grained information about flows and tasks for monitoring, troubleshooting, and auditing.
-
-The **Configure runtime behavior** section introduces advanced features for interacting with running workflows and automatically taking actions based on workflow state.
-
-- [Configure task caching](/3.0/develop/task-caching/) explores how to cache task results to save money and time.
-- [Write transactions](/3.0/develop/transactions/) shows how to group tasks together to ensure they run atomically.
-- [Pause and resume flow runs](/3.0/develop/pause-resume/) explains how to halt flow runs for input.
-- [Send and receive flow run inputs](/3.0/develop/inputs/) highlights how to run Human-in-the-Loop workflows with Prefect.
-- [Access runtime context](/3.0/develop/runtime-context/) shows how to access information about the current run.
-
-The **Manage state and configuration** section explores how to manage data and configuration in Prefect.
-
-- [Manage results](/3.0/develop/results/) explains how to store and retrieve task results.
-- [Manage states](/3.0/develop/manage-states/) discusses how to take action based on task and flow run states.
-- [Create run artifacts](/3.0/develop/artifacts/) demonstrates how to create artifacts for human consumption in the UI.
-- [Set and get variables](/3.0/develop/variables/) shows how to store and retrieve configuration values from API.
-- [Connect to external systems](/3.0/develop/blocks/) discusses how Prefect blocks can help you connect to external systems.
-
-The **Manage concurrency** section explains how to speed up your workflows and limit activities that can take place concurrently.
-
-- [Run tasks concurrently or in parallel](/3.0/develop/task-runners/) explains how to run tasks concurrently or in parallel with Dask or Ray.
-- [Limit concurrent task runs](/3.0/develop/task-run-limits/) shows how to prevent too many tasks from running simultaneously.
-- [Apply concurrency and rate limits](/3.0/develop/global-concurrency-limits/) demonstrates how to control concurrency and apply rate limits using Prefect's provided utilities.
-
-Finally, [Test workflows](/3.0/develop/test-workflows/) discusses tools for testing workflows.
diff --git a/docs/3.0/get-started/quickstart.mdx b/docs/3.0/get-started/quickstart.mdx
deleted file mode 100644
index a248585231eb..000000000000
--- a/docs/3.0/get-started/quickstart.mdx
+++ /dev/null
@@ -1,243 +0,0 @@
----
-title: Quickstart
-description: Get started with Prefect, the easiest way to orchestrate and observe your data pipelines
----
-
-import Installation from '/snippets/installation.mdx'
-
-Prefect is an orchestration and observability platform that empowers developers to build and scale workflows quickly.
-In this quickstart, you will use Prefect to convert the following Python script to a schedulable, observable, resilient, and deployable workflow in minutes:
-
-```python
-import httpx
-
-def get_repo_info():
- """Fetch statistics about the Prefect repository"""
- url = "https://api.github.com/repos/PrefectHQ/prefect"
- response = httpx.get(url)
- repo = response.json()
- print("PrefectHQ/prefect repository statistics 🤓:")
- print(f"Stars 🌠 : {repo['stargazers_count']}")
-
-if __name__ == "__main__":
- get_repo_info()
-```
-
-## Install Prefect
-
-
-## Welcome!
+## Welcome
`prefect-aws` makes it easy to leverage the capabilities of AWS in your flows, featuring support for ECS, S3, Secrets Manager, and Batch.
@@ -18,7 +18,3 @@ To start using `prefect-aws`:
```bash
pip install prefect-aws
```
-
-### Contributing
-
-Thanks for thinking about chipping in! Check out this [step-by-step guide](https://prefecthq.github.io/prefect-aws/#installation) on how to get started.
diff --git a/src/integrations/prefect-aws/prefect_aws/batch.py b/src/integrations/prefect-aws/prefect_aws/batch.py
index c9ef8dca2a9c..17ac00146cae 100644
--- a/src/integrations/prefect-aws/prefect_aws/batch.py
+++ b/src/integrations/prefect-aws/prefect_aws/batch.py
@@ -3,14 +3,14 @@
from typing import Any, Dict, Optional
from prefect import task
+from prefect._internal.compatibility.async_dispatch import async_dispatch
from prefect.logging import get_run_logger
-from prefect.utilities.asyncutils import run_sync_in_worker_thread, sync_compatible
+from prefect.utilities.asyncutils import run_sync_in_worker_thread
from prefect_aws.credentials import AwsCredentials
@task
-@sync_compatible
-async def batch_submit(
+async def abatch_submit(
job_name: str,
job_queue: str,
job_definition: str,
@@ -18,7 +18,7 @@ async def batch_submit(
**batch_kwargs: Optional[Dict[str, Any]],
) -> str:
"""
- Submit a job to the AWS Batch job service.
+ Asynchronously submit a job to the AWS Batch job service.
Args:
job_name: The AWS batch job name.
@@ -73,3 +73,69 @@ def example_batch_submit_flow():
**batch_kwargs,
)
return response["jobId"]
+
+
+@task
+@async_dispatch(abatch_submit)
+def batch_submit(
+ job_name: str,
+ job_queue: str,
+ job_definition: str,
+ aws_credentials: AwsCredentials,
+ **batch_kwargs: Optional[Dict[str, Any]],
+) -> str:
+ """
+ Submit a job to the AWS Batch job service.
+
+ Args:
+ job_name: The AWS batch job name.
+ job_queue: Name of the AWS batch job queue.
+ job_definition: The AWS batch job definition.
+ aws_credentials: Credentials to use for authentication with AWS.
+ **batch_kwargs: Additional keyword arguments to pass to the boto3
+ `submit_job` function. See the documentation for
+ [submit_job](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/batch.html#Batch.Client.submit_job)
+ for more details.
+
+ Returns:
+ The id corresponding to the job.
+
+ Example:
+ Submits a job to batch.
+
+ ```python
+ from prefect import flow
+ from prefect_aws import AwsCredentials
+ from prefect_aws.batch import batch_submit
+
+
+ @flow
+ def example_batch_submit_flow():
+ aws_credentials = AwsCredentials(
+ aws_access_key_id="acccess_key_id",
+ aws_secret_access_key="secret_access_key"
+ )
+ job_id = batch_submit(
+ "job_name",
+ "job_queue",
+ "job_definition",
+ aws_credentials
+ )
+ return job_id
+
+ example_batch_submit_flow()
+ ```
+
+ """ # noqa
+ logger = get_run_logger()
+ logger.info("Preparing to submit %s job to %s job queue", job_name, job_queue)
+
+ batch_client = aws_credentials.get_boto3_session().client("batch")
+
+ response = batch_client.submit_job(
+ jobName=job_name,
+ jobQueue=job_queue,
+ jobDefinition=job_definition,
+ **batch_kwargs,
+ )
+ return response["jobId"]
diff --git a/src/integrations/prefect-aws/prefect_aws/credentials.py b/src/integrations/prefect-aws/prefect_aws/credentials.py
index 987a4c722660..7a8a4add88b3 100644
--- a/src/integrations/prefect-aws/prefect_aws/credentials.py
+++ b/src/integrations/prefect-aws/prefect_aws/credentials.py
@@ -70,7 +70,7 @@ class AwsCredentials(CredentialsBlock):
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png" # noqa
_block_type_name = "AWS Credentials"
- _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.AwsCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-aws" # noqa
aws_access_key_id: Optional[str] = Field(
default=None,
@@ -209,7 +209,7 @@ class MinIOCredentials(CredentialsBlock):
"docs: https://docs.min.io/docs/minio-server-configuration-guide.html "
"for more info about the possible credential configurations."
)
- _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.MinIOCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-aws" # noqa
minio_root_user: str = Field(default=..., description="Admin or root user.")
minio_root_password: SecretStr = Field(
diff --git a/src/integrations/prefect-aws/prefect_aws/lambda_function.py b/src/integrations/prefect-aws/prefect_aws/lambda_function.py
index 8cbd14de1bca..6b778c48f407 100644
--- a/src/integrations/prefect-aws/prefect_aws/lambda_function.py
+++ b/src/integrations/prefect-aws/prefect_aws/lambda_function.py
@@ -79,7 +79,7 @@ class LambdaFunction(Block):
_block_type_name = "Lambda Function"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-aws/s3/#prefect_aws.lambda_function.LambdaFunction" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-aws" # noqa
function_name: str = Field(
title="Function Name",
diff --git a/src/integrations/prefect-aws/prefect_aws/s3.py b/src/integrations/prefect-aws/prefect_aws/s3.py
index 84570728c3d2..39a3e9b7ea9c 100644
--- a/src/integrations/prefect-aws/prefect_aws/s3.py
+++ b/src/integrations/prefect-aws/prefect_aws/s3.py
@@ -403,9 +403,7 @@ class S3Bucket(WritableFileSystem, WritableDeploymentStorage, ObjectStorageBlock
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png" # noqa
_block_type_name = "S3 Bucket"
- _documentation_url = (
- "https://prefecthq.github.io/prefect-aws/s3/#prefect_aws.s3.S3Bucket" # noqa
- )
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-aws" # noqa
bucket_name: str = Field(default=..., description="Name of your bucket.")
diff --git a/src/integrations/prefect-aws/prefect_aws/secrets_manager.py b/src/integrations/prefect-aws/prefect_aws/secrets_manager.py
index 0536f8175b9e..82a695d02141 100644
--- a/src/integrations/prefect-aws/prefect_aws/secrets_manager.py
+++ b/src/integrations/prefect-aws/prefect_aws/secrets_manager.py
@@ -367,7 +367,7 @@ class AwsSecret(SecretBlock):
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png" # noqa
_block_type_name = "AWS Secret"
- _documentation_url = "https://prefecthq.github.io/prefect-aws/secrets_manager/#prefect_aws.secrets_manager.AwsSecret" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-aws" # noqa
aws_credentials: AwsCredentials
secret_name: str = Field(default=..., description="The name of the secret.")
diff --git a/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py b/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py
index 4d6da6127392..54cf62958350 100644
--- a/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py
+++ b/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py
@@ -654,7 +654,7 @@ class ECSWorker(BaseWorker):
"and Fargate clusters. Requires an AWS account."
)
_display_name = "AWS Elastic Container Service"
- _documentation_url = "https://prefecthq.github.io/prefect-aws/ecs_worker/"
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-aws/"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png" # noqa
async def run(
diff --git a/src/integrations/prefect-aws/pyproject.toml b/src/integrations/prefect-aws/pyproject.toml
index 8f6be1fb19fb..bf24bb96be85 100644
--- a/src/integrations/prefect-aws/pyproject.toml
+++ b/src/integrations/prefect-aws/pyproject.toml
@@ -27,7 +27,7 @@ dependencies = [
"botocore>=1.27.53",
"mypy_boto3_s3>=1.24.94",
"mypy_boto3_secretsmanager>=1.26.49",
- "prefect>=3.0.0rc1",
+ "prefect>=3.1.3",
"pyparsing>=3.1.1",
"tenacity>=8.0.0",
]
@@ -50,7 +50,7 @@ dev = [
"mypy",
"pillow",
"pre-commit",
- "pytest",
+ "pytest >= 8.3",
"pytest-asyncio >= 0.18.2, != 0.22.0, < 0.23.0", # Cannot override event loop in 0.23.0. See https://github.com/pytest-dev/pytest-asyncio/issues/706 for more details.
"pytest-cov",
"pytest-env",
@@ -86,7 +86,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-aws/tests/test_batch.py b/src/integrations/prefect-aws/tests/test_batch.py
index 404aa38dee57..8f8acb47dfbe 100644
--- a/src/integrations/prefect-aws/tests/test_batch.py
+++ b/src/integrations/prefect-aws/tests/test_batch.py
@@ -1,13 +1,23 @@
+from typing import Optional
from uuid import UUID
import boto3
import pytest
from moto import mock_batch, mock_iam
-from prefect_aws.batch import batch_submit
+from prefect_aws.batch import abatch_submit, batch_submit
from prefect import flow
+def assert_valid_job_id(job_id: Optional[str]):
+ assert job_id is not None, "job_id is None"
+ try:
+ UUID(str(job_id))
+ assert True, f"{job_id} is a valid UUID"
+ except ValueError:
+ assert False, f"{job_id} is not a valid UUID"
+
+
@pytest.fixture(scope="function")
def batch_client(aws_credentials):
with mock_batch():
@@ -62,20 +72,57 @@ def job_definition_arn(batch_client):
return job_definition_arn
-def test_batch_submit(job_queue_arn, job_definition_arn, aws_credentials):
- @flow
- def test_flow():
- return batch_submit(
+class TestBatchSubmit:
+ def test_batch_submit(self, job_queue_arn, job_definition_arn, aws_credentials):
+ @flow
+ def test_flow():
+ return batch_submit(
+ "batch_test_job",
+ job_queue_arn,
+ job_definition_arn,
+ aws_credentials,
+ )
+
+ job_id = test_flow()
+
+ assert_valid_job_id(job_id)
+
+ async def test_batch_submit_async_dispatch(
+ self, job_queue_arn, job_definition_arn, aws_credentials
+ ):
+ @flow
+ async def test_flow():
+ return await batch_submit(
+ "batch_test_job",
+ job_queue_arn,
+ job_definition_arn,
+ aws_credentials,
+ )
+
+ job_id = await test_flow()
+ assert_valid_job_id(job_id)
+
+ async def test_batch_submit_force_sync_from_async(
+ self, job_queue_arn, job_definition_arn, aws_credentials
+ ):
+ job_id = batch_submit(
"batch_test_job",
job_queue_arn,
job_definition_arn,
aws_credentials,
+ _sync=True,
)
+ assert_valid_job_id(job_id)
- job_id = test_flow()
- try:
- UUID(str(job_id))
- assert True, f"{job_id} is a valid UUID"
- except ValueError:
- assert False, f"{job_id} is not a valid UUID"
+class TestBatchSubmitAsync:
+ async def test_batch_submit_explicit_async(
+ self, job_queue_arn, job_definition_arn, aws_credentials
+ ):
+ job_id = await abatch_submit(
+ "batch_test_job",
+ job_queue_arn,
+ job_definition_arn,
+ aws_credentials,
+ )
+ assert_valid_job_id(job_id)
diff --git a/src/integrations/prefect-azure/prefect_azure/blob_storage.py b/src/integrations/prefect-azure/prefect_azure/blob_storage.py
index 7a6edb5af999..6f3c348762d1 100644
--- a/src/integrations/prefect-azure/prefect_azure/blob_storage.py
+++ b/src/integrations/prefect-azure/prefect_azure/blob_storage.py
@@ -63,8 +63,9 @@ def example_blob_storage_download_flow():
logger = get_run_logger()
logger.info("Downloading blob from container %s with key %s", container, blob)
- async with blob_storage_credentials.get_blob_client(container, blob) as blob_client:
- blob_obj = await blob_client.download_blob()
+ async with blob_storage_credentials as credentials:
+ async with credentials.get_blob_client(container, blob) as blob_client:
+ blob_obj = await blob_client.download_blob()
output = await blob_obj.content_as_bytes()
return output
@@ -123,8 +124,9 @@ def example_blob_storage_upload_flow():
if blob is None:
blob = str(uuid.uuid4())
- async with blob_storage_credentials.get_blob_client(container, blob) as blob_client:
- await blob_client.upload_blob(data, overwrite=overwrite)
+ async with blob_storage_credentials as credentials:
+ async with credentials.get_blob_client(container, blob) as blob_client:
+ await blob_client.upload_blob(data, overwrite=overwrite)
return blob
@@ -176,17 +178,16 @@ def example_blob_storage_list_flow():
logger = get_run_logger()
logger.info("Listing blobs from container %s", container)
- async with blob_storage_credentials.get_container_client(
- container
- ) as container_client:
- blobs = [
- blob
- async for blob in container_client.list_blobs(
- name_starts_with=name_starts_with, include=include, **kwargs
- )
- ]
+ async with blob_storage_credentials as credentials:
+ async with credentials.get_container_client(container) as container_client:
+ blobs = [
+ blob
+ async for blob in container_client.list_blobs(
+ name_starts_with=name_starts_with, include=include, **kwargs
+ )
+ ]
- return blobs
+ return blobs
class AzureBlobStorageContainer(
@@ -207,7 +208,7 @@ class AzureBlobStorageContainer(
_block_type_name = "Azure Blob Storage Container"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-azure/blob_storage/#prefect_azure.blob_storabe.AzureBlobStorageContainer" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-azure" # noqa
container_name: str = Field(
default=..., description="The name of a Azure Blob Storage container."
@@ -278,32 +279,35 @@ async def download_folder_to_path(
to_folder,
)
full_container_path = self._get_path_relative_to_base_folder(from_folder)
- async with self.credentials.get_container_client(
- self.container_name
- ) as container_client:
- try:
- async for blob in container_client.list_blobs(
- name_starts_with=full_container_path
- ):
- blob_path = blob.name
- local_path = Path(to_folder) / Path(blob_path).relative_to(
- full_container_path
- )
- local_path.parent.mkdir(parents=True, exist_ok=True)
- async with container_client.get_blob_client(
- blob_path
- ) as blob_client:
- blob_obj = await blob_client.download_blob(**download_kwargs)
-
- with local_path.open(mode="wb") as to_file:
- await blob_obj.readinto(to_file)
- except ResourceNotFoundError as exc:
- raise RuntimeError(
- "An error occurred when attempting to download from container"
- f" {self.container_name}: {exc.reason}"
- ) from exc
+ async with self.credentials as credentials:
+ async with credentials.get_container_client(
+ self.container_name
+ ) as container_client:
+ try:
+ async for blob in container_client.list_blobs(
+ name_starts_with=full_container_path
+ ):
+ blob_path = blob.name
+ local_path = Path(to_folder) / Path(blob_path).relative_to(
+ full_container_path
+ )
+ local_path.parent.mkdir(parents=True, exist_ok=True)
+ async with container_client.get_blob_client(
+ blob_path
+ ) as blob_client:
+ blob_obj = await blob_client.download_blob(
+ **download_kwargs
+ )
+
+ with local_path.open(mode="wb") as to_file:
+ await blob_obj.readinto(to_file)
+ except ResourceNotFoundError as exc:
+ raise RuntimeError(
+ "An error occurred when attempting to download from container"
+ f" {self.container_name}: {exc.reason}"
+ ) from exc
- return Path(to_folder)
+ return Path(to_folder)
@sync_compatible
async def download_object_to_file_object(
@@ -349,19 +353,20 @@ async def download_object_to_file_object(
"Downloading object from container %s to file object", self.container_name
)
full_container_path = self._get_path_relative_to_base_folder(from_path)
- async with self.credentials.get_blob_client(
- self.container_name, full_container_path
- ) as blob_client:
- try:
- blob_obj = await blob_client.download_blob(**download_kwargs)
- await blob_obj.download_to_stream(to_file_object)
- except ResourceNotFoundError as exc:
- raise RuntimeError(
- "An error occurred when attempting to download from container"
- f" {self.container_name}: {exc.reason}"
- ) from exc
-
- return to_file_object
+ async with self.credentials as credentials:
+ async with credentials.get_blob_client(
+ self.container_name, full_container_path
+ ) as blob_client:
+ try:
+ blob_obj = await blob_client.download_blob(**download_kwargs)
+ await blob_obj.download_to_stream(to_file_object)
+ except ResourceNotFoundError as exc:
+ raise RuntimeError(
+ "An error occurred when attempting to download from container"
+ f" {self.container_name}: {exc.reason}"
+ ) from exc
+
+ return to_file_object
@sync_compatible
async def download_object_to_path(
@@ -409,24 +414,25 @@ async def download_object_to_path(
to_path,
)
full_container_path = self._get_path_relative_to_base_folder(from_path)
- async with self.credentials.get_blob_client(
- self.container_name, full_container_path
- ) as blob_client:
- try:
- blob_obj = await blob_client.download_blob(**download_kwargs)
+ async with self.credentials as credentials:
+ async with credentials.get_blob_client(
+ self.container_name, full_container_path
+ ) as blob_client:
+ try:
+ blob_obj = await blob_client.download_blob(**download_kwargs)
- path = Path(to_path)
+ path = Path(to_path)
- path.parent.mkdir(parents=True, exist_ok=True)
+ path.parent.mkdir(parents=True, exist_ok=True)
- with path.open(mode="wb") as to_file:
- await blob_obj.readinto(to_file)
- except ResourceNotFoundError as exc:
- raise RuntimeError(
- "An error occurred when attempting to download from container"
- f" {self.container_name}: {exc.reason}"
- ) from exc
- return Path(to_path)
+ with path.open(mode="wb") as to_file:
+ await blob_obj.readinto(to_file)
+ except ResourceNotFoundError as exc:
+ raise RuntimeError(
+ "An error occurred when attempting to download from container"
+ f" {self.container_name}: {exc.reason}"
+ ) from exc
+ return path
@sync_compatible
async def upload_from_file_object(
@@ -471,18 +477,19 @@ async def upload_from_file_object(
"Uploading object to container %s with key %s", self.container_name, to_path
)
full_container_path = self._get_path_relative_to_base_folder(to_path)
- async with self.credentials.get_blob_client(
- self.container_name, full_container_path
- ) as blob_client:
- try:
- await blob_client.upload_blob(from_file_object, **upload_kwargs)
- except ResourceNotFoundError as exc:
- raise RuntimeError(
- "An error occurred when attempting to upload from container"
- f" {self.container_name}: {exc.reason}"
- ) from exc
-
- return to_path
+ async with self.credentials as credentials:
+ async with credentials.get_blob_client(
+ self.container_name, full_container_path
+ ) as blob_client:
+ try:
+ await blob_client.upload_blob(from_file_object, **upload_kwargs)
+ except ResourceNotFoundError as exc:
+ raise RuntimeError(
+ "An error occurred when attempting to upload from container"
+ f" {self.container_name}: {exc.reason}"
+ ) from exc
+
+ return to_path
@sync_compatible
async def upload_from_path(
@@ -526,19 +533,20 @@ async def upload_from_path(
"Uploading object to container %s with key %s", self.container_name, to_path
)
full_container_path = self._get_path_relative_to_base_folder(to_path)
- async with self.credentials.get_blob_client(
- self.container_name, full_container_path
- ) as blob_client:
- try:
- with open(from_path, "rb") as f:
- await blob_client.upload_blob(f, **upload_kwargs)
- except ResourceNotFoundError as exc:
- raise RuntimeError(
- "An error occurred when attempting to upload to container"
- f" {self.container_name}: {exc.reason}"
- ) from exc
-
- return to_path
+ async with self.credentials as credentials:
+ async with credentials.get_blob_client(
+ self.container_name, full_container_path
+ ) as blob_client:
+ try:
+ with open(from_path, "rb") as f:
+ await blob_client.upload_blob(f, **upload_kwargs)
+ except ResourceNotFoundError as exc:
+ raise RuntimeError(
+ "An error occurred when attempting to upload to container"
+ f" {self.container_name}: {exc.reason}"
+ ) from exc
+
+ return to_path
@sync_compatible
async def upload_from_folder(
@@ -587,29 +595,30 @@ async def upload_from_folder(
to_folder,
)
full_container_path = self._get_path_relative_to_base_folder(to_folder)
- async with self.credentials.get_container_client(
- self.container_name
- ) as container_client:
- if not Path(from_folder).is_dir():
- raise ValueError(f"{from_folder} is not a directory")
- for path in Path(from_folder).rglob("*"):
- if path.is_file():
- blob_path = Path(full_container_path) / path.relative_to(
- from_folder
- )
- async with container_client.get_blob_client(
- blob_path.as_posix()
- ) as blob_client:
- try:
- await blob_client.upload_blob(
- path.read_bytes(), **upload_kwargs
- )
- except ResourceNotFoundError as exc:
- raise RuntimeError(
- "An error occurred when attempting to upload to "
- f"container {self.container_name}: {exc.reason}"
- ) from exc
- return full_container_path
+ async with self.credentials as credentials:
+ async with credentials.get_container_client(
+ self.container_name
+ ) as container_client:
+ if not Path(from_folder).is_dir():
+ raise ValueError(f"{from_folder} is not a directory")
+ for path in Path(from_folder).rglob("*"):
+ if path.is_file():
+ blob_path = Path(full_container_path) / path.relative_to(
+ from_folder
+ )
+ async with container_client.get_blob_client(
+ blob_path.as_posix()
+ ) as blob_client:
+ try:
+ await blob_client.upload_blob(
+ path.read_bytes(), **upload_kwargs
+ )
+ except ResourceNotFoundError as exc:
+ raise RuntimeError(
+ "An error occurred when attempting to upload to "
+ f"container {self.container_name}: {exc.reason}"
+ ) from exc
+ return full_container_path
@sync_compatible
async def get_directory(
@@ -736,11 +745,12 @@ async def list_blobs(self) -> List[str]:
self.container_name,
)
- async with self.credentials.get_container_client(
- self.container_name
- ) as container_client:
- blobs = container_client.list_blobs()
- filenames = []
- async for blob in blobs:
- filenames.append(blob.name)
- return filenames
+ async with self.credentials as credentials:
+ async with credentials.get_container_client(
+ self.container_name
+ ) as container_client:
+ blobs = container_client.list_blobs()
+ filenames = []
+ async for blob in blobs:
+ filenames.append(blob.name)
+ return filenames
diff --git a/src/integrations/prefect-azure/prefect_azure/credentials.py b/src/integrations/prefect-azure/prefect_azure/credentials.py
index 2a3f776bc26d..3f0ffefbfa24 100644
--- a/src/integrations/prefect-azure/prefect_azure/credentials.py
+++ b/src/integrations/prefect-azure/prefect_azure/credentials.py
@@ -7,7 +7,7 @@
from azure.identity.aio import DefaultAzureCredential as ADefaultAzureCredential
from azure.mgmt.containerinstance import ContainerInstanceManagementClient
from azure.mgmt.resource import ResourceManagementClient
-from pydantic import Field, SecretStr, model_validator
+from pydantic import Field, PrivateAttr, SecretStr, model_validator
try:
from azure.cosmos import CosmosClient
@@ -95,7 +95,8 @@ class AzureBlobStorageCredentials(Block):
_block_type_name = "Azure Blob Storage Credentials"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureBlobStorageCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-azure" # noqa
+ _credential: Optional[ADefaultAzureCredential] = PrivateAttr(default=None)
connection_string: Optional[SecretStr] = Field(
default=None,
@@ -162,9 +163,10 @@ async def example_get_client_flow():
```
"""
if self.connection_string is None:
+ self._credential = self._credential or ADefaultAzureCredential()
return BlobServiceClient(
account_url=self.account_url,
- credential=ADefaultAzureCredential(),
+ credential=self._credential,
)
return BlobServiceClient.from_connection_string(
@@ -205,10 +207,11 @@ async def example_get_blob_client_flow():
```
"""
if self.connection_string is None:
+ self._credential = self._credential or ADefaultAzureCredential()
return BlobClient(
account_url=self.account_url,
container_name=container,
- credential=ADefaultAzureCredential(),
+ credential=self._credential,
blob_name=blob,
)
@@ -250,10 +253,11 @@ async def example_get_container_client_flow():
```
"""
if self.connection_string is None:
+ self._credential = self._credential or ADefaultAzureCredential()
return ContainerClient(
account_url=self.account_url,
container_name=container,
- credential=ADefaultAzureCredential(),
+ credential=self._credential,
)
container_client = ContainerClient.from_connection_string(
@@ -261,6 +265,17 @@ async def example_get_container_client_flow():
)
return container_client
+ async def aclose(self):
+ """Cleanup resources."""
+ if self._credential:
+ await self._credential.close()
+
+ async def __aenter__(self):
+ return self
+
+ async def __aexit__(self, exc_type, exc_val, exc_tb):
+ await self.aclose()
+
class AzureCosmosDbCredentials(Block):
"""
@@ -281,7 +296,7 @@ class AzureCosmosDbCredentials(Block):
_block_type_name = "Azure Cosmos DB Credentials"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureCosmosDbCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-azure" # noqa
connection_string: SecretStr = Field(
default=..., description="Includes the authorization information required."
@@ -401,7 +416,7 @@ class AzureMlCredentials(Block):
_block_type_name = "AzureML Credentials"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureMlCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-azure" # noqa
tenant_id: str = Field(
default=...,
@@ -474,7 +489,7 @@ class AzureContainerInstanceCredentials(Block):
_block_type_name = "Azure Container Instance Credentials"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/54e3fa7e00197a4fbd1d82ed62494cb58d08c96a-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-azure/credentials/#prefect_azure.credentials.AzureContainerInstanceCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-azure" # noqa
client_id: Optional[str] = Field(
default=None,
diff --git a/src/integrations/prefect-azure/prefect_azure/workers/container_instance.py b/src/integrations/prefect-azure/prefect_azure/workers/container_instance.py
index 81d1b057a49d..3b293db87fe7 100644
--- a/src/integrations/prefect-azure/prefect_azure/workers/container_instance.py
+++ b/src/integrations/prefect-azure/prefect_azure/workers/container_instance.py
@@ -526,9 +526,7 @@ class AzureContainerWorker(BaseWorker):
"Execute flow runs within containers on Azure's Container Instances "
"service. Requires an Azure account."
)
- _documentation_url = (
- "https://prefecthq.github.io/prefect-azure/container_instance_worker/"
- )
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-azure"
async def run(
self,
diff --git a/src/integrations/prefect-azure/pyproject.toml b/src/integrations/prefect-azure/pyproject.toml
index 864b796698e3..60bb213454e3 100644
--- a/src/integrations/prefect-azure/pyproject.toml
+++ b/src/integrations/prefect-azure/pyproject.toml
@@ -27,8 +27,8 @@ dependencies = [
"azure_identity>=1.10",
"azure_mgmt_containerinstance>=10.0",
"azure-mgmt-resource>=21.2",
- "prefect>=3.0.0rc1",
- "setuptools", #required in 3.12 to get pkg_resources (used by azureml.core)
+ "prefect>=3.0.0",
+ "setuptools", #required in 3.12 to get pkg_resources (used by azureml.core)
]
dynamic = ["version"]
@@ -52,7 +52,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -85,7 +85,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-azure/tests/conftest.py b/src/integrations/prefect-azure/tests/conftest.py
index 21f8fbf8eea9..8efe3bb9b8c1 100644
--- a/src/integrations/prefect-azure/tests/conftest.py
+++ b/src/integrations/prefect-azure/tests/conftest.py
@@ -90,6 +90,9 @@ async def close(self):
@pytest.fixture
def blob_storage_credentials():
blob_storage_credentials = MagicMock()
+ blob_storage_credentials.__aenter__ = AsyncMock(
+ return_value=blob_storage_credentials
+ )
blob_storage_credentials.get_client.side_effect = (
lambda: BlobStorageClientMethodsMock()
)
diff --git a/src/integrations/prefect-bitbucket/pyproject.toml b/src/integrations/prefect-bitbucket/pyproject.toml
index 20ab9b222773..8498167ec1b0 100644
--- a/src/integrations/prefect-bitbucket/pyproject.toml
+++ b/src/integrations/prefect-bitbucket/pyproject.toml
@@ -23,7 +23,7 @@ classifiers = [
"Topic :: Software Development :: Libraries",
]
dependencies = [
- "prefect>=3.0.0rc1",
+ "prefect>=3.0.0",
"pydantic>=2.4",
"atlassian-python-api>=3.32.1,!=3.41.5,!=3.41.6,!=3.41.7,!=3.41.8",
]
@@ -42,7 +42,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -75,7 +75,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-dask/prefect_dask/task_runners.py b/src/integrations/prefect-dask/prefect_dask/task_runners.py
index f4edbab25295..60066faaf870 100644
--- a/src/integrations/prefect-dask/prefect_dask/task_runners.py
+++ b/src/integrations/prefect-dask/prefect_dask/task_runners.py
@@ -319,7 +319,7 @@ def submit(
self,
task: "Task[P, Coroutine[Any, Any, R]]",
parameters: Dict[str, Any],
- wait_for: Optional[Iterable[PrefectFuture]] = None,
+ wait_for: Optional[Iterable[PrefectDaskFuture[R]]] = None,
dependencies: Optional[Dict[str, Set[TaskRunInput]]] = None,
) -> PrefectDaskFuture[R]:
...
@@ -329,26 +329,26 @@ def submit(
self,
task: "Task[Any, R]",
parameters: Dict[str, Any],
- wait_for: Optional[Iterable[PrefectFuture]] = None,
+ wait_for: Optional[Iterable[PrefectDaskFuture[R]]] = None,
dependencies: Optional[Dict[str, Set[TaskRunInput]]] = None,
) -> PrefectDaskFuture[R]:
...
def submit(
self,
- task: Task,
+ task: "Union[Task[P, R], Task[P, Coroutine[Any, Any, R]]]",
parameters: Dict[str, Any],
- wait_for: Optional[Iterable[PrefectFuture]] = None,
+ wait_for: Optional[Iterable[PrefectDaskFuture[R]]] = None,
dependencies: Optional[Dict[str, Set[TaskRunInput]]] = None,
- ) -> PrefectDaskFuture:
+ ) -> PrefectDaskFuture[R]:
if not self._started:
raise RuntimeError(
"The task runner must be started before submitting work."
)
- # unpack the upstream call in order to cast Prefect futures to Dask futures
- # where possible to optimize Dask task scheduling
+ # Convert both parameters and wait_for futures to Dask futures
parameters = self._optimize_futures(parameters)
+ wait_for = self._optimize_futures(wait_for) if wait_for else None
future = self._client.submit(
task,
@@ -357,7 +357,9 @@ def submit(
dependencies=dependencies,
return_type="state",
)
- return PrefectDaskFuture(wrapped_future=future, task_run_id=future.task_run_id)
+ return PrefectDaskFuture[R](
+ wrapped_future=future, task_run_id=future.task_run_id
+ )
@overload
def map(
diff --git a/src/integrations/prefect-dask/pyproject.toml b/src/integrations/prefect-dask/pyproject.toml
index 81525fa4c7f6..d76206730ca0 100644
--- a/src/integrations/prefect-dask/pyproject.toml
+++ b/src/integrations/prefect-dask/pyproject.toml
@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "prefect-dask"
dependencies = [
- "prefect>=3.0.0rc1",
+ "prefect>=3.0.0",
# don't allow versions from 2023.3.2 to 2023.5 (inclusive) due to issue with
# get_client starting in 2023.3.2 (fixed in 2023.6.0)
# https://github.com/dask/distributed/issues/7763
@@ -47,7 +47,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio >= 0.18.2, != 0.22.0, < 0.23.0",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -80,7 +80,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-dask/tests/conftest.py b/src/integrations/prefect-dask/tests/conftest.py
index 7292781b3329..338bc8fdae28 100644
--- a/src/integrations/prefect-dask/tests/conftest.py
+++ b/src/integrations/prefect-dask/tests/conftest.py
@@ -4,7 +4,6 @@
import pytest
-from prefect.settings import PREFECT_ASYNC_FETCH_STATE_RESULT, temporary_settings
from prefect.testing.utilities import prefect_test_harness
@@ -46,9 +45,3 @@ def event_loop(request):
# Workaround for failures in pytest_asyncio 0.17;
# see https://github.com/pytest-dev/pytest-asyncio/issues/257
policy.set_event_loop(loop)
-
-
-@pytest.fixture(autouse=True)
-def fetch_state_result():
- with temporary_settings(updates={PREFECT_ASYNC_FETCH_STATE_RESULT: True}):
- yield
diff --git a/src/integrations/prefect-dask/tests/test_task_runners.py b/src/integrations/prefect-dask/tests/test_task_runners.py
index 748de299c65e..d1bc5c8bec0b 100644
--- a/src/integrations/prefect-dask/tests/test_task_runners.py
+++ b/src/integrations/prefect-dask/tests/test_task_runners.py
@@ -452,3 +452,49 @@ def umbrella_flow():
return future.result()
assert umbrella_flow() == "nested task"
+
+ def test_state_dependencies_via_wait_for(self, task_runner):
+ @task
+ def task_a():
+ return time.time()
+
+ @task
+ def task_b():
+ return time.time()
+
+ @flow(task_runner=task_runner)
+ def test_flow() -> tuple[float, float]:
+ a = task_a.submit()
+ b = task_b.submit(wait_for=[a])
+ return a.result(), b.result()
+
+ a_time, b_time = test_flow()
+
+ assert b_time > a_time, "task_b timestamp should be after task_a timestamp"
+
+ def test_state_dependencies_via_wait_for_disparate_upstream_tasks(
+ self, task_runner
+ ):
+ @task
+ def task_a():
+ return time.time()
+
+ @task
+ def task_b():
+ return time.time()
+
+ @task
+ def task_c():
+ return time.time()
+
+ @flow(task_runner=task_runner)
+ def test_flow() -> tuple[float, float, float]:
+ a = task_a.submit()
+ b = task_b.submit()
+ c = task_c.submit(wait_for=[a, b])
+
+ return a.result(), b.result(), c.result()
+
+ a_time, b_time, c_time = test_flow()
+
+ assert c_time > a_time and c_time > b_time
diff --git a/src/integrations/prefect-databricks/pyproject.toml b/src/integrations/prefect-databricks/pyproject.toml
index 935c5e724276..3b75338caad7 100644
--- a/src/integrations/prefect-databricks/pyproject.toml
+++ b/src/integrations/prefect-databricks/pyproject.toml
@@ -22,7 +22,7 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries",
]
-dependencies = ["prefect>=3.0.0rc1"]
+dependencies = ["prefect>=3.0.0"]
dynamic = ["version"]
[project.optional-dependencies]
@@ -37,7 +37,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
"respx",
@@ -71,7 +71,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-dbt/prefect_dbt/__init__.py b/src/integrations/prefect-dbt/prefect_dbt/__init__.py
index f0e463e9bcfd..5e39db25c398 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/__init__.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/__init__.py
@@ -1,6 +1,6 @@
from . import _version
-from .cloud import DbtCloudCredentials # noqa
+from .cloud import DbtCloudCredentials, DbtCloudJob # noqa
from .cli import ( # noqa
DbtCliProfile,
GlobalConfigs,
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
index 2193de75f037..f7da28a31fe3 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/commands.py
@@ -290,7 +290,7 @@ class DbtCoreOperation(ShellOperation):
_block_type_name = "dbt Core Operation"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/commands/#prefect_dbt.cli.commands.DbtCoreOperation" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
profiles_dir: Optional[Path] = Field(
default=None,
@@ -929,9 +929,11 @@ def _create_unsuccessful_markdown(run_results: dict) -> str:
n.node.resource_type,
n.message,
n.node.path,
- n.node.compiled_code
- if n.node.resource_type not in ["seed", "source"]
- else None,
+ (
+ n.node.compiled_code
+ if n.node.resource_type not in ["seed", "source"]
+ else None
+ ),
)
if len(run_results["Fail"]) > 0:
markdown += "\n### Failed Nodes:\n"
@@ -941,9 +943,11 @@ def _create_unsuccessful_markdown(run_results: dict) -> str:
n.node.resource_type,
n.message,
n.node.path,
- n.node.compiled_code
- if n.node.resource_type not in ["seed", "source"]
- else None,
+ (
+ n.node.compiled_code
+ if n.node.resource_type not in ["seed", "source"]
+ else None
+ ),
)
if len(run_results["Skipped"]) > 0:
markdown += "\n### Skipped Nodes:\n"
@@ -953,9 +957,11 @@ def _create_unsuccessful_markdown(run_results: dict) -> str:
n.node.resource_type,
n.message,
n.node.path,
- n.node.compiled_code
- if n.node.resource_type not in ["seed", "source"]
- else None,
+ (
+ n.node.compiled_code
+ if n.node.resource_type not in ["seed", "source"]
+ else None
+ ),
)
if len(run_results["Warn"]) > 0:
markdown += "\n### Warned Nodes:\n"
@@ -965,9 +971,11 @@ def _create_unsuccessful_markdown(run_results: dict) -> str:
n.node.resource_type,
n.message,
n.node.path,
- n.node.compiled_code
- if n.node.resource_type not in ["seed", "source"]
- else None,
+ (
+ n.node.compiled_code
+ if n.node.resource_type not in ["seed", "source"]
+ else None
+ ),
)
return markdown
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/base.py b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/base.py
index 1028973e55c1..4aff4a005c1c 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/base.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/base.py
@@ -34,7 +34,7 @@ class DbtConfigs(Block, abc.ABC):
"fields provided in extras and credentials."
),
)
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/configs/base/#prefect_dbt.cli.configs.base.DbtConfigs" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
def _populate_configs_json(
self,
@@ -145,7 +145,7 @@ class TargetConfigs(BaseTargetConfigs):
_block_type_name = "dbt CLI Target Configs"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/configs/base/#prefect_dbt.cli.configs.base.TargetConfigs" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
class GlobalConfigs(DbtConfigs):
@@ -189,7 +189,7 @@ class GlobalConfigs(DbtConfigs):
_block_type_name = "dbt CLI Global Configs"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/configs/base/#prefect_dbt.cli.configs.base.GlobalConfigs" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
send_anonymous_usage_stats: Optional[bool] = Field(
default=None,
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/bigquery.py b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/bigquery.py
index 27eb98701f84..e799f674496c 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/bigquery.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/bigquery.py
@@ -56,7 +56,7 @@ class BigQueryTargetConfigs(BaseTargetConfigs):
_block_type_name = "dbt CLI BigQuery Target Configs"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
_description = "dbt CLI target configs containing credentials and settings, specific to BigQuery." # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/configs/bigquery/#prefect_dbt.cli.configs.bigquery.BigQueryTargetConfigs" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
type: Literal["bigquery"] = Field(
default="bigquery", description="The type of target."
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/postgres.py b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/postgres.py
index 19561ee25de2..a207465840d1 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/postgres.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/postgres.py
@@ -30,7 +30,7 @@ class PostgresTargetConfigs(BaseTargetConfigs):
_block_type_name = "dbt CLI Postgres Target Configs"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
_description = "dbt CLI target configs containing credentials and settings specific to Postgres." # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/configs/postgres/#prefect_dbt.cli.configs.postgres.PostgresTargetConfigs" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
type: Literal["postgres"] = Field(
default="postgres", description="The type of the target."
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/snowflake.py b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/snowflake.py
index 124e5bd62ca9..3398f54a1bfd 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/configs/snowflake.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/configs/snowflake.py
@@ -64,7 +64,7 @@ class SnowflakeTargetConfigs(BaseTargetConfigs):
_block_type_name = "dbt CLI Snowflake Target Configs"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/configs/snowflake/#prefect_dbt.cli.configs.snowflake.SnowflakeTargetConfigs" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
type: Literal["snowflake"] = Field(
default="snowflake", description="The type of the target configs."
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cli/credentials.py b/src/integrations/prefect-dbt/prefect_dbt/cli/credentials.py
index c6bfe0077162..5fa4e9cd9163 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cli/credentials.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cli/credentials.py
@@ -108,7 +108,7 @@ class DbtCliProfile(Block):
_block_type_name = "dbt CLI Profile"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cli/credentials/#prefect_dbt.cli.credentials.DbtCliProfile" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
name: str = Field(
default=..., description="Profile name used for populating profiles.yml."
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cloud/credentials.py b/src/integrations/prefect-dbt/prefect_dbt/cloud/credentials.py
index a4a756207634..82d4eb179efa 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cloud/credentials.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cloud/credentials.py
@@ -61,7 +61,7 @@ def trigger_dbt_cloud_job_run_flow():
_block_type_name = "dbt Cloud Credentials"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cloud/credentials/#prefect_dbt.cloud.credentials.DbtCloudCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
api_key: SecretStr = Field(
default=...,
diff --git a/src/integrations/prefect-dbt/prefect_dbt/cloud/jobs.py b/src/integrations/prefect-dbt/prefect_dbt/cloud/jobs.py
index d1694f3d74bc..b493cc1e3bd9 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/cloud/jobs.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/cloud/jobs.py
@@ -1006,7 +1006,7 @@ def dbt_cloud_job_flow():
_block_type_name = "dbt Cloud Job"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/5zE9lxfzBHjw3tnEup4wWL/9a001902ed43a84c6c96d23b24622e19/dbt-bit_tm.png?h=250" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-dbt/cloud/jobs/#prefect_dbt.cloud.jobs.DbtCloudJob" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-dbt" # noqa
dbt_cloud_credentials: DbtCloudCredentials = Field(
default=...,
diff --git a/src/integrations/prefect-dbt/pyproject.toml b/src/integrations/prefect-dbt/pyproject.toml
index 64b15b1a9fd5..32d6b9bd1d98 100644
--- a/src/integrations/prefect-dbt/pyproject.toml
+++ b/src/integrations/prefect-dbt/pyproject.toml
@@ -23,23 +23,23 @@ classifiers = [
"Topic :: Software Development :: Libraries",
]
dependencies = [
- "prefect>=3.0.0rc1",
+ "prefect>=3.0.0",
"dbt-core>=1.7.0",
- "prefect_shell>=0.3.0rc1",
+ "prefect_shell>=0.3.0",
"sgqlc>=16.0.0",
]
dynamic = ["version"]
[project.optional-dependencies]
-snowflake = ["prefect-snowflake>=0.28.0rc1", "dbt-snowflake"]
-bigquery = ["prefect-gcp[bigquery]>=0.6.0rc1", "dbt-bigquery"]
+snowflake = ["prefect-snowflake>=0.28.0", "dbt-snowflake"]
+bigquery = ["prefect-gcp[bigquery]>=0.6.0", "dbt-bigquery"]
postgres = ["prefect-sqlalchemy>=0.5.1", "dbt-postgres"]
all_extras = [
"dbt-bigquery",
"dbt-postgres",
"dbt-snowflake",
- "prefect-gcp[bigquery]>=0.6.0rc1",
- "prefect-snowflake>=0.28.0rc1",
+ "prefect-gcp[bigquery]>=0.6.0",
+ "prefect-snowflake>=0.28.0",
"prefect-sqlalchemy>=0.5.1",
]
dev = [
@@ -55,10 +55,10 @@ dev = [
"mypy",
"pillow",
"pre-commit",
- "prefect-gcp[bigquery]>=0.6.0rc1",
- "prefect-snowflake>=0.28.0rc1",
+ "prefect-gcp[bigquery]>=0.6.0",
+ "prefect-snowflake>=0.28.0",
"prefect-sqlalchemy>=0.5.1",
- "pytest",
+ "pytest >= 8.3",
"pytest-asyncio",
"pytest-env",
"pytest-xdist",
@@ -93,7 +93,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-docker/prefect_docker/worker.py b/src/integrations/prefect-docker/prefect_docker/worker.py
index e529b123375d..24a57631046b 100644
--- a/src/integrations/prefect-docker/prefect_docker/worker.py
+++ b/src/integrations/prefect-docker/prefect_docker/worker.py
@@ -396,7 +396,7 @@ class DockerWorker(BaseWorker):
"Docker daemon."
)
_display_name = "Docker"
- _documentation_url = "https://prefecthq.github.io/prefect-docker/worker/"
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-docker"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/2IfXXfMq66mrzJBDFFCHTp/6d8f320d9e4fc4393f045673d61ab612/Moby-logo.png?h=250" # noqa
def __init__(
diff --git a/src/integrations/prefect-docker/pyproject.toml b/src/integrations/prefect-docker/pyproject.toml
index db89b60136cd..e947a46f52f9 100644
--- a/src/integrations/prefect-docker/pyproject.toml
+++ b/src/integrations/prefect-docker/pyproject.toml
@@ -22,11 +22,7 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries",
]
-dependencies = [
- "prefect>=3.0.0rc1",
- "docker>=6.1.1",
- "exceptiongroup",
-]
+dependencies = ["prefect>=3.0.0", "docker>=6.1.1", "exceptiongroup"]
dynamic = ["version"]
[project.optional-dependencies]
@@ -41,7 +37,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -74,6 +70,7 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
env = ["PREFECT_TEST_MODE=1"]
filterwarnings = [
diff --git a/src/integrations/prefect-email/prefect_email/credentials.py b/src/integrations/prefect-email/prefect_email/credentials.py
index fa9e24db0484..1280d161dec1 100644
--- a/src/integrations/prefect-email/prefect_email/credentials.py
+++ b/src/integrations/prefect-email/prefect_email/credentials.py
@@ -97,7 +97,7 @@ class EmailServerCredentials(Block):
_block_type_name = "Email Server Credentials"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/82bc6ed16ca42a2252a5512c72233a253b8a58eb-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-email/credentials/#prefect_email.credentials.EmailServerCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-email" # noqa
username: Optional[str] = Field(
default=None,
diff --git a/src/integrations/prefect-email/pyproject.toml b/src/integrations/prefect-email/pyproject.toml
index 1cce0fee97e4..8a3c67d19fbe 100644
--- a/src/integrations/prefect-email/pyproject.toml
+++ b/src/integrations/prefect-email/pyproject.toml
@@ -22,7 +22,7 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries",
]
-dependencies = ["prefect>=3.0.0rc1"]
+dependencies = ["prefect>=3.0.0"]
dynamic = ["version"]
[project.optional-dependencies]
@@ -38,7 +38,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -71,7 +71,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-gcp/README.md b/src/integrations/prefect-gcp/README.md
index 586fd4e082d7..f8e76ae31397 100644
--- a/src/integrations/prefect-gcp/README.md
+++ b/src/integrations/prefect-gcp/README.md
@@ -9,18 +9,12 @@
`prefect-gcp` makes it easy to leverage the capabilities of Google Cloud Platform (GCP) in your flows, featuring support for Vertex AI, Cloud Run, BigQuery, Cloud Storage, and Secret Manager.
-Visit the full docs [here](https://PrefectHQ.github.io/prefect-gcp).
+Visit the full docs [here](https://docs.prefect.io/integrations/prefect-gcp).
-### Installation
+## Installation
To start using `prefect-gcp`:
```bash
pip install prefect-gcp
```
-
-To install extras, see [here](https://prefecthq.github.io/prefect-gcp/#installation).
-
-### Contributing
-
-Thanks for thinking about chipping in! Check out this [step-by-step guide](https://prefecthq.github.io/prefect-gcp/#installation) on how to get started.
diff --git a/src/integrations/prefect-gcp/prefect_gcp/bigquery.py b/src/integrations/prefect-gcp/prefect_gcp/bigquery.py
index 5e0c3522ada9..0b69c0496838 100644
--- a/src/integrations/prefect-gcp/prefect_gcp/bigquery.py
+++ b/src/integrations/prefect-gcp/prefect_gcp/bigquery.py
@@ -560,7 +560,7 @@ class BigQueryWarehouse(DatabaseBlock):
_block_type_name = "BigQuery Warehouse"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/10424e311932e31c477ac2b9ef3d53cefbaad708-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-gcp/bigquery/#prefect_gcp.bigquery.BigQueryWarehouse" # noqa: E501
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-gcp" # noqa: E501
gcp_credentials: GcpCredentials
fetch_size: int = Field(
diff --git a/src/integrations/prefect-gcp/prefect_gcp/cloud_storage.py b/src/integrations/prefect-gcp/prefect_gcp/cloud_storage.py
index 68ea9bdbace0..b7b6183c5e8c 100644
--- a/src/integrations/prefect-gcp/prefect_gcp/cloud_storage.py
+++ b/src/integrations/prefect-gcp/prefect_gcp/cloud_storage.py
@@ -581,7 +581,7 @@ class GcsBucket(WritableDeploymentStorage, WritableFileSystem, ObjectStorageBloc
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/10424e311932e31c477ac2b9ef3d53cefbaad708-250x250.png" # noqa
_block_type_name = "GCS Bucket"
- _documentation_url = "https://prefecthq.github.io/prefect-gcp/cloud_storage/#prefect_gcp.cloud_storage.GcsBucket" # noqa: E501
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-gcp" # noqa: E501
bucket: str = Field(..., description="Name of the bucket.")
gcp_credentials: GcpCredentials = Field(
diff --git a/src/integrations/prefect-gcp/prefect_gcp/credentials.py b/src/integrations/prefect-gcp/prefect_gcp/credentials.py
index 33703fe0cf72..45165c8c63e8 100644
--- a/src/integrations/prefect-gcp/prefect_gcp/credentials.py
+++ b/src/integrations/prefect-gcp/prefect_gcp/credentials.py
@@ -124,7 +124,7 @@ class GcpCredentials(CredentialsBlock):
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/10424e311932e31c477ac2b9ef3d53cefbaad708-250x250.png" # noqa
_block_type_name = "GCP Credentials"
- _documentation_url = "https://prefecthq.github.io/prefect-gcp/credentials/#prefect_gcp.credentials.GcpCredentials" # noqa: E501
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-gcp" # noqa: E501
service_account_file: Optional[Path] = Field(
default=None, description="Path to the service account JSON keyfile."
diff --git a/src/integrations/prefect-gcp/prefect_gcp/secret_manager.py b/src/integrations/prefect-gcp/prefect_gcp/secret_manager.py
index 6f90f8f4ec98..edd90a90f337 100644
--- a/src/integrations/prefect-gcp/prefect_gcp/secret_manager.py
+++ b/src/integrations/prefect-gcp/prefect_gcp/secret_manager.py
@@ -307,7 +307,7 @@ class GcpSecret(SecretBlock):
"""
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/10424e311932e31c477ac2b9ef3d53cefbaad708-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-gcp/secret_manager/#prefect_gcp.secret_manager.GcpSecret" # noqa: E501
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-gcp" # noqa: E501
gcp_credentials: GcpCredentials
secret_name: str = Field(default=..., description="Name of the secret to manage.")
diff --git a/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run.py b/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run.py
index bc20017ee06e..298386ef168d 100644
--- a/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run.py
+++ b/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run.py
@@ -541,7 +541,7 @@ class CloudRunWorker(BaseWorker):
"a Google Cloud Platform account."
)
_display_name = "Google Cloud Run"
- _documentation_url = "https://prefecthq.github.io/prefect-gcp/cloud_run_worker/"
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-gcp"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/10424e311932e31c477ac2b9ef3d53cefbaad708-250x250.png" # noqa
def _create_job_error(self, exc, configuration):
diff --git a/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run_v2.py b/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run_v2.py
index c7836d324695..0d3b6989378a 100644
--- a/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run_v2.py
+++ b/src/integrations/prefect-gcp/prefect_gcp/workers/cloud_run_v2.py
@@ -211,7 +211,7 @@ def _populate_env(self):
]
envs.extend(envs_from_secrets)
- self.job_body["template"]["template"]["containers"][0]["env"] = envs
+ self.job_body["template"]["template"]["containers"][0]["env"].extend(envs)
def _configure_cloudsql_volumes(self):
"""
@@ -226,7 +226,10 @@ def _configure_cloudsql_volumes(self):
if "volumes" not in template:
template["volumes"] = []
template["volumes"].append(
- {"name": "cloudsql", "cloudSqlInstance": self.cloudsql_instances}
+ {
+ "name": "cloudsql",
+ "cloudSqlInstance": {"instances": self.cloudsql_instances},
+ }
)
if "volumeMounts" not in containers[0]:
containers[0]["volumeMounts"] = []
@@ -469,7 +472,7 @@ class CloudRunWorkerV2(BaseWorker):
job_configuration_variables = CloudRunWorkerV2Variables
_description = "Execute flow runs within containers on Google Cloud Run (V2 API). Requires a Google Cloud Platform account." # noqa
_display_name = "Google Cloud Run V2"
- _documentation_url = "https://prefecthq.github.io/prefect-gcp/worker_v2/"
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-gcp"
_logo_url = "https://images.ctfassets.net/gm98wzqotmnx/4SpnOBvMYkHp6z939MDKP6/549a91bc1ce9afd4fb12c68db7b68106/social-icon-google-cloud-1200-630.png?h=250" # noqa
async def run(
diff --git a/src/integrations/prefect-gcp/prefect_gcp/workers/vertex.py b/src/integrations/prefect-gcp/prefect_gcp/workers/vertex.py
index add6ec12798a..d11efff4c3d0 100644
--- a/src/integrations/prefect-gcp/prefect_gcp/workers/vertex.py
+++ b/src/integrations/prefect-gcp/prefect_gcp/workers/vertex.py
@@ -372,7 +372,7 @@ class VertexAIWorker(BaseWorker):
"a Google Cloud Platform account."
)
_display_name = "Google Vertex AI"
- _documentation_url = "https://prefecthq.github.io/prefect-gcp/vertex_worker/"
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-gcp" # noqa
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/10424e311932e31c477ac2b9ef3d53cefbaad708-250x250.png" # noqa
async def run(
diff --git a/src/integrations/prefect-gcp/pyproject.toml b/src/integrations/prefect-gcp/pyproject.toml
index 69f90adbd206..21c9e5b1e855 100644
--- a/src/integrations/prefect-gcp/pyproject.toml
+++ b/src/integrations/prefect-gcp/pyproject.toml
@@ -23,7 +23,7 @@ classifiers = [
"Topic :: Software Development :: Libraries",
]
dependencies = [
- "prefect>=3.0.0rc1",
+ "prefect>=3.0.0",
"google-api-python-client>=2.20.0",
"google-cloud-storage>=2.0.0",
"tenacity>=8.0.0",
@@ -61,7 +61,7 @@ dev = [
"pillow",
"pre-commit",
"pyarrow",
- "pytest",
+ "pytest >= 8.3",
"pytest-asyncio",
"pytest-env",
"pytest-xdist",
@@ -95,10 +95,9 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
filterwarnings = [
"ignore:Type google._upb._message.* uses PyType_Spec with a metaclass that has custom tp_new. This is deprecated and will no longer be allowed in Python 3.14:DeprecationWarning",
]
diff --git a/src/integrations/prefect-gcp/tests/test_cloud_run_worker_v2.py b/src/integrations/prefect-gcp/tests/test_cloud_run_worker_v2.py
index 597edd0352a0..5b422df5c6be 100644
--- a/src/integrations/prefect-gcp/tests/test_cloud_run_worker_v2.py
+++ b/src/integrations/prefect-gcp/tests/test_cloud_run_worker_v2.py
@@ -120,6 +120,29 @@ def test_populate_env_with_secrets(self, cloud_run_worker_v2_job_config):
},
]
+ def test_populate_env_with_existing_envs(self, cloud_run_worker_v2_job_config):
+ cloud_run_worker_v2_job_config.job_body["template"]["template"]["containers"][
+ 0
+ ]["env"] = [{"name": "ENV0", "value": "VALUE0"}]
+ cloud_run_worker_v2_job_config.env_from_secrets = {
+ "SECRET_ENV1": SecretKeySelector(secret="SECRET1", version="latest")
+ }
+ cloud_run_worker_v2_job_config._populate_env()
+
+ assert cloud_run_worker_v2_job_config.job_body["template"]["template"][
+ "containers"
+ ][0]["env"] == [
+ {"name": "ENV0", "value": "VALUE0"},
+ {"name": "ENV1", "value": "VALUE1"},
+ {"name": "ENV2", "value": "VALUE2"},
+ {
+ "name": "SECRET_ENV1",
+ "valueSource": {
+ "secretKeyRef": {"secret": "SECRET1", "version": "latest"}
+ },
+ },
+ ]
+
def test_populate_image_if_not_present(self, cloud_run_worker_v2_job_config):
cloud_run_worker_v2_job_config._populate_image_if_not_present()
@@ -231,7 +254,7 @@ def test_configure_cloudsql_volumes_preserves_existing_volumes(
assert template["volumes"][0] == {"name": "existing-volume", "emptyDir": {}}
assert template["volumes"][1] == {
"name": "cloudsql",
- "cloudSqlInstance": ["project:region:instance1"],
+ "cloudSqlInstance": {"instances": ["project:region:instance1"]},
}
assert len(template["containers"][0]["volumeMounts"]) == 2
@@ -261,7 +284,7 @@ class MockFlowRun:
assert any(
vol["name"] == "cloudsql"
- and vol["cloudSqlInstance"] == ["project:region:instance1"]
+ and vol["cloudSqlInstance"]["instances"] == ["project:region:instance1"]
for vol in template["volumes"]
)
assert any(
diff --git a/src/integrations/prefect-github/prefect_github/credentials.py b/src/integrations/prefect-github/prefect_github/credentials.py
index 24fe819be558..1e5971a6e9d3 100644
--- a/src/integrations/prefect-github/prefect_github/credentials.py
+++ b/src/integrations/prefect-github/prefect_github/credentials.py
@@ -25,7 +25,7 @@ class GitHubCredentials(CredentialsBlock):
_block_type_name = "GitHub Credentials"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/41971cfecfea5f79ff334164f06ecb34d1038dd4-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-github/credentials/#prefect_github.credentials.GitHubCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-github"
token: Optional[SecretStr] = Field(
default=None, description="A GitHub personal access token (PAT)."
diff --git a/src/integrations/prefect-github/prefect_github/repository.py b/src/integrations/prefect-github/prefect_github/repository.py
index 4742bc2d8117..57551f2909f3 100644
--- a/src/integrations/prefect-github/prefect_github/repository.py
+++ b/src/integrations/prefect-github/prefect_github/repository.py
@@ -39,7 +39,7 @@ class GitHubRepository(ReadableDeploymentStorage):
_block_type_name = "GitHub Repository"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/41971cfecfea5f79ff334164f06ecb34d1038dd4-250x250.png" # noqa: E501
- _documentation_url = "https://prefecthq.github.io/prefect-github/repository/#prefect_github.repository.GitHubRepository" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-github" # noqa
repository_url: str = Field(
default=...,
diff --git a/src/integrations/prefect-github/pyproject.toml b/src/integrations/prefect-github/pyproject.toml
index c7ded155c8ce..3d5259993096 100644
--- a/src/integrations/prefect-github/pyproject.toml
+++ b/src/integrations/prefect-github/pyproject.toml
@@ -4,10 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "prefect-github"
-dependencies = [
- "sgqlc>=15.0",
- "prefect>=3.0.0rc1",
-]
+dependencies = ["sgqlc>=15.0", "prefect>=3.0.0"]
dynamic = ["version"]
description = "Prefect integrations interacting with GitHub"
readme = "README.md"
@@ -40,7 +37,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -73,7 +70,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-gitlab/pyproject.toml b/src/integrations/prefect-gitlab/pyproject.toml
index 2e57871fd70d..516b1af20e5a 100644
--- a/src/integrations/prefect-gitlab/pyproject.toml
+++ b/src/integrations/prefect-gitlab/pyproject.toml
@@ -22,11 +22,7 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries",
]
-dependencies = [
- "prefect>=3.0.0rc1",
- "python-gitlab>=3.12.0",
- "tenacity>=8.2.3",
-]
+dependencies = ["prefect>=3.0.0", "python-gitlab>=3.12.0", "tenacity>=8.2.3"]
dynamic = ["version"]
[project.optional-dependencies]
@@ -42,7 +38,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -75,7 +71,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py b/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py
index 802edc9b5007..d753317824d2 100644
--- a/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py
+++ b/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py
@@ -52,7 +52,7 @@ class KubernetesClusterConfig(Block):
_block_type_name = "Kubernetes Cluster Config"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/2d0b896006ad463b49c28aaac14f31e00e32cfab-250x250.png"
- _documentation_url = "https://prefecthq.github.io/prefect-kubernetes/credentials/#prefect_kubernetes.credentials.KubernetesClusterConfig" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-kubernetes" # noqa
config: Dict = Field(
default=..., description="The entire contents of a kubectl config file."
)
@@ -142,7 +142,7 @@ class KubernetesCredentials(Block):
_block_type_name = "Kubernetes Credentials"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/2d0b896006ad463b49c28aaac14f31e00e32cfab-250x250.png" # noqa
- _documentation_url = "https://prefecthq.github.io/prefect-kubernetes/credentials/#prefect_kubernetes.credentials.KubernetesCredentials" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-kubernetes" # noqa
cluster_config: Optional[KubernetesClusterConfig] = None
diff --git a/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py b/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py
index d3b9ec828429..5650ca551f7b 100644
--- a/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py
+++ b/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py
@@ -540,7 +540,7 @@ class KubernetesJob(JobBlock):
_block_type_name = "Kubernetes Job"
_block_type_slug = "k8s-job"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/2d0b896006ad463b49c28aaac14f31e00e32cfab-250x250.png" # noqa: E501
- _documentation_url = "https://prefecthq.github.io/prefect-kubernetes/jobs/#prefect_kubernetes.jobs.KubernetesJob" # noqa
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-kubernetes" # noqa
@sync_compatible
async def trigger(self):
diff --git a/src/integrations/prefect-kubernetes/prefect_kubernetes/settings.py b/src/integrations/prefect-kubernetes/prefect_kubernetes/settings.py
new file mode 100644
index 000000000000..4dbada798b57
--- /dev/null
+++ b/src/integrations/prefect-kubernetes/prefect_kubernetes/settings.py
@@ -0,0 +1,53 @@
+from typing import Optional
+
+from pydantic import AliasChoices, AliasPath, Field
+
+from prefect.settings.base import PrefectBaseSettings, _build_settings_config
+
+
+class KubernetesWorkerSettings(PrefectBaseSettings):
+ model_config = _build_settings_config(("integrations", "kubernetes", "worker"))
+
+ api_key_secret_name: Optional[str] = Field(
+ default=None,
+ description="The name of the secret the worker's API key is stored in.",
+ )
+
+ create_secret_for_api_key: bool = Field(
+ default=False,
+ description="If `True`, the worker will create a secret in the same namespace as created Kubernetes jobs to store the Prefect API key.",
+ validation_alias=AliasChoices(
+ AliasPath("create_secret_for_api_key"),
+ "prefect_integrations_kubernetes_worker_create_secret_for_api_key",
+ "prefect_kubernetes_worker_store_prefect_api_in_secret",
+ ),
+ )
+
+ add_tcp_keepalive: bool = Field(
+ default=True,
+ description="If `True`, the worker will add TCP keepalive to the Kubernetes client.",
+ validation_alias=AliasChoices(
+ AliasPath("add_tcp_keepalive"),
+ "prefect_integrations_kubernetes_worker_add_tcp_keepalive",
+ "prefect_kubernetes_worker_add_tcp_keepalive",
+ ),
+ )
+
+
+class KubernetesSettings(PrefectBaseSettings):
+ model_config = _build_settings_config(("integrations", "kubernetes"))
+
+ cluster_uid: Optional[str] = Field(
+ default=None,
+ description="A unique identifier for the current cluster being used.",
+ validation_alias=AliasChoices(
+ AliasPath("cluster_uid"),
+ "prefect_integrations_kubernetes_cluster_uid",
+ "prefect_kubernetes_cluster_uid",
+ ),
+ )
+
+ worker: KubernetesWorkerSettings = Field(
+ description="Settings for controlling Kubernetes worker behavior.",
+ default_factory=KubernetesWorkerSettings,
+ )
diff --git a/src/integrations/prefect-kubernetes/prefect_kubernetes/worker.py b/src/integrations/prefect-kubernetes/prefect_kubernetes/worker.py
index e51e66667fe8..0d3f58637d66 100644
--- a/src/integrations/prefect-kubernetes/prefect_kubernetes/worker.py
+++ b/src/integrations/prefect-kubernetes/prefect_kubernetes/worker.py
@@ -14,11 +14,11 @@
### Securing your Prefect Cloud API key
If you are using Prefect Cloud and would like to pass your Prefect Cloud API key to
created jobs via a Kubernetes secret, set the
-`PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET` environment variable before
+`PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY` environment variable before
starting your worker:
```bash
-export PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET="true"
+export PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY="true"
prefect worker start --pool 'my-work-pool' --type kubernetes
```
@@ -105,7 +105,6 @@
import enum
import json
import logging
-import os
import shlex
from contextlib import asynccontextmanager
from datetime import datetime
@@ -160,6 +159,7 @@
)
from prefect_kubernetes.credentials import KubernetesClusterConfig
from prefect_kubernetes.events import KubernetesEventsReplicator
+from prefect_kubernetes.settings import KubernetesSettings
from prefect_kubernetes.utilities import (
KeepAliveClientRequest,
_slugify_label_key,
@@ -566,7 +566,7 @@ class KubernetesWorker(BaseWorker):
"Kubernetes cluster."
)
_display_name = "Kubernetes"
- _documentation_url = "https://prefecthq.github.io/prefect-kubernetes/worker/"
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-kubernetes"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/2d0b896006ad463b49c28aaac14f31e00e32cfab-250x250.png" # noqa
def __init__(self, *args, **kwargs):
@@ -650,6 +650,7 @@ async def _get_configured_kubernetes_client(
Returns a configured Kubernetes client.
"""
client = None
+ settings = KubernetesSettings()
if configuration.cluster_config:
config_dict = configuration.cluster_config.config
@@ -667,9 +668,7 @@ async def _get_configured_kubernetes_client(
# If in-cluster config fails, load the local kubeconfig
client = await config.new_client_from_config()
- if os.environ.get(
- "PREFECT_KUBERNETES_WORKER_ADD_TCP_KEEPALIVE", "TRUE"
- ).strip().lower() in ("true", "1"):
+ if settings.worker.add_tcp_keepalive:
client.rest_client.pool_manager._request_class = KeepAliveClientRequest
try:
@@ -678,7 +677,10 @@ async def _get_configured_kubernetes_client(
await client.close()
async def _replace_api_key_with_secret(
- self, configuration: KubernetesWorkerJobConfiguration, client: "ApiClient"
+ self,
+ configuration: KubernetesWorkerJobConfiguration,
+ client: "ApiClient",
+ secret_name: Optional[str] = None,
):
"""Replaces the PREFECT_API_KEY environment variable with a Kubernetes secret"""
manifest_env = configuration.job_manifest["spec"]["template"]["spec"][
@@ -693,7 +695,7 @@ async def _replace_api_key_with_secret(
{},
)
api_key = manifest_api_key_env.get("value")
- if api_key:
+ if api_key and not secret_name:
secret_name = f"prefect-{_slugify_name(self.name)}-api-key"
secret = await self._upsert_secret(
name=secret_name,
@@ -706,6 +708,7 @@ async def _replace_api_key_with_secret(
self._created_secrets[
(secret.metadata.name, secret.metadata.namespace)
] = configuration
+ if secret_name:
new_api_env_entry = {
"name": "PREFECT_API_KEY",
"valueFrom": {"secretKeyRef": {"name": secret_name, "key": "value"}},
@@ -733,9 +736,14 @@ async def _create_job(
"""
Creates a Kubernetes job from a job manifest.
"""
- if os.environ.get(
- "PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET", ""
- ).strip().lower() in ("true", "1"):
+ settings = KubernetesSettings()
+ if settings.worker.api_key_secret_name:
+ await self._replace_api_key_with_secret(
+ configuration=configuration,
+ client=client,
+ secret_name=settings.worker.api_key_secret_name,
+ )
+ elif settings.worker.create_secret_for_api_key:
await self._replace_api_key_with_secret(
configuration=configuration, client=client
)
@@ -838,8 +846,9 @@ async def _get_cluster_uid(self, client: "ApiClient") -> str:
See https://github.com/kubernetes/kubernetes/issues/44954
"""
+ settings = KubernetesSettings()
# Default to an environment variable
- env_cluster_uid = os.environ.get("PREFECT_KUBERNETES_CLUSTER_UID")
+ env_cluster_uid = settings.cluster_uid
if env_cluster_uid:
return env_cluster_uid
diff --git a/src/integrations/prefect-kubernetes/pyproject.toml b/src/integrations/prefect-kubernetes/pyproject.toml
index 37245b5e0fec..427f4c839aea 100644
--- a/src/integrations/prefect-kubernetes/pyproject.toml
+++ b/src/integrations/prefect-kubernetes/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
dependencies = [
- "prefect>=3.0.0rc1",
+ "prefect>=3.1.0",
"kubernetes-asyncio>=29.0.0",
"tenacity>=8.2.3",
"exceptiongroup",
@@ -45,7 +45,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-timeout",
"pytest-xdist",
@@ -79,6 +79,7 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
env = ["PREFECT_TEST_MODE=1"]
timeout = 30
diff --git a/src/integrations/prefect-kubernetes/tests/test_settings.py b/src/integrations/prefect-kubernetes/tests/test_settings.py
new file mode 100644
index 000000000000..1f808d5c005a
--- /dev/null
+++ b/src/integrations/prefect-kubernetes/tests/test_settings.py
@@ -0,0 +1,111 @@
+import os
+
+import toml
+from prefect_kubernetes.settings import KubernetesSettings
+
+
+def test_set_values_via_environment_variables(monkeypatch):
+ monkeypatch.setenv(
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_API_KEY_SECRET_NAME", "test-secret"
+ )
+ monkeypatch.setenv(
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY", "true"
+ )
+ monkeypatch.setenv(
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_ADD_TCP_KEEPALIVE", "false"
+ )
+ monkeypatch.setenv(
+ "PREFECT_INTEGRATIONS_KUBERNETES_CLUSTER_UID", "test-cluster-uid"
+ )
+
+ settings = KubernetesSettings()
+
+ assert settings.worker.api_key_secret_name == "test-secret"
+ assert settings.worker.create_secret_for_api_key is True
+ assert settings.worker.add_tcp_keepalive is False
+ assert settings.cluster_uid == "test-cluster-uid"
+
+
+def test_set_values_via_dot_env_file(tmp_path):
+ dot_env_path = tmp_path / ".env"
+ with open(dot_env_path, "w") as f:
+ f.write(
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_API_KEY_SECRET_NAME=test-secret\n"
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY=true\n"
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_ADD_TCP_KEEPALIVE=false\n"
+ "PREFECT_INTEGRATIONS_KUBERNETES_CLUSTER_UID=test-cluster-uid\n"
+ )
+
+ original_dir = os.getcwd()
+ try:
+ os.chdir(tmp_path)
+ settings = KubernetesSettings()
+ finally:
+ os.chdir(original_dir)
+
+ assert settings.worker.api_key_secret_name == "test-secret"
+ assert settings.worker.create_secret_for_api_key is True
+ assert settings.worker.add_tcp_keepalive is False
+ assert settings.cluster_uid == "test-cluster-uid"
+
+
+def test_set_values_via_prefect_toml_file(tmp_path):
+ toml_path = tmp_path / "prefect.toml"
+ toml_data = {
+ "integrations": {
+ "kubernetes": {
+ "worker": {
+ "api_key_secret_name": "test-secret",
+ "create_secret_for_api_key": True,
+ "add_tcp_keepalive": False,
+ },
+ "cluster_uid": "test-cluster-uid",
+ },
+ },
+ }
+ toml_path.write_text(toml.dumps(toml_data))
+
+ original_dir = os.getcwd()
+ try:
+ os.chdir(tmp_path)
+ settings = KubernetesSettings()
+ finally:
+ os.chdir(original_dir)
+
+ assert settings.worker.api_key_secret_name == "test-secret"
+ assert settings.worker.create_secret_for_api_key is True
+ assert settings.worker.add_tcp_keepalive is False
+ assert settings.cluster_uid == "test-cluster-uid"
+
+
+def test_set_values_via_pyproject_toml_file(tmp_path):
+ pyproject_toml_path = tmp_path / "pyproject.toml"
+ pyproject_toml_data = {
+ "tool": {
+ "prefect": {
+ "integrations": {
+ "kubernetes": {
+ "cluster_uid": "test-cluster-uid",
+ "worker": {
+ "api_key_secret_name": "test-secret",
+ "create_secret_for_api_key": True,
+ "add_tcp_keepalive": False,
+ },
+ },
+ },
+ },
+ },
+ }
+ pyproject_toml_path.write_text(toml.dumps(pyproject_toml_data))
+
+ original_dir = os.getcwd()
+ try:
+ os.chdir(tmp_path)
+ settings = KubernetesSettings()
+ finally:
+ os.chdir(original_dir)
+
+ assert settings.worker.api_key_secret_name == "test-secret"
+ assert settings.worker.create_secret_for_api_key is True
+ assert settings.worker.add_tcp_keepalive is False
+ assert settings.cluster_uid == "test-cluster-uid"
diff --git a/src/integrations/prefect-kubernetes/tests/test_worker.py b/src/integrations/prefect-kubernetes/tests/test_worker.py
index 993841e0c23c..04deb4ce2c5a 100644
--- a/src/integrations/prefect-kubernetes/tests/test_worker.py
+++ b/src/integrations/prefect-kubernetes/tests/test_worker.py
@@ -188,7 +188,17 @@ async def mock_stream(*args, **kwargs):
@pytest.fixture
def enable_store_api_key_in_secret(monkeypatch):
- monkeypatch.setenv("PREFECT_KUBERNETES_WORKER_STORE_PREFECT_API_IN_SECRET", "true")
+ monkeypatch.setenv(
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_CREATE_SECRET_FOR_API_KEY", "true"
+ )
+
+
+@pytest.fixture
+def mock_api_key_secret_name(monkeypatch):
+ monkeypatch.setenv(
+ "PREFECT_INTEGRATIONS_KUBERNETES_WORKER_API_KEY_SECRET_NAME", "test-secret"
+ )
+ return "test-secret"
from_template_and_values_cases = [
@@ -1609,6 +1619,104 @@ async def test_store_api_key_in_existing_secret(
),
)
+ async def test_use_existing_secret_name(
+ self,
+ flow_run,
+ mock_core_client,
+ mock_watch,
+ mock_pods_stream_that_returns_running_pod,
+ mock_batch_client,
+ mock_api_key_secret_name,
+ ):
+ mock_watch.return_value.stream = mock_pods_stream_that_returns_running_pod
+
+ configuration = await KubernetesWorkerJobConfiguration.from_template_and_values(
+ KubernetesWorker.get_default_base_job_template(), {"image": "foo"}
+ )
+ with temporary_settings(updates={PREFECT_API_KEY: "fake"}):
+ async with KubernetesWorker(work_pool_name="test") as k8s_worker:
+ mock_core_client.return_value.read_namespaced_secret.return_value = (
+ V1Secret(
+ api_version="v1",
+ kind="Secret",
+ metadata=V1ObjectMeta(
+ name=f"prefect-{_slugify_name(k8s_worker.name)}-api-key",
+ namespace=configuration.namespace,
+ ),
+ data={
+ "value": base64.b64encode("fake".encode("utf-8")).decode(
+ "utf-8"
+ )
+ },
+ )
+ )
+
+ configuration.prepare_for_flow_run(flow_run=flow_run)
+ await k8s_worker.run(flow_run, configuration)
+ mock_batch_client.return_value.create_namespaced_job.assert_called_once()
+ env = mock_batch_client.return_value.create_namespaced_job.call_args[0][
+ 1
+ ]["spec"]["template"]["spec"]["containers"][0]["env"]
+ assert {
+ "name": "PREFECT_API_KEY",
+ "valueFrom": {
+ "secretKeyRef": {
+ "name": mock_api_key_secret_name,
+ "key": "value",
+ }
+ },
+ } in env
+
+ async def test_existing_secret_name_takes_precedence(
+ self,
+ flow_run,
+ mock_core_client,
+ mock_watch,
+ mock_pods_stream_that_returns_running_pod,
+ mock_batch_client,
+ mock_api_key_secret_name,
+ enable_store_api_key_in_secret,
+ ):
+ mock_watch.return_value.stream = mock_pods_stream_that_returns_running_pod
+
+ configuration = await KubernetesWorkerJobConfiguration.from_template_and_values(
+ KubernetesWorker.get_default_base_job_template(), {"image": "foo"}
+ )
+ with temporary_settings(updates={PREFECT_API_KEY: "fake"}):
+ async with KubernetesWorker(work_pool_name="test") as k8s_worker:
+ mock_core_client.return_value.read_namespaced_secret.return_value = (
+ V1Secret(
+ api_version="v1",
+ kind="Secret",
+ metadata=V1ObjectMeta(
+ name=f"prefect-{_slugify_name(k8s_worker.name)}-api-key",
+ namespace=configuration.namespace,
+ ),
+ data={
+ "value": base64.b64encode("fake".encode("utf-8")).decode(
+ "utf-8"
+ )
+ },
+ )
+ )
+
+ configuration.prepare_for_flow_run(flow_run=flow_run)
+ await k8s_worker.run(flow_run, configuration)
+ mock_batch_client.return_value.create_namespaced_job.assert_called_once()
+ env = mock_batch_client.return_value.create_namespaced_job.call_args[0][
+ 1
+ ]["spec"]["template"]["spec"]["containers"][0]["env"]
+ assert {
+ "name": "PREFECT_API_KEY",
+ "valueFrom": {
+ "secretKeyRef": {
+ "name": mock_api_key_secret_name,
+ "key": "value",
+ }
+ },
+ } in env
+ mock_core_client.return_value.replace_namespaced_secret.assert_not_called()
+
async def test_create_job_failure(
self,
flow_run,
diff --git a/src/integrations/prefect-ray/pyproject.toml b/src/integrations/prefect-ray/pyproject.toml
index f6cc61d94e93..1205e6ea1f99 100644
--- a/src/integrations/prefect-ray/pyproject.toml
+++ b/src/integrations/prefect-ray/pyproject.toml
@@ -22,10 +22,7 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries",
]
-dependencies = [
- "prefect>=3.0.0rc1",
- "ray[default]>=2.0.0",
-]
+dependencies = ["prefect>=3.0.0", "ray[default]>=2.0.0"]
dynamic = ["version"]
[project.optional-dependencies]
@@ -41,7 +38,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-asyncio>=0.18.2,!=0.22.0,<0.23.0",
"pytest-env",
"pytest-xdist",
@@ -75,7 +72,6 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-ray/tests/conftest.py b/src/integrations/prefect-ray/tests/conftest.py
index fe650e4b8b90..3723fce5f3d5 100644
--- a/src/integrations/prefect-ray/tests/conftest.py
+++ b/src/integrations/prefect-ray/tests/conftest.py
@@ -1,6 +1,5 @@
import pytest
-from prefect.settings import PREFECT_ASYNC_FETCH_STATE_RESULT, temporary_settings
from prefect.testing.utilities import prefect_test_harness
@@ -8,9 +7,3 @@
def prefect_db():
with prefect_test_harness():
yield
-
-
-@pytest.fixture(autouse=True)
-def fetch_state_result():
- with temporary_settings(updates={PREFECT_ASYNC_FETCH_STATE_RESULT: True}):
- yield
diff --git a/src/integrations/prefect-redis/prefect_redis/blocks.py b/src/integrations/prefect-redis/prefect_redis/blocks.py
index 400cf235b91c..05f390d8de02 100644
--- a/src/integrations/prefect-redis/prefect_redis/blocks.py
+++ b/src/integrations/prefect-redis/prefect_redis/blocks.py
@@ -77,7 +77,7 @@ async def read_path(self, path: str) -> bytes:
Returns:
Contents at key as bytes
"""
- client = self.get_client()
+ client = self.get_async_client()
ret = await client.get(path)
await client.close()
@@ -90,7 +90,7 @@ async def write_path(self, path: str, content: bytes) -> None:
path: Redis key to write to
content: Binary object to write
"""
- client = self.get_client()
+ client = self.get_async_client()
ret = await client.set(path, content)
await client.close()
diff --git a/src/integrations/prefect-redis/prefect_redis/locking.py b/src/integrations/prefect-redis/prefect_redis/locking.py
index ce3ed377d2ba..47429fbddff1 100644
--- a/src/integrations/prefect-redis/prefect_redis/locking.py
+++ b/src/integrations/prefect-redis/prefect_redis/locking.py
@@ -148,7 +148,7 @@ async def await_for_lock(self, key: str, timeout: Optional[float] = None) -> boo
lock = AsyncLock(self.async_client, lock_name)
lock_freed = await lock.acquire(blocking_timeout=timeout)
if lock_freed:
- lock.release()
+ await lock.release()
return lock_freed
def is_locked(self, key: str) -> bool:
diff --git a/src/integrations/prefect-redis/pyproject.toml b/src/integrations/prefect-redis/pyproject.toml
index 682e47025971..6fbc6e2016a6 100644
--- a/src/integrations/prefect-redis/pyproject.toml
+++ b/src/integrations/prefect-redis/pyproject.toml
@@ -22,7 +22,7 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries",
]
-dependencies = ["prefect>=3.0.0rc1", "redis>=5.0.1"]
+dependencies = ["prefect>=3.0.0", "redis>=5.0.1"]
dynamic = ["version"]
[project.optional-dependencies]
@@ -37,7 +37,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-timeout",
"pytest-xdist",
@@ -71,8 +71,7 @@ fail_under = 80
show_missing = true
[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
asyncio_mode = "auto"
timeout = "30"
-env = [
- "PREFECT_TEST_MODE=1",
-]
+env = ["PREFECT_TEST_MODE=1"]
diff --git a/src/integrations/prefect-shell/prefect_shell/commands.py b/src/integrations/prefect-shell/prefect_shell/commands.py
index 40395586f422..50360595730b 100644
--- a/src/integrations/prefect-shell/prefect_shell/commands.py
+++ b/src/integrations/prefect-shell/prefect_shell/commands.py
@@ -227,7 +227,7 @@ class ShellOperation(JobBlock):
_block_type_name = "Shell Operation"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/0b47a017e1b40381de770c17647c49cdf6388d1c-250x250.png" # noqa: E501
- _documentation_url = "https://prefecthq.github.io/prefect-shell/commands/#prefect_shell.commands.ShellOperation" # noqa: E501
+ _documentation_url = "https://docs.prefect.io/integrations/prefect-shell" # noqa
commands: List[str] = Field(
default=..., description="A list of commands to execute sequentially."
diff --git a/src/integrations/prefect-shell/pyproject.toml b/src/integrations/prefect-shell/pyproject.toml
index bd2ae207af8a..7a0a64af6307 100644
--- a/src/integrations/prefect-shell/pyproject.toml
+++ b/src/integrations/prefect-shell/pyproject.toml
@@ -22,7 +22,7 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries",
]
-dependencies = ["prefect>=3.0.0rc1"]
+dependencies = ["prefect>=3.0.0"]
dynamic = ["version"]
[project.optional-dependencies]
@@ -38,7 +38,7 @@ dev = [
"pillow",
"pre-commit",
"pytest-asyncio",
- "pytest",
+ "pytest >= 8.3",
"pytest-env",
"pytest-xdist",
]
@@ -54,7 +54,15 @@ version_file = "prefect_shell/_version.py"
root = "../../.."
tag_regex = "^prefect-shell-(?P