We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Deprecation warning missing from some blocks 
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the Prefect documentation for this issue.
- [X] I checked that this issue is related to Prefect and not one of its dependencies.

### Bug summary

Several blocks show a deprecation warning in the UI. Jowever, several blocks that I believe should have a deprecation warning are missing it:

1. ECS Task
1. Vertex AI Custom Training Job
1. GCP Cloud Run Job
1. Azure Container Instance Job
1. Docker Registry

Example:
![Screenshot 2024-05-08 at 4 23 41 PM](https://github.com/PrefectHQ/prefect/assets/7703961/6461e6c4-53ed-4d30-9c3e-a1e03cc1fc65)


### Reproduction

```python3
See example above.
```


### Error

_No response_

### Versions

```Text
Many of my integration libraries and their block types are updated as of Prefect Cloud May 8, 2024.
```

cc: @desertaxle 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/server/models/block_registration.py`
Content:
```
1 import json
2 from pathlib import Path
3 from typing import cast
4 
5 import sqlalchemy as sa
6 
7 from prefect.blocks.core import Block
8 from prefect.blocks.system import JSON, DateTime, Secret
9 from prefect.blocks.webhook import Webhook
10 from prefect.filesystems import LocalFileSystem
11 from prefect.logging import get_logger
12 from prefect.server import models, schemas
13 
14 logger = get_logger("server")
15 
16 COLLECTIONS_BLOCKS_DATA_PATH = (
17     Path(__file__).parent.parent / "collection_blocks_data.json"
18 )
19 
20 
21 async def _install_protected_system_blocks(session):
22     """Install block types that the system expects to be present"""
23 
24     for block in [
25         Webhook,
26         JSON,
27         DateTime,
28         Secret,
29         LocalFileSystem,
30     ]:
31         block = cast(Block, block)
32         async with session.begin():
33             block_type = block._to_block_type()
34 
35             server_block_type = schemas.core.BlockType.model_validate(
36                 block_type.model_dump()
37             )
38             block_type.is_protected = True
39             server_block_type.is_protected = True
40 
41             block_type = await models.block_types.create_block_type(
42                 session=session, block_type=server_block_type, override=True
43             )
44             await models.block_schemas.create_block_schema(
45                 session=session,
46                 block_schema=block._to_block_schema(block_type_id=block_type.id),
47                 override=True,
48             )
49 
50 
51 async def register_block_schema(
52     session: sa.orm.Session,
53     block_schema: schemas.core.BlockSchema,
54 ):
55     """
56     Stores the provided block schema in the Prefect REST API database.
57 
58     If a block schema with a matching checksum and version is already saved,
59     then the ID of the existing block schema will be returned.
60 
61     Args:
62         session: A database session.
63         block_schema: A block schema object.
64 
65     Returns:
66         The ID of the registered block schema.
67     """
68 
69     from prefect.server.models.block_schemas import (
70         create_block_schema,
71         read_block_schema_by_checksum,
72     )
73 
74     existing_block_schema = await read_block_schema_by_checksum(
75         session=session, checksum=block_schema.checksum, version=block_schema.version
76     )
77     if existing_block_schema is None:
78         block_schema = await create_block_schema(
79             session=session,
80             block_schema=block_schema,
81         )
82         return block_schema.id
83     else:
84         return existing_block_schema.id
85 
86 
87 async def register_block_type(
88     session: sa.orm.Session,
89     block_type: schemas.core.BlockType,
90 ):
91     """
92     Stores the provided block type in the Prefect REST API database.
93 
94     If a block type with a matching slug is already saved, then the block type
95     will be updated to match the passed in block type.
96 
97     Args:
98         session: A database session.
99         block_type: A block type object.
100 
101     Returns:
102         The ID of the registered block type.
103     """
104     from prefect.server.models.block_types import (
105         create_block_type,
106         read_block_type_by_slug,
107         update_block_type,
108     )
109 
110     existing_block_type = await read_block_type_by_slug(
111         session=session,
112         block_type_slug=block_type.slug,
113     )
114     if existing_block_type is None:
115         block_type = await create_block_type(
116             session=session,
117             block_type=block_type,
118         )
119         return block_type.id
120     else:
121         await update_block_type(
122             session=session,
123             block_type_id=existing_block_type.id,
124             block_type=block_type,
125         )
126         return existing_block_type.id
127 
128 
129 async def _load_collection_blocks_data():
130     """Loads blocks data for whitelisted collections."""
131     import anyio
132 
133     async with await anyio.open_file(COLLECTIONS_BLOCKS_DATA_PATH, "r") as f:
134         return json.loads(await f.read())
135 
136 
137 async def _register_registry_blocks(session: sa.orm.Session):
138     """Registers block from the client block registry."""
139     from prefect.blocks.core import Block
140     from prefect.utilities.dispatch import get_registry_for_type
141 
142     block_registry = get_registry_for_type(Block) or {}
143 
144     for block_class in block_registry.values():
145         # each block schema gets its own transaction
146         async with session.begin():
147             block_type_id = await register_block_type(
148                 session=session,
149                 block_type=block_class._to_block_type(),
150             )
151             await register_block_schema(
152                 session=session,
153                 block_schema=block_class._to_block_schema(block_type_id=block_type_id),
154             )
155 
156 
157 async def _register_collection_blocks(session: sa.orm.Session):
158     """Registers blocks from whitelisted collections."""
159     collections_blocks_data = await _load_collection_blocks_data()
160 
161     block_types = [
162         block_type
163         for collection in collections_blocks_data["collections"].values()
164         for block_type in collection["block_types"].values()
165     ]
166     for block_type in block_types:
167         # each block schema gets its own transaction
168         async with session.begin():
169             block_schemas = block_type.pop("block_schemas", [])
170             block_type_id = await register_block_type(
171                 session=session,
172                 block_type=schemas.core.BlockType.model_validate(block_type),
173             )
174             for block_schema in block_schemas:
175                 await register_block_schema(
176                     session=session,
177                     block_schema=schemas.core.BlockSchema.model_validate(
178                         {**block_schema, "block_type_id": block_type_id}
179                     ),
180                 )
181 
182 
183 async def run_block_auto_registration(session: sa.orm.Session):
184     """
185     Registers all blocks in the client block registry and any blocks from Prefect
186     Collections that are configured for auto-registration.
187 
188     Args:
189         session: A database session.
190     """
191     await _install_protected_system_blocks(session)
192     await _register_registry_blocks(session)
193     await _register_collection_blocks(session=session)
194
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/server/models/block_registration.py b/src/prefect/server/models/block_registration.py
--- a/src/prefect/server/models/block_registration.py
+++ b/src/prefect/server/models/block_registration.py
@@ -163,20 +163,35 @@
         for collection in collections_blocks_data["collections"].values()
         for block_type in collection["block_types"].values()
     ]
-    for block_type in block_types:
-        # each block schema gets its own transaction
-        async with session.begin():
-            block_schemas = block_type.pop("block_schemas", [])
+
+    # due to schema reference dependencies, we need to register all block types first
+    # and then register all block schemas
+    block_schemas: dict[str, dict] = {}
+
+    async with session.begin():
+        for block_type in block_types:
+            block_schema = block_type.pop("block_schema", None)
+            if not block_schema:
+                raise RuntimeError(
+                    f"Block schema not found for block type {block_type.get('slug')!r}"
+                )
             block_type_id = await register_block_type(
                 session=session,
                 block_type=schemas.core.BlockType.model_validate(block_type),
             )
-            for block_schema in block_schemas:
+            block_schema["block_type_id"] = block_type_id
+            block_schemas[block_type["slug"]] = block_schema
+
+    async with session.begin():
+        for block_type_slug, block_schema in block_schemas.items():
+            try:
                 await register_block_schema(
                     session=session,
-                    block_schema=schemas.core.BlockSchema.model_validate(
-                        {**block_schema, "block_type_id": block_type_id}
-                    ),
+                    block_schema=schemas.core.BlockSchema.model_validate(block_schema),
+                )
+            except Exception:
+                logger.exception(
+                    f"Failed to register block schema for block type {block_type_slug}"
                 )

{"golden_diff": "diff --git a/src/prefect/server/models/block_registration.py b/src/prefect/server/models/block_registration.py\n--- a/src/prefect/server/models/block_registration.py\n+++ b/src/prefect/server/models/block_registration.py\n@@ -163,20 +163,35 @@\n         for collection in collections_blocks_data[\"collections\"].values()\n         for block_type in collection[\"block_types\"].values()\n     ]\n-    for block_type in block_types:\n-        # each block schema gets its own transaction\n-        async with session.begin():\n-            block_schemas = block_type.pop(\"block_schemas\", [])\n+\n+    # due to schema reference dependencies, we need to register all block types first\n+    # and then register all block schemas\n+    block_schemas: dict[str, dict] = {}\n+\n+    async with session.begin():\n+        for block_type in block_types:\n+            block_schema = block_type.pop(\"block_schema\", None)\n+            if not block_schema:\n+                raise RuntimeError(\n+                    f\"Block schema not found for block type {block_type.get('slug')!r}\"\n+                )\n             block_type_id = await register_block_type(\n                 session=session,\n                 block_type=schemas.core.BlockType.model_validate(block_type),\n             )\n-            for block_schema in block_schemas:\n+            block_schema[\"block_type_id\"] = block_type_id\n+            block_schemas[block_type[\"slug\"]] = block_schema\n+\n+    async with session.begin():\n+        for block_type_slug, block_schema in block_schemas.items():\n+            try:\n                 await register_block_schema(\n                     session=session,\n-                    block_schema=schemas.core.BlockSchema.model_validate(\n-                        {**block_schema, \"block_type_id\": block_type_id}\n-                    ),\n+                    block_schema=schemas.core.BlockSchema.model_validate(block_schema),\n+                )\n+            except Exception:\n+                logger.exception(\n+                    f\"Failed to register block schema for block type {block_type_slug}\"\n                 )\n", "issue": "Deprecation warning missing from some blocks \n### First check\r\n\r\n- [X] I added a descriptive title to this issue.\r\n- [X] I used the GitHub search to find a similar issue and didn't find it.\r\n- [X] I searched the Prefect documentation for this issue.\r\n- [X] I checked that this issue is related to Prefect and not one of its dependencies.\r\n\r\n### Bug summary\r\n\r\nSeveral blocks show a deprecation warning in the UI. Jowever, several blocks that I believe should have a deprecation warning are missing it:\r\n\r\n1. ECS Task\r\n1. Vertex AI Custom Training Job\r\n1. GCP Cloud Run Job\r\n1. Azure Container Instance Job\r\n1. Docker Registry\r\n\r\nExample:\r\n![Screenshot 2024-05-08 at 4 23 41\u202fPM](https://github.com/PrefectHQ/prefect/assets/7703961/6461e6c4-53ed-4d30-9c3e-a1e03cc1fc65)\r\n\r\n\r\n### Reproduction\r\n\r\n```python3\r\nSee example above.\r\n```\r\n\r\n\r\n### Error\r\n\r\n_No response_\r\n\r\n### Versions\r\n\r\n```Text\r\nMany of my integration libraries and their block types are updated as of Prefect Cloud May 8, 2024.\r\n```\r\n\r\ncc: @desertaxle \n", "before_files": [{"content": "import json\nfrom pathlib import Path\nfrom typing import cast\n\nimport sqlalchemy as sa\n\nfrom prefect.blocks.core import Block\nfrom prefect.blocks.system import JSON, DateTime, Secret\nfrom prefect.blocks.webhook import Webhook\nfrom prefect.filesystems import LocalFileSystem\nfrom prefect.logging import get_logger\nfrom prefect.server import models, schemas\n\nlogger = get_logger(\"server\")\n\nCOLLECTIONS_BLOCKS_DATA_PATH = (\n    Path(__file__).parent.parent / \"collection_blocks_data.json\"\n)\n\n\nasync def _install_protected_system_blocks(session):\n    \"\"\"Install block types that the system expects to be present\"\"\"\n\n    for block in [\n        Webhook,\n        JSON,\n        DateTime,\n        Secret,\n        LocalFileSystem,\n    ]:\n        block = cast(Block, block)\n        async with session.begin():\n            block_type = block._to_block_type()\n\n            server_block_type = schemas.core.BlockType.model_validate(\n                block_type.model_dump()\n            )\n            block_type.is_protected = True\n            server_block_type.is_protected = True\n\n            block_type = await models.block_types.create_block_type(\n                session=session, block_type=server_block_type, override=True\n            )\n            await models.block_schemas.create_block_schema(\n                session=session,\n                block_schema=block._to_block_schema(block_type_id=block_type.id),\n                override=True,\n            )\n\n\nasync def register_block_schema(\n    session: sa.orm.Session,\n    block_schema: schemas.core.BlockSchema,\n):\n    \"\"\"\n    Stores the provided block schema in the Prefect REST API database.\n\n    If a block schema with a matching checksum and version is already saved,\n    then the ID of the existing block schema will be returned.\n\n    Args:\n        session: A database session.\n        block_schema: A block schema object.\n\n    Returns:\n        The ID of the registered block schema.\n    \"\"\"\n\n    from prefect.server.models.block_schemas import (\n        create_block_schema,\n        read_block_schema_by_checksum,\n    )\n\n    existing_block_schema = await read_block_schema_by_checksum(\n        session=session, checksum=block_schema.checksum, version=block_schema.version\n    )\n    if existing_block_schema is None:\n        block_schema = await create_block_schema(\n            session=session,\n            block_schema=block_schema,\n        )\n        return block_schema.id\n    else:\n        return existing_block_schema.id\n\n\nasync def register_block_type(\n    session: sa.orm.Session,\n    block_type: schemas.core.BlockType,\n):\n    \"\"\"\n    Stores the provided block type in the Prefect REST API database.\n\n    If a block type with a matching slug is already saved, then the block type\n    will be updated to match the passed in block type.\n\n    Args:\n        session: A database session.\n        block_type: A block type object.\n\n    Returns:\n        The ID of the registered block type.\n    \"\"\"\n    from prefect.server.models.block_types import (\n        create_block_type,\n        read_block_type_by_slug,\n        update_block_type,\n    )\n\n    existing_block_type = await read_block_type_by_slug(\n        session=session,\n        block_type_slug=block_type.slug,\n    )\n    if existing_block_type is None:\n        block_type = await create_block_type(\n            session=session,\n            block_type=block_type,\n        )\n        return block_type.id\n    else:\n        await update_block_type(\n            session=session,\n            block_type_id=existing_block_type.id,\n            block_type=block_type,\n        )\n        return existing_block_type.id\n\n\nasync def _load_collection_blocks_data():\n    \"\"\"Loads blocks data for whitelisted collections.\"\"\"\n    import anyio\n\n    async with await anyio.open_file(COLLECTIONS_BLOCKS_DATA_PATH, \"r\") as f:\n        return json.loads(await f.read())\n\n\nasync def _register_registry_blocks(session: sa.orm.Session):\n    \"\"\"Registers block from the client block registry.\"\"\"\n    from prefect.blocks.core import Block\n    from prefect.utilities.dispatch import get_registry_for_type\n\n    block_registry = get_registry_for_type(Block) or {}\n\n    for block_class in block_registry.values():\n        # each block schema gets its own transaction\n        async with session.begin():\n            block_type_id = await register_block_type(\n                session=session,\n                block_type=block_class._to_block_type(),\n            )\n            await register_block_schema(\n                session=session,\n                block_schema=block_class._to_block_schema(block_type_id=block_type_id),\n            )\n\n\nasync def _register_collection_blocks(session: sa.orm.Session):\n    \"\"\"Registers blocks from whitelisted collections.\"\"\"\n    collections_blocks_data = await _load_collection_blocks_data()\n\n    block_types = [\n        block_type\n        for collection in collections_blocks_data[\"collections\"].values()\n        for block_type in collection[\"block_types\"].values()\n    ]\n    for block_type in block_types:\n        # each block schema gets its own transaction\n        async with session.begin():\n            block_schemas = block_type.pop(\"block_schemas\", [])\n            block_type_id = await register_block_type(\n                session=session,\n                block_type=schemas.core.BlockType.model_validate(block_type),\n            )\n            for block_schema in block_schemas:\n                await register_block_schema(\n                    session=session,\n                    block_schema=schemas.core.BlockSchema.model_validate(\n                        {**block_schema, \"block_type_id\": block_type_id}\n                    ),\n                )\n\n\nasync def run_block_auto_registration(session: sa.orm.Session):\n    \"\"\"\n    Registers all blocks in the client block registry and any blocks from Prefect\n    Collections that are configured for auto-registration.\n\n    Args:\n        session: A database session.\n    \"\"\"\n    await _install_protected_system_blocks(session)\n    await _register_registry_blocks(session)\n    await _register_collection_blocks(session=session)\n", "path": "src/prefect/server/models/block_registration.py"}], "after_files": [{"content": "import json\nfrom pathlib import Path\nfrom typing import cast\n\nimport sqlalchemy as sa\n\nfrom prefect.blocks.core import Block\nfrom prefect.blocks.system import JSON, DateTime, Secret\nfrom prefect.blocks.webhook import Webhook\nfrom prefect.filesystems import LocalFileSystem\nfrom prefect.logging import get_logger\nfrom prefect.server import models, schemas\n\nlogger = get_logger(\"server\")\n\nCOLLECTIONS_BLOCKS_DATA_PATH = (\n    Path(__file__).parent.parent / \"collection_blocks_data.json\"\n)\n\n\nasync def _install_protected_system_blocks(session):\n    \"\"\"Install block types that the system expects to be present\"\"\"\n\n    for block in [\n        Webhook,\n        JSON,\n        DateTime,\n        Secret,\n        LocalFileSystem,\n    ]:\n        block = cast(Block, block)\n        async with session.begin():\n            block_type = block._to_block_type()\n\n            server_block_type = schemas.core.BlockType.model_validate(\n                block_type.model_dump()\n            )\n            block_type.is_protected = True\n            server_block_type.is_protected = True\n\n            block_type = await models.block_types.create_block_type(\n                session=session, block_type=server_block_type, override=True\n            )\n            await models.block_schemas.create_block_schema(\n                session=session,\n                block_schema=block._to_block_schema(block_type_id=block_type.id),\n                override=True,\n            )\n\n\nasync def register_block_schema(\n    session: sa.orm.Session,\n    block_schema: schemas.core.BlockSchema,\n):\n    \"\"\"\n    Stores the provided block schema in the Prefect REST API database.\n\n    If a block schema with a matching checksum and version is already saved,\n    then the ID of the existing block schema will be returned.\n\n    Args:\n        session: A database session.\n        block_schema: A block schema object.\n\n    Returns:\n        The ID of the registered block schema.\n    \"\"\"\n\n    from prefect.server.models.block_schemas import (\n        create_block_schema,\n        read_block_schema_by_checksum,\n    )\n\n    existing_block_schema = await read_block_schema_by_checksum(\n        session=session, checksum=block_schema.checksum, version=block_schema.version\n    )\n    if existing_block_schema is None:\n        block_schema = await create_block_schema(\n            session=session,\n            block_schema=block_schema,\n        )\n        return block_schema.id\n    else:\n        return existing_block_schema.id\n\n\nasync def register_block_type(\n    session: sa.orm.Session,\n    block_type: schemas.core.BlockType,\n):\n    \"\"\"\n    Stores the provided block type in the Prefect REST API database.\n\n    If a block type with a matching slug is already saved, then the block type\n    will be updated to match the passed in block type.\n\n    Args:\n        session: A database session.\n        block_type: A block type object.\n\n    Returns:\n        The ID of the registered block type.\n    \"\"\"\n    from prefect.server.models.block_types import (\n        create_block_type,\n        read_block_type_by_slug,\n        update_block_type,\n    )\n\n    existing_block_type = await read_block_type_by_slug(\n        session=session,\n        block_type_slug=block_type.slug,\n    )\n    if existing_block_type is None:\n        block_type = await create_block_type(\n            session=session,\n            block_type=block_type,\n        )\n        return block_type.id\n    else:\n        await update_block_type(\n            session=session,\n            block_type_id=existing_block_type.id,\n            block_type=block_type,\n        )\n        return existing_block_type.id\n\n\nasync def _load_collection_blocks_data():\n    \"\"\"Loads blocks data for whitelisted collections.\"\"\"\n    import anyio\n\n    async with await anyio.open_file(COLLECTIONS_BLOCKS_DATA_PATH, \"r\") as f:\n        return json.loads(await f.read())\n\n\nasync def _register_registry_blocks(session: sa.orm.Session):\n    \"\"\"Registers block from the client block registry.\"\"\"\n    from prefect.blocks.core import Block\n    from prefect.utilities.dispatch import get_registry_for_type\n\n    block_registry = get_registry_for_type(Block) or {}\n\n    for block_class in block_registry.values():\n        # each block schema gets its own transaction\n        async with session.begin():\n            block_type_id = await register_block_type(\n                session=session,\n                block_type=block_class._to_block_type(),\n            )\n            await register_block_schema(\n                session=session,\n                block_schema=block_class._to_block_schema(block_type_id=block_type_id),\n            )\n\n\nasync def _register_collection_blocks(session: sa.orm.Session):\n    \"\"\"Registers blocks from whitelisted collections.\"\"\"\n    collections_blocks_data = await _load_collection_blocks_data()\n\n    block_types = [\n        block_type\n        for collection in collections_blocks_data[\"collections\"].values()\n        for block_type in collection[\"block_types\"].values()\n    ]\n\n    # due to schema reference dependencies, we need to register all block types first\n    # and then register all block schemas\n    block_schemas: dict[str, dict] = {}\n\n    async with session.begin():\n        for block_type in block_types:\n            block_schema = block_type.pop(\"block_schema\", None)\n            if not block_schema:\n                raise RuntimeError(\n                    f\"Block schema not found for block type {block_type.get('slug')!r}\"\n                )\n            block_type_id = await register_block_type(\n                session=session,\n                block_type=schemas.core.BlockType.model_validate(block_type),\n            )\n            block_schema[\"block_type_id\"] = block_type_id\n            block_schemas[block_type[\"slug\"]] = block_schema\n\n    async with session.begin():\n        for block_type_slug, block_schema in block_schemas.items():\n            try:\n                await register_block_schema(\n                    session=session,\n                    block_schema=schemas.core.BlockSchema.model_validate(block_schema),\n                )\n            except Exception:\n                logger.exception(\n                    f\"Failed to register block schema for block type {block_type_slug}\"\n                )\n\n\nasync def run_block_auto_registration(session: sa.orm.Session):\n    \"\"\"\n    Registers all blocks in the client block registry and any blocks from Prefect\n    Collections that are configured for auto-registration.\n\n    Args:\n        session: A database session.\n    \"\"\"\n    await _install_protected_system_blocks(session)\n    await _register_registry_blocks(session)\n    await _register_collection_blocks(session=session)\n", "path": "src/prefect/server/models/block_registration.py"}]}