clean up examples

flows/access_run_context.py

@@ -0,0 +1,43 @@
+"""
+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

@@ -5,15 +5,15 @@
 """
 
 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/local-concurrency-with-task-runner.py → flows/local_concurrency_with_task_runner.py

@@ -1,7 +1,10 @@
 """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 +15,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 +30,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 +46,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

@@ -3,9 +3,9 @@
 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 +25,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)
     )