Raise type completeness of prefect.server to 100% (#16729)

docs/v3/api-ref/rest-api/server/schema.json

@@ -9355,7 +9355,10 @@
                         "description": "Successful Response",
                         "content": {
                             "application/json": {
-                                "schema": {}
+                                "schema": {
+                                    "type": "object",
+                                    "title": "Response Validate Obj Ui Schemas Validate Post"
+                                }
                             }
                         }
                     },
@@ -9727,7 +9730,10 @@
                         "description": "Successful Response",
                         "content": {
                             "application/json": {
-                                "schema": {}
+                                "schema": {
+                                    "type": "string",
+                                    "title": "Response Hello Hello Get"
+                                }
                             }
                         }
                     },
@@ -22594,6 +22600,9 @@
                 "description": "An ORM representation of task run data."
             },
             "TaskRunCount": {
+                "additionalProperties": {
+                    "type": "integer"
+                },
                 "type": "object"
             },
             "TaskRunCreate": {

src/prefect/server/api/admin.py

@@ -9,7 +9,7 @@
 from prefect.server.database import PrefectDBInterface, provide_database_interface
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(prefix="/admin", tags=["Admin"])
+router: PrefectRouter = PrefectRouter(prefix="/admin", tags=["Admin"])
 
 
 @router.get("/settings")
@@ -37,7 +37,7 @@ async def clear_database(
         description="Pass confirm=True to confirm you want to modify the database.",
     ),
     response: Response = None,  # type: ignore
-):
+) -> None:
     """Clear all database tables without dropping them."""
     if not confirm:
         response.status_code = status.HTTP_400_BAD_REQUEST
@@ -58,7 +58,7 @@ async def drop_database(
         description="Pass confirm=True to confirm you want to modify the database.",
     ),
     response: Response = None,
-):
+) -> None:
     """Drop all database objects."""
     if not confirm:
         response.status_code = status.HTTP_400_BAD_REQUEST
@@ -76,7 +76,7 @@ async def create_database(
         description="Pass confirm=True to confirm you want to modify the database.",
     ),
     response: Response = None,
-):
+) -> None:
     """Create all database objects."""
     if not confirm:
         response.status_code = status.HTTP_400_BAD_REQUEST

src/prefect/server/api/artifacts.py

@@ -14,7 +14,7 @@
 from prefect.server.schemas import actions, core, filters, sorting
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(
+router: PrefectRouter = PrefectRouter(
     prefix="/artifacts",
     tags=["Artifacts"],
 )
@@ -191,7 +191,7 @@ async def update_artifact(
         ..., description="The ID of the artifact to update.", alias="id"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     """
     Update an artifact in the database.
     """
@@ -211,7 +211,7 @@ async def delete_artifact(
         ..., description="The ID of the artifact to delete.", alias="id"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     """
     Delete an artifact from the database.
     """

src/prefect/server/api/automations.py

@@ -27,7 +27,7 @@
     ValidationError as JSONSchemaValidationError,
 )
 
-router = PrefectRouter(
+router: PrefectRouter = PrefectRouter(
     prefix="/automations",
     tags=["Automations"],
     dependencies=[],
@@ -91,7 +91,7 @@ async def update_automation(
     automation: AutomationUpdate,
     automation_id: UUID = Path(..., alias="id"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     # reset any client-provided IDs on the provided triggers
     automation.trigger.reset_ids()
 
@@ -134,7 +134,7 @@ async def patch_automation(
     automation: AutomationPartialUpdate,
     automation_id: UUID = Path(..., alias="id"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     try:
         async with db.session_context(begin_transaction=True) as session:
             updated = await automations_models.update_automation(
@@ -159,7 +159,7 @@ async def patch_automation(
 async def delete_automation(
     automation_id: UUID = Path(..., alias="id"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         deleted = await automations_models.delete_automation(
             session=session,
@@ -228,7 +228,7 @@ async def read_automations_related_to_resource(
 async def delete_automations_owned_by_resource(
     resource_id: str = Path(..., alias="resource_id"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         await automations_models.delete_automations_owned_by_resource(
             session,

src/prefect/server/api/block_capabilities.py

@@ -10,7 +10,9 @@
 from prefect.server.database import PrefectDBInterface, provide_database_interface
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(prefix="/block_capabilities", tags=["Block capabilities"])
+router: PrefectRouter = PrefectRouter(
+    prefix="/block_capabilities", tags=["Block capabilities"]
+)
 
 
 @router.get("/")

src/prefect/server/api/block_documents.py

@@ -12,7 +12,9 @@
 from prefect.server.database import PrefectDBInterface, provide_database_interface
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(prefix="/block_documents", tags=["Block documents"])
+router: PrefectRouter = PrefectRouter(
+    prefix="/block_documents", tags=["Block documents"]
+)
 
 
 @router.post("/", status_code=status.HTTP_201_CREATED)
@@ -124,7 +126,7 @@ async def delete_block_document(
         ..., description="The block document id", alias="id"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         result = await models.block_documents.delete_block_document(
             session=session, block_document_id=block_document_id
@@ -142,7 +144,7 @@ async def update_block_document_data(
         ..., description="The block document id", alias="id"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     try:
         async with db.session_context(begin_transaction=True) as session:
             result = await models.block_documents.update_block_document(

src/prefect/server/api/block_schemas.py

@@ -21,7 +21,7 @@
 from prefect.server.models.block_schemas import MissingBlockTypeException
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(prefix="/block_schemas", tags=["Block schemas"])
+router: PrefectRouter = PrefectRouter(prefix="/block_schemas", tags=["Block schemas"])
 
 
 @router.post("/", status_code=status.HTTP_201_CREATED)
@@ -68,8 +68,8 @@ async def create_block_schema(
 async def delete_block_schema(
     block_schema_id: UUID = Path(..., description="The block schema id", alias="id"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-    api_version=Depends(dependencies.provide_request_api_version),
-):
+    api_version: str = Depends(dependencies.provide_request_api_version),
+) -> None:
     """
     Delete a block schema by id.
     """

src/prefect/server/api/block_types.py

@@ -10,7 +10,7 @@
 from prefect.server.database import PrefectDBInterface, provide_database_interface
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(prefix="/block_types", tags=["Block types"])
+router: PrefectRouter = PrefectRouter(prefix="/block_types", tags=["Block types"])
 
 
 @router.post("/", status_code=status.HTTP_201_CREATED)
@@ -101,7 +101,7 @@ async def update_block_type(
     block_type: schemas.actions.BlockTypeUpdate,
     block_type_id: UUID = Path(..., description="The block type ID", alias="id"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     """
     Update a block type.
     """
@@ -131,7 +131,7 @@ async def update_block_type(
 async def delete_block_type(
     block_type_id: UUID = Path(..., description="The block type ID", alias="id"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         db_block_type = await models.block_types.read_block_type(
             session=session, block_type_id=block_type_id
@@ -204,7 +204,7 @@ async def read_block_document_by_name_for_block_type(
 @router.post("/install_system_block_types")
 async def install_system_block_types(
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     # Don't begin a transaction. _install_protected_system_blocks will manage
     # the transactions.
     async with db.session_context(begin_transaction=False) as session:

src/prefect/server/api/clients.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 from urllib.parse import quote
 from uuid import UUID
 
@@ -19,7 +19,10 @@
 from prefect.server.schemas.responses import DeploymentResponse, OrchestrationResult
 from prefect.types import StrictVariableValue
 
-logger = get_logger(__name__)
+if TYPE_CHECKING:
+    import logging
+
+logger: "logging.Logger" = get_logger(__name__)
 
 
 class BaseClient:

src/prefect/server/api/collections.py

@@ -8,9 +8,11 @@
 
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(prefix="/collections", tags=["Collections"])
+router: PrefectRouter = PrefectRouter(prefix="/collections", tags=["Collections"])
 
-GLOBAL_COLLECTIONS_VIEW_CACHE: TTLCache = TTLCache(maxsize=200, ttl=60 * 10)
+GLOBAL_COLLECTIONS_VIEW_CACHE: TTLCache[str, dict[str, Any]] = TTLCache(
+    maxsize=200, ttl=60 * 10
+)
 
 REGISTRY_VIEWS = (
     "https://raw.githubusercontent.com/PrefectHQ/prefect-collection-registry/main/views"
@@ -43,7 +45,7 @@ async def read_view_content(view: str) -> Dict[str, Any]:
             raise
 
 
-async def get_collection_view(view: str):
+async def get_collection_view(view: str) -> dict[str, Any]:
     try:
         return GLOBAL_COLLECTIONS_VIEW_CACHE[view]
     except KeyError:

src/prefect/server/api/concurrency_limits.py

@@ -17,7 +17,9 @@
 from prefect.server.utilities.server import PrefectRouter
 from prefect.settings import PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS
 
-router = PrefectRouter(prefix="/concurrency_limits", tags=["Concurrency Limits"])
+router: PrefectRouter = PrefectRouter(
+    prefix="/concurrency_limits", tags=["Concurrency Limits"]
+)
 
 
 @router.post("/")
@@ -119,7 +121,7 @@ async def reset_concurrency_limit_by_tag(
         description="Manual override for active concurrency limit slots.",
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         model = await models.concurrency_limits.reset_concurrency_limit_by_tag(
             session=session, tag=tag, slot_override=slot_override
@@ -136,7 +138,7 @@ async def delete_concurrency_limit(
         ..., description="The concurrency limit id", alias="id"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         result = await models.concurrency_limits.delete_concurrency_limit(
             session=session, concurrency_limit_id=concurrency_limit_id
@@ -151,7 +153,7 @@ async def delete_concurrency_limit(
 async def delete_concurrency_limit_by_tag(
     tag: str = Path(..., description="The tag name"),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         result = await models.concurrency_limits.delete_concurrency_limit_by_tag(
             session=session, tag=tag
@@ -263,7 +265,7 @@ async def decrement_concurrency_limits_v1(
         ..., description="The ID of the task run releasing the slot"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     async with db.session_context(begin_transaction=True) as session:
         filtered_limits = (
             await concurrency_limits.filter_concurrency_limits_for_orchestration(

src/prefect/server/api/concurrency_limits_v2.py

@@ -11,7 +11,9 @@
 from prefect.server.utilities.schemas import PrefectBaseModel
 from prefect.server.utilities.server import PrefectRouter
 
-router = PrefectRouter(prefix="/v2/concurrency_limits", tags=["Concurrency Limits V2"])
+router: PrefectRouter = PrefectRouter(
+    prefix="/v2/concurrency_limits", tags=["Concurrency Limits V2"]
+)
 
 
 @router.post("/", status_code=status.HTTP_201_CREATED)
@@ -85,7 +87,7 @@ async def update_concurrency_limit_v2(
         ..., description="The ID or name of the concurrency limit", alias="id_or_name"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     if isinstance(id_or_name, str):  # TODO: this seems like it shouldn't be necessary
         try:
             id_or_name = UUID(id_or_name)
@@ -115,7 +117,7 @@ async def delete_concurrency_limit_v2(
         ..., description="The ID or name of the concurrency limit", alias="id_or_name"
     ),
     db: PrefectDBInterface = Depends(provide_database_interface),
-):
+) -> None:
     if isinstance(id_or_name, str):  # TODO: this seems like it shouldn't be necessary
         try:
             id_or_name = UUID(id_or_name)

src/prefect/server/api/csrf_token.py

@@ -1,3 +1,5 @@
+from typing import TYPE_CHECKING
+
 from fastapi import Depends, Query, status
 from starlette.exceptions import HTTPException
 
@@ -7,9 +9,12 @@
 from prefect.server.utilities.server import PrefectRouter
 from prefect.settings import PREFECT_SERVER_CSRF_PROTECTION_ENABLED
 
-logger = get_logger("server.api")
+if TYPE_CHECKING:
+    import logging
+
+logger: "logging.Logger" = get_logger("server.api")
 
-router = PrefectRouter(prefix="/csrf-token")
+router: PrefectRouter = PrefectRouter(prefix="/csrf-token")
 
 
 @router.get("")