Use Block.aload internally

src/prefect/cli/_prompts.py

@@ -503,7 +503,7 @@ async def prompt_push_custom_docker_image(
             docker_registry_creds_name = f"deployment-{slugify(deployment_config['name'])}-{slugify(deployment_config['work_pool']['name'])}-registry-creds"
             create_new_block = False
             try:
-                await credentials_block.load(docker_registry_creds_name)
+                await credentials_block.aload(docker_registry_creds_name)
                 if not confirm(
                     (
                         "Would you like to use the existing Docker registry credentials"

src/prefect/cli/deploy.py

@@ -1028,7 +1028,7 @@ async def _generate_git_clone_pull_step(
         )
 
         try:
-            await Secret.load(token_secret_block_name)
+            await Secret.aload(token_secret_block_name)
             if not confirm(
                 (
                     "We found an existing token saved for this deployment. Would"

src/prefect/deployments/steps/pull.py

@@ -190,7 +190,7 @@ async def pull_with_block(block_document_name: str, block_type_slug: str):
 
     full_slug = f"{block_type_slug}/{block_document_name}"
     try:
-        block = await Block.load(full_slug)
+        block = await Block.aload(full_slug)
     except Exception:
         deployment_logger.exception("Unable to load block '%s'", full_slug)
         raise

src/prefect/results.py

@@ -129,7 +129,7 @@ async def resolve_result_storage(
     elif isinstance(result_storage, Path):
         storage_block = LocalFileSystem(basepath=str(result_storage))
     elif isinstance(result_storage, str):
-        storage_block = await Block.load(result_storage, client=client)
+        storage_block = await Block.aload(result_storage, client=client)
         storage_block_id = storage_block._block_document_id
         assert storage_block_id is not None, "Loaded storage blocks must have ids"
     elif isinstance(result_storage, UUID):
@@ -168,7 +168,7 @@ async def get_or_create_default_task_scheduling_storage() -> ResultStorage:
     default_block = settings.tasks.scheduling.default_storage_block
 
     if default_block is not None:
-        return await Block.load(default_block)
+        return await Block.aload(default_block)
 
     # otherwise, use the local file system
     basepath = settings.results.local_storage_path

src/prefect/testing/utilities.py

@@ -249,7 +249,7 @@ async def assert_uses_result_storage(
             (
                 storage
                 if isinstance(storage, Block)
-                else await Block.load(storage, client=client)
+                else await Block.aload(storage, client=client)
             ),
         )
     else:
@@ -260,7 +260,7 @@ async def assert_uses_result_storage(
             (
                 storage
                 if isinstance(storage, Block)
-                else await Block.load(storage, client=client)
+                else await Block.aload(storage, client=client)
             ),
         )