inline script deps

.gitignore

@@ -160,3 +160,7 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+# uv stuff
+.python-version
+uv.lock

README.md

@@ -2,6 +2,23 @@
 
 These are examples that show how to use Prefect. They are intended to be simple starting points rather than complete solutions. We hope that you will find them useful!
 
+> [!IMPORTANT]
+> This repository uses [uv](https://docs.astral.sh/uv/) for python environment management.
+
+If you've cloned the repository, you can run any of the examples with `uv run flows/<example_name>.py`, for example:
+
+```bash
+uv run flows/hello_world.py
+```
+
+If you haven't, you can point `uv run` at the url of the python file, for example:
+
+```bash
+uv run https://github.com/PrefectHQ/examples/blob/main/flows/hello_world.py
+```
+
+
+
 ## Development
 
-This repository uses [uv](https://docs.astral.sh/uv/) for dependency management. To set up pre-commit checks, run `uvx pre-commit`.
+To set up pre-commit checks, run `uvx pre-commit`.

flows/access_run_context.py

@@ -0,0 +1,47 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
+"""
+Examples of accessing Prefect's runtime context to get information about the current flow run:
+- List the fields in the context
+- Access a few specific fields
+- Get the URL for the flow run
+"""
+
+import prefect.runtime.flow_run
+import prefect.runtime.task_run
+from prefect import flow, task
+
+
+@flow(log_prints=True)
+def example_flow():
+    # let's see what's available in the flow run runtime context
+    for field in prefect.runtime.flow_run.__all__:
+        print(f"{field}: {getattr(prefect.runtime.flow_run, field)}")
+
+    # The runtime module supports dot notation for accessing attributes
+    # for example, let's grab the flow run id
+    print(f"\nThis flow run has id '{prefect.runtime.flow_run.id}'")
+
+    # We can also get the dashboard URL for the flow run
+    print(f"\n See this flow run in the UI: '{prefect.runtime.flow_run.ui_url}'")
+
+
+@task(log_prints=True)
+def example_task():
+    # let's see what's available in the task run runtime context
+    for field in prefect.runtime.task_run.__all__:
+        print(f"{field}: {getattr(prefect.runtime.task_run, field)}")
+
+
+if __name__ == "__main__":
+    # Run the example flow from above
+    example_flow()
+
+    # Run the example task from above
+    example_task()
+
+    # Outside of a run context, `prefect.runtime` will return empty values
+    assert prefect.runtime.flow_run.id is None
+    assert prefect.runtime.task_run.id is None

flows/force-out-of-memory.py → flows/force_out_of_memory.py

@@ -1,19 +1,23 @@
+# /// script
+# dependencies = ["prefect", "psutil"]
+# ///
+
 """
 This script is used to force an out-of-memory error in a Prefect flow.
 
 Use it to reproduce and diagnose memory issues.
 """
 
 import os
-import psutil
+from logging import Logger, LoggerAdapter
 
+import psutil
 from prefect import flow, get_run_logger
 
-MEGABYTE = 10**6
+MEGABYTE: int = 10**6
 
 
-def log_memory_usage(process, logger):
-    logger = get_run_logger()
+def log_memory_usage(process: psutil.Process, logger: Logger | LoggerAdapter) -> None:
     mem_stats = psutil.virtual_memory()
     mem_process = process.memory_info().rss / MEGABYTE
     logger.info(

flows/hello-world.py → flows/hello_world.py

@@ -1,3 +1,7 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
 """
 A simple flow that says hello.
 """

flows/local-concurrency-with-async.py → flows/local_concurrency_with_async.py

@@ -1,3 +1,7 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
 """Implement extract flow using Python native async concurrency"""
 
 import asyncio

flows/local-concurrency-with-task-runner.py → flows/local_concurrency_with_task_runner.py

@@ -1,7 +1,14 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
 """Implements a data extract flow using task runner concurrency"""
 
+from typing import Any
+
 import httpx
 from prefect import flow, get_run_logger, tags, task, unmapped
+from prefect.futures import as_completed
 from prefect.task_runners import ThreadPoolTaskRunner
 
 BASE_URL = "https://dev.to/api"
@@ -12,7 +19,7 @@
     retries=3,
     retry_delay_seconds=[10, 30, 60],
 )
-def fetch_url(url: str, params: dict | None = None) -> dict:
+def fetch_url(url: str, params: dict | None = None) -> dict[str, Any]:
     """Generic task for fetching a URL"""
     get_run_logger().info(f"Fetching {url}")
     response = httpx.get(url, params=params)
@@ -27,11 +34,10 @@ def list_articles(pages: int, per_page: int = 10) -> list[str]:
     _pages = fetch_url.map(
         unmapped(f"{BASE_URL}/articles"),
         [{"page": page, "per_page": per_page} for page in range(1, pages + 1)],
-    )
-    pages = [_page.result() for _page in _pages]
+    ).result()
 
     return [
-        f"{BASE_URL}/articles/{article['id']}" for page in pages for article in page
+        f"{BASE_URL}/articles/{article['id']}" for page in _pages for article in page
     ]
 
 
@@ -44,7 +50,7 @@ def extract(pages: int) -> None:
 
     # Log the title of each article as they become ready
     # alternatively, _articles.wait() will wait for all articles
-    for _article in _articles:
+    for _article in as_completed(_articles):
         get_run_logger().info(_article.result()["title"])
 
 

flows/return-custom-state.py → flows/return_custom_state.py

@@ -1,11 +1,15 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
 """
 Demonstrates directly returning a state. In this case, an AwaitingRetry state
 which causes a task to be retried after a delay.
 """
 
-from datetime import datetime, timedelta, UTC
+from datetime import datetime, timedelta, timezone
 
-from prefect import task, get_run_logger
+from prefect import get_run_logger, task
 from prefect.context import get_run_context
 from prefect.states import AwaitingRetry
 
@@ -25,7 +29,7 @@ def main():
 
     # Schedule the task to run again in {retry_after} seconds
     return AwaitingRetry(
-        scheduled_time=datetime.now(UTC) + timedelta(seconds=retry_after)
+        scheduled_time=datetime.now(timezone.utc) + timedelta(seconds=retry_after)
     )
 
 

flows/update-flow-run-state-from-hook.py → flows/update_flow_run_state_from_hook.py

@@ -1,3 +1,7 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
 """
 Demonstrates using the client to update flow run state from a hook.
 """

flows/update-flow-run-tags.py → flows/update_flow_run_tags.py

@@ -1,8 +1,12 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
 """
 Demonstrates using the client to update flow run tags.
 """
 
-from prefect import flow, get_run_logger, get_client, tags
+from prefect import flow, get_client, get_run_logger, tags
 from prefect.context import get_run_context
 
 

flows/whoami.py

@@ -1,3 +1,7 @@
+# /// script
+# dependencies = ["prefect"]
+# ///
+
 """
 This flow will log information about the current environment. Use it to
 diagnose issues with your environment, especially when deploying to
@@ -18,10 +22,10 @@
 ```
 """
 
-import sys
 import os
 import platform
 import socket
+import sys
 
 import prefect
 

pyproject.toml

@@ -4,11 +4,6 @@ version = "0.1.0"
 description = "Examples of how to use Prefect"
 readme = "README.md"
 requires-python = ">=3.9"
-dependencies = [
-    "prefect-docker>=0.6.1",
-    "prefect>=3.1.2",
-    "psutil>=6.1.0",
-]
 
 [project.urls]
 Code = "https://github.com/PrefectHQ/examples"