diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index a6332abeac16..db8993868510 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -477,7 +477,7 @@ See the [flows documentation](https://docs.prefect.io/latest/concepts/flows/#vis
 ### Documentation
 - Elevate Guides in navigation and remove migration guide — https://github.com/PrefectHQ/prefect/pull/10361
 - Update notes about community support — https://github.com/PrefectHQ/prefect/pull/10322
-- Update concepts page to clean up table and remove unneccessary header — https://github.com/PrefectHQ/prefect/pull/10374
+- Update concepts page to clean up table and remove unnecessary header — https://github.com/PrefectHQ/prefect/pull/10374
 - Improve headings on deployments concept page — https://github.com/PrefectHQ/prefect/pull/10366
 - Update the storage guide for Bitbucket to add `x-token-auth` — https://github.com/PrefectHQ/prefect/pull/10379
 - Add Planetary Computer collection — https://github.com/PrefectHQ/prefect/pull/10387
@@ -1285,7 +1285,7 @@ See the following pull requests for implementation details:
 
 ### Fixes
 - Fix bug where `SIGTERM` was not properly captured as a flow run crash for flow runs created by a deployment — https://github.com/PrefectHQ/prefect/pull/9543
-- Fix deadlock when logging is overriden from an asynchronous context — https://github.com/PrefectHQ/prefect/pull/9602
+- Fix deadlock when logging is overridden from an asynchronous context — https://github.com/PrefectHQ/prefect/pull/9602
 - Fix orchestration race conditions by adding lock for update to flow run state transitions — https://github.com/PrefectHQ/prefect/pull/9590
 - Fix date range filter on flow runs page — https://github.com/PrefectHQ/prefect/pull/9636
 - Fix bug where ephemeral server raised exceptions client-side — https://github.com/PrefectHQ/prefect/pull/9637
@@ -1417,7 +1417,7 @@ See the [pull request](https://github.com/PrefectHQ/prefect/pull/9418) for imple
 ### Enhancements
 - Prevent unnecessarily verbose logs by updating `log_prints` to ignore prints where a custom `file` is used — https://github.com/PrefectHQ/prefect/pull/9358
 - Create a process work pool by default when a new worker is started with a new work pool name and no type — https://github.com/PrefectHQ/prefect/pull/9326
-- Add support for asyncronous project steps — https://github.com/PrefectHQ/prefect/pull/9388
+- Add support for asynchronous project steps — https://github.com/PrefectHQ/prefect/pull/9388
 - Update `critical_service_loop` to retry on all 5XX HTTP status codes — https://github.com/PrefectHQ/prefect/pull/9400
 - Add backoff on failure to agent critical loop services — https://github.com/PrefectHQ/prefect/pull/9402
 - Add print statement to `git pull` to isolate issues between clone and execution — https://github.com/PrefectHQ/prefect/pull/9328
@@ -2246,7 +2246,7 @@ if __name__ == "__main__":
 ## Release 2.8.2
 
 ### Fixes
-- Reenable plugin loading in `prefect` module init — https://github.com/PrefectHQ/prefect/pull/8569
+- Re-enable plugin loading in `prefect` module init — https://github.com/PrefectHQ/prefect/pull/8569
 
 ### Documentation
 - Fix logging format override example — https://github.com/PrefectHQ/prefect/pull/8565
@@ -2411,7 +2411,7 @@ To learn more about work pools, check out the [docs](https://docs.prefect.io/con
 ### Fixes
 - Fix `prefect dev start` command — https://github.com/PrefectHQ/prefect/pull/8176
 - Fix display of long log messages when in the UI — https://github.com/PrefectHQ/prefect/pull/8449
-- Update `get_run_logger` to accomodate returning `logging.LoggerAdapter` — https://github.com/PrefectHQ/prefect/pull/8422
+- Update `get_run_logger` to accommodate returning `logging.LoggerAdapter` — https://github.com/PrefectHQ/prefect/pull/8422
 - Restore Prefect wrapper around HTTP errors for nicer error messages — https://github.com/PrefectHQ/prefect/pull/8391
 - Fix display of work pool flow run filter in the UI — https://github.com/PrefectHQ/prefect/pull/8453
 
@@ -2528,7 +2528,7 @@ Updated profile 'default'.
 - Update Docker images to update preinstalled packages on build — https://github.com/PrefectHQ/prefect/pull/8288
 - Add PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW to allow loggers to be used outside of flows — https://github.com/PrefectHQ/prefect/pull/8311
 - Display Runs before Deployments on flow pages — https://github.com/PrefectHQ/prefect/pull/8386
-- Clearify output CLI message when switching profiles — https://github.com/PrefectHQ/prefect/pull/8383
+- Clarify output CLI message when switching profiles — https://github.com/PrefectHQ/prefect/pull/8383
 
 ### Fixes
 - Fix bug preventing agents from properly updating Cancelling runs to a Cancelled state — https://github.com/PrefectHQ/prefect/pull/8315
@@ -3994,13 +3994,13 @@ Blocks are a convenient way to secure store and retrieve configuration. Now, ret
 
 ## 2.0b14
 
-### Retreive the state of your tasks or flows with the `return_state` kwarg
+### Retrieve the state of your tasks or flows with the `return_state` kwarg
 Beginning with 2.0b9, Prefect 2.0 began returning function results, instead of Prefect futures and states, by default. States are still an important concept in Prefect 2. They can be used to dictate and understand the behavior of your flows. Now, you can access the state for _any_ task or flow with the new `return_state` kwarg. Just set `return_state=True` in you flow or task call and you can access its state with the `.result()` method, even if it's been submitted to a task runner.
 
 ### `prefect cloud` commands are easier to use
 The `prefect cloud login` command no longer overwrites your current profile with a new API URL and auth key. Instead, the command will prompt you to create a new profile when logging into Prefect Cloud 2.0. Subsequent calls to prefect cloud login using the same key will simply "log in" to prefect cloud by switching to the profile associated with that authentication key.
 
-The new `prefect cloud workspace ls` command lists availible workspaces.
+The new `prefect cloud workspace ls` command lists available workspaces.
 
 ### Other improvements and bug fixes
 - The anchor datetime (aka start datetime) for all newly created interval schedules will be the current date & time
@@ -4338,7 +4338,7 @@ You might not see these fixes in your day-to-day, but we're dedicated to improvi
 
 ### Bug fixes
 
-- The CLI no longer displays tracebacks on sucessful exit.
+- The CLI no longer displays tracebacks on successful exit.
 - Returning pandas objects from tasks does not error.
 - Flows are listed correctly in the UI dashboard.
 
diff --git a/docs/cloud/cloud-quickstart.md b/docs/cloud/cloud-quickstart.md
index 116c7bafa5d0..0f6474b6163b 100644
--- a/docs/cloud/cloud-quickstart.md
+++ b/docs/cloud/cloud-quickstart.md
@@ -23,7 +23,7 @@ Get started with Prefect Cloud in just a few steps:
 1. [Run a flow](#run-a-flow-with-prefect-cloud) locally and view flow run execution in Prefect Cloud.
 
 <!-- 
-TK commented out until video udpated
+TK commented out until video updated
 TK to hide TOC for video: add hide: toc 
 
 Prefer to follow this tutorial in a video? We've got exactly what you need. Happy engineering!
diff --git a/docs/cloud/index.md b/docs/cloud/index.md
index b4068b8f0237..9dffa9137fbd 100644
--- a/docs/cloud/index.md
+++ b/docs/cloud/index.md
@@ -19,7 +19,7 @@ search:
 
 Prefect Cloud is a workflow orchestration platform that provides all the capabilities of Prefect server plus additional features, such as:
 
-- automations, events, and webhooks so you can create for event-driven workfows
+- automations, events, and webhooks so you can create for event-driven workflows
 - workspaces, RBAC, SSO, audit logs and related user management tools for collaboration
 - push work pools for running flows on serverless infrastructure without a worker
 - error summaries powered by Marvin AI to help you resolve errors faster
diff --git a/docs/concepts/automations.md b/docs/concepts/automations.md
index 1acd49964ff8..372dde8f6ebe 100644
--- a/docs/concepts/automations.md
+++ b/docs/concepts/automations.md
@@ -50,7 +50,7 @@ Triggers specify the conditions under which your action should be performed. Tri
 - Custom event triggers
 
 !!! note "Automations API"
-    The [automations API](https://app.prefect.cloud/api/docs#tag/Automations) enables further programatic customization of trigger and action policies based on arbitrary [events](https://app.prefect.cloud/api/docs#tag/Events).
+    The [automations API](https://app.prefect.cloud/api/docs#tag/Automations) enables further programmatic customization of trigger and action policies based on arbitrary [events](https://app.prefect.cloud/api/docs#tag/Events).
 
 Importantly, triggers can be configured not only in reaction to events, but also proactively: to trigger in the absence of an event you expect to see.
 
@@ -233,7 +233,7 @@ You can access properties of the underlying flow run objects including:
 - [work_queue](/api-ref/server/schemas/core/#prefect.server.schemas.core.WorkQueue)
 - [work_pool](/api-ref/server/schemas/core/#prefect.server.schemas.core.WorkPool)
 
-In addition to its native properites, each object includes an `id` along with `created` and `updated` timestamps.
+In addition to its native properties, each object includes an `id` along with `created` and `updated` timestamps.
 
 The `flow_run|ui_url` token returns the URL for viewing the flow run in Prefect Cloud.
 
diff --git a/docs/concepts/states.md b/docs/concepts/states.md
index 9317e95565ca..e551d7627c78 100644
--- a/docs/concepts/states.md
+++ b/docs/concepts/states.md
@@ -50,7 +50,7 @@ The full complement of states and state types includes:
 | Pending | PENDING | No | The run has been submitted to run, but is waiting on necessary preconditions to be satisfied. |
 | Running | RUNNING | No | The run code is currently executing. |
 | Retrying | RUNNING | No | The run code is currently executing after previously not complete successfully. |
-| Paused | PAUSED | No | The run code has stopped executing until it recieves manual approval to proceed. |
+| Paused | PAUSED | No | The run code has stopped executing until it receives manual approval to proceed. |
 | Cancelling | CANCELLING | No | The infrastructure on which the code was running is being cleaned up. |
 | Cancelled | CANCELLED | Yes | The run did not complete because a user determined that it should not. |
 | Completed | COMPLETED | Yes | The run completed successfully. |
diff --git a/docs/concepts/task-runners.md b/docs/concepts/task-runners.md
index eec1bf00bb2f..1fc2ef70f285 100644
--- a/docs/concepts/task-runners.md
+++ b/docs/concepts/task-runners.md
@@ -1,5 +1,5 @@
 ---
-description: Prefect task runners let you specify the executors for tasks in a flow run, including returning a PrefectFuture contianing both results and state. 
+description: Prefect task runners let you specify the executors for tasks in a flow run, including returning a PrefectFuture containing both results and state. 
 tags:
     - tasks
     - task runners
diff --git a/docs/contributing/style.md b/docs/contributing/style.md
index 43bab5fff600..17e0cbd9f130 100644
--- a/docs/contributing/style.md
+++ b/docs/contributing/style.md
@@ -190,7 +190,7 @@ Additionally:
 - Capitalize placeholder labels and wrap them in less than (<) and greater than (>) signs.
 - Utilize `textwrap.dedent` to remove extraneous spacing for strings that are written with triple quotes (""").
 
-Placholder Example:
+Placeholder Example:
 ```bash
 Create a work queue with tags:
     prefect work-queue create '<WORK QUEUE NAME>' -t '<OPTIONAL TAG 1>' -t '<OPTIONAL TAG 2>'
diff --git a/docs/contributing/versioning.md b/docs/contributing/versioning.md
index e253b805f719..6cafd3b31259 100644
--- a/docs/contributing/versioning.md
+++ b/docs/contributing/versioning.md
@@ -13,7 +13,7 @@ search:
 
 Versions are composed of three parts: MAJOR.MINOR.PATCH. For example, the version 2.5.0 has a major version of 2, a minor version of 5, and patch version of 0.
 
-Ocassionally, we will add a suffix to the version such as `rc`, `a`, or `b`. These indicate pre-release versions that users can opt-into installing to test functionality before it is ready for release.
+Occasionally, we will add a suffix to the version such as `rc`, `a`, or `b`. These indicate pre-release versions that users can opt-into installing to test functionality before it is ready for release.
 
 Each release will increase one of the version numbers. If we increase a number other than the patch version, the versions to the right of it will be reset to zero.
 
@@ -24,7 +24,7 @@ Prefect will increase the major version when significant and widespread changes
 Prefect will increase the minor version when:
 
 - Introducing a new concept that changes how Prefect can be used
-- Changing an existing concept in a way that fundementally alters how it is used
+- Changing an existing concept in a way that fundamentally alters how it is used
 - Removing a deprecated feature
 
 Prefect will increase the patch version when:
diff --git a/docs/guides/runtime-context.md b/docs/guides/runtime-context.md
index f94abcd2dca5..3ddf6a78efe2 100644
--- a/docs/guides/runtime-context.md
+++ b/docs/guides/runtime-context.md
@@ -14,7 +14,7 @@ search:
 
 Prefect tracks information about the current flow or task run with a run context. The run context can be thought of as a global variable that allows the Prefect engine to determine relationships between your runs, e.g. which flow your task is called from.
 
-The run context itself contains many internal objects used by Prefect to manage execution of your run and is only available in specific situtations. For this reason, we expose a simple interface that only includes the items you care about and dynamically retrieves additional information when necessary. We call this the "runtime context" as it contains information that can only be accessed when a run is happening.
+The run context itself contains many internal objects used by Prefect to manage execution of your run and is only available in specific situations. For this reason, we expose a simple interface that only includes the items you care about and dynamically retrieves additional information when necessary. We call this the "runtime context" as it contains information that can only be accessed when a run is happening.
 
 !!! tip "Mock values via environment variable"
     Oftentimes, you may want to mock certain values for testing purposes.  For example, by manually setting an ID or a scheduled start time to ensure your code is functioning properly.  Starting in version `2.10.3`, you can mock values in runtime via environment variable using the schema `PREFECT__RUNTIME__{SUBMODULE}__{KEY_NAME}=value`:
diff --git a/flows/deployment_run.py b/flows/deployment_run.py
index be6e8f08bc6a..18a90954d2d2 100644
--- a/flows/deployment_run.py
+++ b/flows/deployment_run.py
@@ -8,7 +8,7 @@
 import prefect
 from prefect.deployments import Deployment
 
-# The version oldest verison this test runs with
+# The version oldest version this test runs with
 SUPPORTED_VERSION = "2.6.0"
 
 
diff --git a/flows/deployment_storage.py b/flows/deployment_storage.py
index 658479ad90ef..e345f494e66c 100644
--- a/flows/deployment_storage.py
+++ b/flows/deployment_storage.py
@@ -10,7 +10,7 @@
 from prefect.deployments import Deployment
 from prefect.filesystems import LocalFileSystem
 
-# The version oldest verison this test runs with
+# The version oldest version this test runs with
 SUPPORTED_VERSION = "2.6.0"
 
 
diff --git a/src/prefect/_internal/compatibility/experimental.py b/src/prefect/_internal/compatibility/experimental.py
index 962ba5b3c0f3..d00965cf843a 100644
--- a/src/prefect/_internal/compatibility/experimental.py
+++ b/src/prefect/_internal/compatibility/experimental.py
@@ -38,7 +38,7 @@
 
 class ExperimentalWarning(Warning):
     """
-    A warning related to expirimental code.
+    A warning related to experimental code.
     """
 
 
diff --git a/src/prefect/_internal/concurrency/calls.py b/src/prefect/_internal/concurrency/calls.py
index 17786d3989a2..ac0447d09b4d 100644
--- a/src/prefect/_internal/concurrency/calls.py
+++ b/src/prefect/_internal/concurrency/calls.py
@@ -120,7 +120,7 @@ def cancel(self):
                 if self._cancel_scope is None:
                     return False
                 elif not self._cancel_scope.cancelled():
-                    # Perfom cancellation
+                    # Perform cancellation
                     if not self._cancel_scope.cancel():
                         return False
 
@@ -366,7 +366,7 @@ def __call__(self) -> T:
         """
         Execute the call and return its result.
 
-        All executions during excecution of the call are re-raised.
+        All executions during execution of the call are re-raised.
         """
         coro = self.run()
 
diff --git a/src/prefect/_internal/concurrency/inspection.py b/src/prefect/_internal/concurrency/inspection.py
index 0bea4383138a..756f780b53a0 100644
--- a/src/prefect/_internal/concurrency/inspection.py
+++ b/src/prefect/_internal/concurrency/inspection.py
@@ -10,7 +10,7 @@
 from typing import List
 
 """
-The following functions are dervived from dask/distributed which is licensed under the 
+The following functions are derived from dask/distributed which is licensed under the 
 BSD 3-Clause License.
 
 Copyright (c) 2015, Anaconda, Inc. and contributors
diff --git a/src/prefect/_internal/concurrency/services.py b/src/prefect/_internal/concurrency/services.py
index 6b81e0ddcdf2..869e394d56c4 100644
--- a/src/prefect/_internal/concurrency/services.py
+++ b/src/prefect/_internal/concurrency/services.py
@@ -102,7 +102,7 @@ def send(self, item: T):
             if self._stopped:
                 raise RuntimeError("Cannot put items in a stopped service instance.")
 
-            logger.debug("Service %r enqueing item %r", self, item)
+            logger.debug("Service %r enqueuing item %r", self, item)
             self._queue.put_nowait(self._prepare_item(item))
 
     def _prepare_item(self, item: T) -> T:
@@ -223,7 +223,7 @@ def drain_all(cls, timeout: Optional[float] = None) -> Union[Awaitable, None]:
                     [asyncio.wrap_future(fut) for fut in futures], timeout=timeout
                 )
                 if futures
-                # `wait` errors if it receieves an empty list but we need to return a
+                # `wait` errors if it receives an empty list but we need to return a
                 # coroutine still
                 else asyncio.sleep(0)
             )
diff --git a/src/prefect/_version.py b/src/prefect/_version.py
index ec4d032fc9a3..cddd2d14b69c 100644
--- a/src/prefect/_version.py
+++ b/src/prefect/_version.py
@@ -330,7 +330,7 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, runner=run_command):
         # TAG-NUM-gHEX
         mo = re.search(r"^(.+)-(\d+)-g([0-9a-f]+)$", git_describe)
         if not mo:
-            # unparseable. Maybe git-describe is misbehaving?
+            # unparsable. Maybe git-describe is misbehaving?
             pieces["error"] = "unable to parse git-describe output: '%s'" % describe_out
             return pieces
 
diff --git a/src/prefect/agent.py b/src/prefect/agent.py
index c146dcd9c8d9..e7b7580f1f83 100644
--- a/src/prefect/agent.py
+++ b/src/prefect/agent.py
@@ -469,7 +469,7 @@ async def submit_run(self, flow_run: FlowRun) -> None:
                         )
                     except Exception:
                         self.logger.exception(
-                            "An error occured while setting the `infrastructure_pid`"
+                            "An error occurred while setting the `infrastructure_pid`"
                             f" on flow run {flow_run.id!r}. The flow run will not be"
                             " cancellable."
                         )
@@ -510,7 +510,7 @@ async def _submit_run_and_capture_errors(
                 )
             else:
                 self.logger.exception(
-                    f"An error occured while monitoring flow run '{flow_run.id}'. "
+                    f"An error occurred while monitoring flow run '{flow_run.id}'. "
                     "The flow run will not be marked as failed, but an issue may have "
                     "occurred."
                 )
diff --git a/src/prefect/blocks/core.py b/src/prefect/blocks/core.py
index ba202ef4d712..92a32e7b0d8b 100644
--- a/src/prefect/blocks/core.py
+++ b/src/prefect/blocks/core.py
@@ -667,14 +667,14 @@ def _event_method_called_resources(self) -> Optional[ResourceTuple]:
     @classmethod
     def get_block_class_from_schema(cls: Type[Self], schema: BlockSchema) -> Type[Self]:
         """
-        Retieve the block class implementation given a schema.
+        Retrieve the block class implementation given a schema.
         """
         return cls.get_block_class_from_key(block_schema_to_key(schema))
 
     @classmethod
     def get_block_class_from_key(cls: Type[Self], key: str) -> Type[Self]:
         """
-        Retieve the block class implementation given a key.
+        Retrieve the block class implementation given a key.
         """
         # Ensure collections are imported and have the opportunity to register types
         # before looking up the block class
diff --git a/src/prefect/cli/dev.py b/src/prefect/cli/dev.py
index de6837d3bf12..5896fcde45ac 100644
--- a/src/prefect/cli/dev.py
+++ b/src/prefect/cli/dev.py
@@ -357,7 +357,7 @@ def build_image(
     Build a docker image for development.
     """
     exit_with_error_if_not_editable_install()
-    # TODO: Once https://github.com/tiangolo/typer/issues/354 is addresesd, the
+    # TODO: Once https://github.com/tiangolo/typer/issues/354 is addressed, the
     #       default can be set in the function signature
     arch = arch or platform.machine()
     python_version = python_version or python_version_minor()
diff --git a/src/prefect/cli/flow_run.py b/src/prefect/cli/flow_run.py
index ec70465ea91a..a711ffd07368 100644
--- a/src/prefect/cli/flow_run.py
+++ b/src/prefect/cli/flow_run.py
@@ -141,7 +141,7 @@ async def cancel(id: UUID):
             f" '{result.details.reason}'"
         )
 
-    exit_with_success(f"Flow run '{id}' was succcessfully scheduled for cancellation.")
+    exit_with_success(f"Flow run '{id}' was successfully scheduled for cancellation.")
 
 
 @flow_run_app.command()
diff --git a/src/prefect/cli/profile.py b/src/prefect/cli/profile.py
index c925323efdf5..d74e57790689 100644
--- a/src/prefect/cli/profile.py
+++ b/src/prefect/cli/profile.py
@@ -271,7 +271,7 @@ async def check_orion_connection():
         return ConnectionStatus.CLOUD_UNAUTHORIZED
     except httpx.HTTPStatusError as exc:
         if exc.response.status_code == status.HTTP_404_NOT_FOUND:
-            # if the route does not exist, attmpt to connect as a hosted Prefect
+            # if the route does not exist, attempt to connect as a hosted Prefect
             # instance
             try:
                 # inform the user if Prefect API endpoints exist, but there are
diff --git a/src/prefect/cli/server.py b/src/prefect/cli/server.py
index e5b11f79b38c..fdfa18d6220c 100644
--- a/src/prefect/cli/server.py
+++ b/src/prefect/cli/server.py
@@ -75,7 +75,7 @@ def generate_welcome_blurb(base_url, ui_enabled: bool):
 
     dashboard_disabled = textwrap.dedent(
         """
-        The dashboard is disabled. Set `PREFECT_UI_ENABLED=1` to reenable it.
+        The dashboard is disabled. Set `PREFECT_UI_ENABLED=1` to re-enable it.
         """
     )
 
diff --git a/src/prefect/client/schemas/objects.py b/src/prefect/client/schemas/objects.py
index e2e21fc0a3f2..150d6ed8039f 100644
--- a/src/prefect/client/schemas/objects.py
+++ b/src/prefect/client/schemas/objects.py
@@ -1212,7 +1212,7 @@ def evaluate_health_status(
         self, late_runs_count: int, last_polled: Optional[DateTimeTZ] = None
     ) -> bool:
         """
-        Given empirical information about the state of the work queue, evaulate its health status.
+        Given empirical information about the state of the work queue, evaluate its health status.
 
         Args:
             late_runs: the count of late runs for the work queue.
diff --git a/src/prefect/concurrency/asyncio.py b/src/prefect/concurrency/asyncio.py
index e11e759727e8..a86ca90bf3aa 100644
--- a/src/prefect/concurrency/asyncio.py
+++ b/src/prefect/concurrency/asyncio.py
@@ -44,7 +44,7 @@ async def rate_limit(names: Union[str, List[str]], occupy: int = 1):
 
     Args:
         names: The names of the concurrency limits to acquire slots from.
-        occupy: The number of slots to acquire and holf from each limit.
+        occupy: The number of slots to acquire and hold from each limit.
     """
     names = names if isinstance(names, list) else [names]
     limits = await _acquire_concurrency_slots(names, occupy, mode="rate_limit")
diff --git a/src/prefect/concurrency/services.py b/src/prefect/concurrency/services.py
index ed6ed6bb9fce..d3f8ba2c6b9b 100644
--- a/src/prefect/concurrency/services.py
+++ b/src/prefect/concurrency/services.py
@@ -63,7 +63,7 @@ def send(self, item: Tuple[int, str]):
             if self._stopped:
                 raise RuntimeError("Cannot put items in a stopped service instance.")
 
-            logger.debug("Service %r enqueing item %r", self, item)
+            logger.debug("Service %r enqueuing item %r", self, item)
             future = concurrent.futures.Future()
 
             occupy, mode = item
diff --git a/src/prefect/concurrency/sync.py b/src/prefect/concurrency/sync.py
index b7f03f290143..5c445f8b5b09 100644
--- a/src/prefect/concurrency/sync.py
+++ b/src/prefect/concurrency/sync.py
@@ -46,7 +46,7 @@ def rate_limit(names: Union[str, List[str]], occupy: int = 1):
 
     Args:
         names: The names of the concurrency limits to acquire slots from.
-        occupy: The number of slots to acquire and holf from each limit.
+        occupy: The number of slots to acquire and hold from each limit.
     """
     names = names if isinstance(names, list) else [names]
     limits = _call_async_function_from_sync(
diff --git a/src/prefect/engine.py b/src/prefect/engine.py
index 14e3a70f1ce9..52fb4a70bdd5 100644
--- a/src/prefect/engine.py
+++ b/src/prefect/engine.py
@@ -884,7 +884,7 @@ async def orchestrate_flow_run(
             )
 
         if not waited_for_task_runs:
-            # An exception occured that prevented us from waiting for task runs to
+            # An exception occurred that prevented us from waiting for task runs to
             # complete. Ensure that we wait for them before proposing a final state
             # for the flow run.
             await wait_for_task_runs_and_report_crashes(
@@ -1241,7 +1241,7 @@ async def begin_task_map(
 
 async def collect_task_run_inputs(expr: Any, max_depth: int = -1) -> Set[TaskRunInput]:
     """
-    This function recurses through an expression to generate a set of any discernable
+    This function recurses through an expression to generate a set of any discernible
     task run inputs it finds in the data structure. It produces a set of all inputs
     found.
 
@@ -2047,7 +2047,7 @@ def resolve_input(expr, context):
         if not state.is_completed() and not (
             # TODO: Note that the contextual annotation here is only at the current level
             #       if `allow_failure` is used then another annotation is used, this will
-            #       incorrectly evaulate to false — to resolve this, we must track all
+            #       incorrectly evaluate to false — to resolve this, we must track all
             #       annotations wrapping the current expression but this is not yet
             #       implemented.
             isinstance(context.get("annotation"), allow_failure)
@@ -2469,7 +2469,7 @@ def _emit_task_run_state_change_event(
         )
     except Exception:
         engine_logger.error(
-            f"Invalid flow run id. Recieved arguments: {sys.argv}", exc_info=True
+            f"Invalid flow run id. Received arguments: {sys.argv}", exc_info=True
         )
         exit(1)
 
diff --git a/src/prefect/events/schemas.py b/src/prefect/events/schemas.py
index 988b0889a5d2..744a87ba9760 100644
--- a/src/prefect/events/schemas.py
+++ b/src/prefect/events/schemas.py
@@ -190,8 +190,8 @@ class Trigger(PrefectBaseModel):
     for_each: Set[str] = Field(
         default_factory=set,
         description=(
-            "Evalute the Automation separately for each distinct value of these labels "
-            "on the resource"
+            "Evaluate the Automation separately for each distinct value of these labels"
+            " on the resource"
         ),
     )
     posture: Posture = Field(
diff --git a/src/prefect/exceptions.py b/src/prefect/exceptions.py
index 189a1fa88f94..98b9e08bbf3e 100644
--- a/src/prefect/exceptions.py
+++ b/src/prefect/exceptions.py
@@ -380,7 +380,7 @@ class InfrastructureNotFound(PrefectException):
 
 class InfrastructureNotAvailable(PrefectException):
     """
-    Raised when infrastructure is not accessable from the current machine. For example,
+    Raised when infrastructure is not accessible from the current machine. For example,
     if a process was spawned on another machine it cannot be managed.
     """
 
diff --git a/src/prefect/filesystems.py b/src/prefect/filesystems.py
index e9cae0330f7a..0f7593940777 100644
--- a/src/prefect/filesystems.py
+++ b/src/prefect/filesystems.py
@@ -318,7 +318,7 @@ async def get_directory(
         self, from_path: Optional[str] = None, local_path: Optional[str] = None
     ) -> None:
         """
-        Downloads a directory from a given remote path to a local direcotry.
+        Downloads a directory from a given remote path to a local directory.
 
         Defaults to downloading the entire contents of the block's basepath to the current working directory.
         """
@@ -345,7 +345,7 @@ async def put_directory(
         overwrite: bool = True,
     ) -> int:
         """
-        Uploads a directory from a given local path to a remote direcotry.
+        Uploads a directory from a given local path to a remote directory.
 
         Defaults to uploading the entire contents of the current working directory to the block's basepath.
         """
@@ -716,7 +716,7 @@ async def get_directory(
         self, from_path: Optional[str] = None, local_path: Optional[str] = None
     ) -> bytes:
         """
-        Downloads a directory from a given remote path to a local direcotry.
+        Downloads a directory from a given remote path to a local directory.
 
         Defaults to downloading the entire contents of the block's basepath to the current working directory.
         """
diff --git a/src/prefect/flows.py b/src/prefect/flows.py
index 8766caeda7cd..8b3ed4c2b04d 100644
--- a/src/prefect/flows.py
+++ b/src/prefect/flows.py
@@ -131,7 +131,7 @@ class Flow(Generic[P, R]):
             should be persisted to result storage. Defaults to `None`, which indicates
             that Prefect should choose whether the result should be persisted depending on
             the features being used.
-        result_storage: An optional block to use to perist the result of this flow.
+        result_storage: An optional block to use to persist the result of this flow.
             This value will be used as the default for any tasks in this flow.
             If not provided, the local file system will be used unless called as
             a subflow, at which point the default will be loaded from the parent flow.
@@ -949,7 +949,7 @@ def flow(
             should be persisted to result storage. Defaults to `None`, which indicates
             that Prefect should choose whether the result should be persisted depending on
             the features being used.
-        result_storage: An optional block to use to perist the result of this flow.
+        result_storage: An optional block to use to persist the result of this flow.
             This value will be used as the default for any tasks in this flow.
             If not provided, the local file system will be used unless called as
             a subflow, at which point the default will be loaded from the parent flow.
diff --git a/src/prefect/futures.py b/src/prefect/futures.py
index 1182e6d73396..09d5d67af15d 100644
--- a/src/prefect/futures.py
+++ b/src/prefect/futures.py
@@ -69,7 +69,7 @@ class PrefectFuture(Generic[R, A]):
         >>>     future = my_task.submit()
         >>>     final_state = future.wait()
 
-        Wait N sconds for the task to complete
+        Wait N seconds for the task to complete
 
         >>> @flow
         >>> def my_flow():
diff --git a/src/prefect/infrastructure/container.py b/src/prefect/infrastructure/container.py
index 98e1ba758356..68bce923a8b4 100644
--- a/src/prefect/infrastructure/container.py
+++ b/src/prefect/infrastructure/container.py
@@ -618,8 +618,8 @@ def _watch_container(
                     )
                 else:
                     self.logger.exception(
-                        "An unexpected Docker API error occured while streaming output "
-                        f"from container {container.name}."
+                        "An unexpected Docker API error occurred while streaming"
+                        f" output from container {container.name}."
                     )
 
             container.reload()
diff --git a/src/prefect/infrastructure/kubernetes.py b/src/prefect/infrastructure/kubernetes.py
index bdcd5af58fb4..5468ea80c0ca 100644
--- a/src/prefect/infrastructure/kubernetes.py
+++ b/src/prefect/infrastructure/kubernetes.py
@@ -184,7 +184,7 @@ def ensure_job_has_compatible_values(cls, value: KubernetesManifest):
         )
         if incompatible:
             raise ValueError(
-                "Job has incompatble values for the following attributes: "
+                "Job has incompatible values for the following attributes: "
                 f"{', '.join(incompatible)}"
             )
         return value
diff --git a/src/prefect/logging/logging.yml b/src/prefect/logging/logging.yml
index dc9f9136a1ed..5c4f6676a1a6 100644
--- a/src/prefect/logging/logging.yml
+++ b/src/prefect/logging/logging.yml
@@ -36,7 +36,7 @@ formatters:
 
 handlers:
 
-    # The handlers we define here will output all logs they receieve by default
+    # The handlers we define here will output all logs they receive by default
     # but we include the `level` so it can be overridden by environment
 
     console:
diff --git a/src/prefect/serializers.py b/src/prefect/serializers.py
index 550387ffb736..4c2ac4c8e007 100644
--- a/src/prefect/serializers.py
+++ b/src/prefect/serializers.py
@@ -255,18 +255,18 @@ def check_compressionlib(cls, value):
         methods.
         """
         try:
-            compresser = from_qualified_name(value)
+            compressor = from_qualified_name(value)
         except (ImportError, AttributeError) as exc:
             raise ValueError(
                 f"Failed to import requested compression library: {value!r}."
             ) from exc
 
-        if not callable(getattr(compresser, "compress", None)):
+        if not callable(getattr(compressor, "compress", None)):
             raise ValueError(
                 f"Compression library at {value!r} does not have a 'compress' method."
             )
 
-        if not callable(getattr(compresser, "decompress", None)):
+        if not callable(getattr(compressor, "decompress", None)):
             raise ValueError(
                 f"Compression library at {value!r} does not have a 'decompress' method."
             )
@@ -275,12 +275,12 @@ def check_compressionlib(cls, value):
 
     def dumps(self, obj: Any) -> bytes:
         blob = self.serializer.dumps(obj)
-        compresser = from_qualified_name(self.compressionlib)
-        return base64.encodebytes(compresser.compress(blob))
+        compressor = from_qualified_name(self.compressionlib)
+        return base64.encodebytes(compressor.compress(blob))
 
     def loads(self, blob: bytes) -> Any:
-        compresser = from_qualified_name(self.compressionlib)
-        uncompressed = compresser.decompress(base64.decodebytes(blob))
+        compressor = from_qualified_name(self.compressionlib)
+        uncompressed = compressor.decompress(base64.decodebytes(blob))
         return self.serializer.loads(uncompressed)
 
 
diff --git a/src/prefect/server/api/dependencies.py b/src/prefect/server/api/dependencies.py
index de57ab94045c..8b443cdf7288 100644
--- a/src/prefect/server/api/dependencies.py
+++ b/src/prefect/server/api/dependencies.py
@@ -92,7 +92,7 @@ async def _notify_of_outdated_version(self, request_version: str):
 def LimitBody() -> Depends:
     """
     A `fastapi.Depends` factory for pulling a `limit: int` parameter from the
-    request body while determing the default from the current settings.
+    request body while determining the default from the current settings.
     """
 
     def get_limit(
diff --git a/src/prefect/server/api/deployments.py b/src/prefect/server/api/deployments.py
index 9fba0ad6aac0..62cebb517437 100644
--- a/src/prefect/server/api/deployments.py
+++ b/src/prefect/server/api/deployments.py
@@ -471,7 +471,7 @@ async def create_flow_run_from_deployment(
         work_queue_id = deployment.work_queue_id
 
         if flow_run.work_queue_name:
-            # cant mutate the ORM model or else it will commit the changes back
+            # can't mutate the ORM model or else it will commit the changes back
             work_queue_id = await worker_lookups._get_work_queue_id_from_name(
                 session=session,
                 work_pool_name=deployment.work_queue.work_pool.name,
diff --git a/src/prefect/server/api/server.py b/src/prefect/server/api/server.py
index c64c803f9d8f..3a7f1926f45c 100644
--- a/src/prefect/server/api/server.py
+++ b/src/prefect/server/api/server.py
@@ -552,7 +552,7 @@ def on_service_exit(service, task):
         allow_headers=["*"],
     )
 
-    # Limit the number of concurrent requests when using a SQLite datbase to reduce
+    # Limit the number of concurrent requests when using a SQLite database to reduce
     # chance of errors where the database cannot be opened due to a high number of
     # concurrent writes
     if (
diff --git a/src/prefect/server/collection_blocks_data.json b/src/prefect/server/collection_blocks_data.json
index b54361af0d79..39222c93755e 100644
--- a/src/prefect/server/collection_blocks_data.json
+++ b/src/prefect/server/collection_blocks_data.json
@@ -170,7 +170,7 @@
                   },
                   "launch_type": {
                     "title": "Launch Type",
-                    "description": "The type of ECS task run infrastructure that should be used. Note that 'FARGATE_SPOT' is not a formal ECS launch type, but we will configure the proper capacity provider stategy if set here.",
+                    "description": "The type of ECS task run infrastructure that should be used. Note that 'FARGATE_SPOT' is not a formal ECS launch type, but we will configure the proper capacity provider strategy if set here.",
                     "default": "FARGATE",
                     "enum": ["FARGATE", "EC2", "EXTERNAL", "FARGATE_SPOT"],
                     "type": "string"
diff --git a/src/prefect/server/database/configurations.py b/src/prefect/server/database/configurations.py
index 0082a270304e..f47edc317b3f 100644
--- a/src/prefect/server/database/configurations.py
+++ b/src/prefect/server/database/configurations.py
@@ -320,7 +320,7 @@ def setup_sqlite(self, conn, record):
         cursor.execute("PRAGMA cache_size = 20000;")
 
         # wait for this amount of time while a table is locked
-        # before returning and rasing an error
+        # before returning and raising an error
         # setting the value very high allows for more 'concurrency'
         # without running into errors, but may result in slow api calls
         cursor.execute("PRAGMA busy_timeout = 60000;")  # 60s
diff --git a/src/prefect/server/database/migrations/MIGRATION-NOTES.md b/src/prefect/server/database/migrations/MIGRATION-NOTES.md
index b2162c9eb8b3..272033b072c4 100644
--- a/src/prefect/server/database/migrations/MIGRATION-NOTES.md
+++ b/src/prefect/server/database/migrations/MIGRATION-NOTES.md
@@ -276,7 +276,7 @@ Adds tables for storing block data.
 SQLite: `619bea85701a`
 Postgres: `5f376def75c3`
 
-## Intial
+## Initial
 
 Creates the database that previously was not managed by migrations.
 
diff --git a/src/prefect/server/database/query_components.py b/src/prefect/server/database/query_components.py
index f01a02864124..3e4523f3f187 100644
--- a/src/prefect/server/database/query_components.py
+++ b/src/prefect/server/database/query_components.py
@@ -495,7 +495,7 @@ async def read_configuration_value(
         Configuration values should not be changed at run time, so retrieved
         values are cached in memory.
 
-        The main use of confiugrations is encrypting blocks, this speeds up nested
+        The main use of configurations is encrypting blocks, this speeds up nested
         block document queries.
         """
         try:
diff --git a/src/prefect/server/database/sql/postgres/get-runs-from-worker-queues.sql.jinja b/src/prefect/server/database/sql/postgres/get-runs-from-worker-queues.sql.jinja
index 5ce832e6d6b8..f165d9770116 100644
--- a/src/prefect/server/database/sql/postgres/get-runs-from-worker-queues.sql.jinja
+++ b/src/prefect/server/database/sql/postgres/get-runs-from-worker-queues.sql.jinja
@@ -16,7 +16,7 @@ WITH pool_slots AS (
         wp.id
 ),
 
--- compute avaialble slots under worker pool queue concurrency limits
+-- compute available slots under worker pool queue concurrency limits
 queue_slots AS (
     SELECT
         wq.id,
diff --git a/src/prefect/server/database/sql/sqlite/get-runs-from-worker-queues.sql.jinja b/src/prefect/server/database/sql/sqlite/get-runs-from-worker-queues.sql.jinja
index 60cab9274ea0..253464fe496c 100644
--- a/src/prefect/server/database/sql/sqlite/get-runs-from-worker-queues.sql.jinja
+++ b/src/prefect/server/database/sql/sqlite/get-runs-from-worker-queues.sql.jinja
@@ -16,7 +16,7 @@ WITH worker_slots AS (
 ),
 
 
--- compute avaialble slots under worker pool queue concurrency limits
+-- compute available slots under worker pool queue concurrency limits
 queue_slots AS (
     SELECT
         wq.id,
diff --git a/src/prefect/server/models/block_schemas.py b/src/prefect/server/models/block_schemas.py
index 292a1c73bce9..febeb17fcc85 100644
--- a/src/prefect/server/models/block_schemas.py
+++ b/src/prefect/server/models/block_schemas.py
@@ -472,7 +472,7 @@ def _add_block_schemas_fields_to_definitions(
 ):
     """
     Returns a new definitions dict with the fields of a block schema and it's child
-    block schemas added to the existing defintions.
+    block schemas added to the existing definitions.
     """
     block_schema_title = child_block_schema.fields.get("title")
     if block_schema_title is not None:
diff --git a/src/prefect/server/models/deployments.py b/src/prefect/server/models/deployments.py
index 6d2ca415f868..92d67f3b497e 100644
--- a/src/prefect/server/models/deployments.py
+++ b/src/prefect/server/models/deployments.py
@@ -38,7 +38,7 @@ async def _delete_scheduled_runs(
     modified in order to ensure that future runs comply with the deployment's latest values.
 
     Args:
-        deployment_id: the deplyment for which we should delete runs.
+        deployment_id: the deployment for which we should delete runs.
         auto_scheduled_only: if True, only delete auto scheduled runs. Defaults to `False`.
     """
     delete_query = sa.delete(db.FlowRun).where(
@@ -661,7 +661,7 @@ async def check_work_queues_for_deployment(
 
     - Our database currently allows either "null" and empty lists as
     null values in filters, so we need to catch both cases with "or".
-    - `json_contains(A, B)` should be interepreted as "True if A
+    - `json_contains(A, B)` should be interpreted as "True if A
     contains B".
 
     Returns:
diff --git a/src/prefect/server/models/flow_runs.py b/src/prefect/server/models/flow_runs.py
index fbcce245fd50..17af5919a127 100644
--- a/src/prefect/server/models/flow_runs.py
+++ b/src/prefect/server/models/flow_runs.py
@@ -264,7 +264,7 @@ async def read_flow_runs(
         flow_filter: only select flow runs whose flows match these filters
         flow_run_filter: only select flow runs match these filters
         task_run_filter: only select flow runs whose task runs match these filters
-        deployment_filter: only sleect flow runs whose deployments match these filters
+        deployment_filter: only select flow runs whose deployments match these filters
         offset: Query offset
         limit: Query limit
         sort: Query sort
diff --git a/src/prefect/server/models/work_queues.py b/src/prefect/server/models/work_queues.py
index 7c447903fe57..daa7f2657eb3 100644
--- a/src/prefect/server/models/work_queues.py
+++ b/src/prefect/server/models/work_queues.py
@@ -327,7 +327,7 @@ async def _legacy_get_runs_in_work_queue(
     # if the work queue has a concurrency limit, check how many runs are currently
     # executing and compare that count to the concurrency limit
     if work_queue.concurrency_limit is not None:
-        # Note this does not guarantee race conditions wont be hit
+        # Note this does not guarantee race conditions won't be hit
         running_frs = await models.flow_runs.count_flow_runs(
             session=session,
             flow_run_filter=schemas.filters.FlowRunFilter(
diff --git a/src/prefect/server/orchestration/global_policy.py b/src/prefect/server/orchestration/global_policy.py
index 0d807f8594e0..adde13fe10d7 100644
--- a/src/prefect/server/orchestration/global_policy.py
+++ b/src/prefect/server/orchestration/global_policy.py
@@ -4,7 +4,7 @@
 For clarity, `GlobalFlowpolicy` and `GlobalTaskPolicy` contain all transition logic
 implemented using [`BaseUniversalTransform`][prefect.server.orchestration.rules.BaseUniversalTransform].
 None of these operations modify state, and regardless of what orchestration Prefect REST API might
-enforce on a transtition, the global policies contain Prefect's necessary bookkeeping.
+enforce on a transition, the global policies contain Prefect's necessary bookkeeping.
 Because these transforms record information about the validated state committed to the
 state database, they should be the most deeply nested contexts in orchestration loop.
 """
diff --git a/src/prefect/server/orchestration/rules.py b/src/prefect/server/orchestration/rules.py
index d841616372c8..c6df4f8abdf4 100644
--- a/src/prefect/server/orchestration/rules.py
+++ b/src/prefect/server/orchestration/rules.py
@@ -662,7 +662,7 @@ async def before_transition(
             mutating the run can also cause unintended writes to the database.
 
         Args:
-            initial_state: The initial state of a transtion
+            initial_state: The initial state of a transition
             proposed_state: The proposed state of a transition
             context: A safe copy of the `OrchestrationContext`, with the exception of
                 `context.run`, mutating this context will have no effect on the broader
@@ -682,7 +682,7 @@ async def after_transition(
         Implements a hook that can fire after a state is committed to the database.
 
         Args:
-            initial_state: The initial state of a transtion
+            initial_state: The initial state of a transition
             validated_state: The governed state that has been committed to the database
             context: A safe copy of the `OrchestrationContext`, with the exception of
                 `context.run`, mutating this context will have no effect on the broader
@@ -707,7 +707,7 @@ async def cleanup(
         keeps track of all other rules that might govern a transition.
 
         Args:
-            initial_state: The initial state of a transtion
+            initial_state: The initial state of a transition
             validated_state: The governed state that has been committed to the database
             context: A safe copy of the `OrchestrationContext`, with the exception of
                 `context.run`, mutating this context will have no effect on the broader
diff --git a/src/prefect/server/schemas/core.py b/src/prefect/server/schemas/core.py
index abf9d52fda2e..61cfbbc41350 100644
--- a/src/prefect/server/schemas/core.py
+++ b/src/prefect/server/schemas/core.py
@@ -938,7 +938,7 @@ def evaluate_health_status(
         self, late_runs_count: int, last_polled: Optional[DateTimeTZ] = None
     ) -> bool:
         """
-        Given empirical information about the state of the work queue, evaulate its health status.
+        Given empirical information about the state of the work queue, evaluate its health status.
 
         Args:
             late_runs: the count of late runs for the work queue.
diff --git a/src/prefect/server/schemas/filters.py b/src/prefect/server/schemas/filters.py
index 07e088c25377..edad8325ec6b 100644
--- a/src/prefect/server/schemas/filters.py
+++ b/src/prefect/server/schemas/filters.py
@@ -21,7 +21,7 @@
 
 sa = lazy_import("sqlalchemy")
 
-# TOOD: Consider moving the `as_sql_filter` functions out of here since they are a
+# TODO: Consider moving the `as_sql_filter` functions out of here since they are a
 #       database model level function and do not properly separate concerns when
 #       present in the schemas module
 
diff --git a/src/prefect/server/schemas/sorting.py b/src/prefect/server/schemas/sorting.py
index c2c03eb6c72a..7809e8b8947a 100644
--- a/src/prefect/server/schemas/sorting.py
+++ b/src/prefect/server/schemas/sorting.py
@@ -11,7 +11,7 @@
 
     from prefect.server.database.interface import PrefectDBInterface
 
-# TOOD: Consider moving the `as_sql_sort` functions out of here since they are a
+# TODO: Consider moving the `as_sql_sort` functions out of here since they are a
 #       database model level function and do not properly separate concerns when
 #       present in the schemas module
 
diff --git a/src/prefect/server/schemas/states.py b/src/prefect/server/schemas/states.py
index 5b8f58ce0d9a..1b96874f8a33 100644
--- a/src/prefect/server/schemas/states.py
+++ b/src/prefect/server/schemas/states.py
@@ -107,7 +107,7 @@ def from_orm_without_result(
         """
         During orchestration, ORM states can be instantiated prior to inserting results
         into the artifact table and the `data` field will not be eagerly loaded. In
-        these cases, sqlalchemy will attept to lazily load the the relationship, which
+        these cases, sqlalchemy will attempt to lazily load the the relationship, which
         will fail when called within a synchronous pydantic method.
 
         This method will construct a `State` object from an ORM model without a loaded
diff --git a/src/prefect/server/services/loop_service.py b/src/prefect/server/services/loop_service.py
index 97d1daf07dd8..c619628d7c42 100644
--- a/src/prefect/server/services/loop_service.py
+++ b/src/prefect/server/services/loop_service.py
@@ -143,7 +143,7 @@ async def stop(self, block=True) -> None:
                     f"`stop(block=True)` was called on {self.name} but more than one"
                     f" loop interval ({self.loop_seconds} seconds) has passed. This"
                     " usually means something is wrong. If `stop()` was called from"
-                    " inside the loop service, use `stop(block=False)` isntead."
+                    " inside the loop service, use `stop(block=False)` instead."
                 )
 
     def _stop(self, *_) -> None:
diff --git a/src/prefect/server/utilities/database.py b/src/prefect/server/utilities/database.py
index e37c14de0148..72ad994460e9 100644
--- a/src/prefect/server/utilities/database.py
+++ b/src/prefect/server/utilities/database.py
@@ -88,7 +88,7 @@ def load_dialect_impl(self, dialect):
                     # as alphanumeric comparisons without regard for actual timestamp
                     # semantics or timezones. Therefore, it's important to have uniform
                     # and sortable datetime representations. The default is an ISO8601-compatible
-                    # string with NO time zone and a space (" ") delimeter between the date
+                    # string with NO time zone and a space (" ") delimiter between the date
                     # and the time. The below settings can be used to add a "T" delimiter but
                     # will require all other sqlite datetimes to be set similarly, including
                     # the custom default value for datetime columns and any handwritten SQL
diff --git a/src/prefect/server/utilities/server.py b/src/prefect/server/utilities/server.py
index aac950bbb200..aba8a76c52f5 100644
--- a/src/prefect/server/utilities/server.py
+++ b/src/prefect/server/utilities/server.py
@@ -30,7 +30,7 @@ def response_scoped_dependency(dependency: Callable):
     default, FastAPI closes dependencies after sending the response.
 
     Uses an async stack that is exited before the response is returned. This is
-    particularly useful for database sesssions which must be committed before the client
+    particularly useful for database sessions which must be committed before the client
     can do more work.
 
     NOTE: Do not use a response-scoped dependency within a FastAPI background task.
@@ -128,7 +128,7 @@ def add_api_route(
         """
         if kwargs.get("status_code") == status.HTTP_204_NO_CONTENT:
             # any routes that return No-Content status codes must
-            # explicilty set a response_class that will handle status codes
+            # explicitly set a response_class that will handle status codes
             # and not return anything in the body
             kwargs["response_class"] = Response
         if kwargs.get("response_model") is None:
diff --git a/src/prefect/settings.py b/src/prefect/settings.py
index b6a30b9139dd..2fbf0ad83fac 100644
--- a/src/prefect/settings.py
+++ b/src/prefect/settings.py
@@ -5,7 +5,7 @@
 caps, matching the environment variable that can be used to change the setting.
 
 All settings defined in this file are used to generate a dynamic Pydantic settings class
-called `Settings`. When insantiated, this class will load settings from environment
+called `Settings`. When instantiated, this class will load settings from environment
 variables and pull default values from the setting definitions.
 
 The current instance of `Settings` being used by the application is stored in a
@@ -680,7 +680,7 @@ def default_cloud_ui_url(settings, value):
 
 PREFECT_EXPERIMENTAL_WARN = Setting(bool, default=True)
 """
-If enabled, warn on usage of expirimental features.
+If enabled, warn on usage of experimental features.
 """
 
 PREFECT_PROFILES_PATH = Setting(
@@ -730,7 +730,7 @@ def default_cloud_ui_url(settings, value):
 PREFECT_FLOW_DEFAULT_RETRY_DELAY_SECONDS = Setting(Union[int, float], default=0)
 """
 This value sets the retry delay seconds for all flows.
-This value does not overwrite invidually set retry delay seconds
+This value does not overwrite individually set retry delay seconds
 """
 
 PREFECT_TASK_DEFAULT_RETRY_DELAY_SECONDS = Setting(
@@ -738,7 +738,7 @@ def default_cloud_ui_url(settings, value):
 )
 """
 This value sets the default retry delay seconds for all tasks.
-This value does not overwrite invidually set retry delay seconds
+This value does not overwrite individually set retry delay seconds
 """
 
 PREFECT_LOCAL_STORAGE_PATH = Setting(
@@ -821,7 +821,7 @@ def default_cloud_ui_url(settings, value):
 )
 """
 If set, `print` statements in flows and tasks will be redirected to the Prefect logger
-for the given run. This setting can be overriden by individual tasks and flows.
+for the given run. This setting can be overridden by individual tasks and flows.
 """
 
 PREFECT_LOGGING_TO_API_ENABLED = Setting(
diff --git a/src/prefect/states.py b/src/prefect/states.py
index 3b564605c32f..74ab6a7d7784 100644
--- a/src/prefect/states.py
+++ b/src/prefect/states.py
@@ -414,7 +414,7 @@ def is_state_iterable(obj: Any) -> TypeGuard[Iterable[State]]:
 
     Other iterables will return `False` even if they contain states.
     """
-    # We do not check for arbitary iterables because this is not intended to be used
+    # We do not check for arbitrary iterables because this is not intended to be used
     # for things like dictionaries, dataframes, or pydantic models
     if (
         not isinstance(obj, BaseAnnotation)
diff --git a/src/prefect/task_runners.py b/src/prefect/task_runners.py
index 1606ed34935d..818f640cbc1a 100644
--- a/src/prefect/task_runners.py
+++ b/src/prefect/task_runners.py
@@ -313,7 +313,7 @@ async def _run_and_store_result(
         """
         Simple utility to store the orchestration result in memory on completion
 
-        Since this run is occuring on the main thread, we capture exceptions to prevent
+        Since this run is occurring on the main thread, we capture exceptions to prevent
         task crashes from crashing the flow run.
         """
         try:
@@ -330,7 +330,7 @@ async def _get_run_result(
         """
         Block until the run result has been populated.
         """
-        result = None  # retval on tiemout
+        result = None  # retval on timeout
 
         # Note we do not use `asyncio.wrap_future` and instead use an `Event` to avoid
         # stdlib behavior where the wrapped future is cancelled if the parent future is
diff --git a/src/prefect/tasks.py b/src/prefect/tasks.py
index 614e002877e8..71ed31e73559 100644
--- a/src/prefect/tasks.py
+++ b/src/prefect/tasks.py
@@ -809,7 +809,7 @@ def map(
 
         Must be called with at least one iterable and all iterables must be
         the same length. Any arguments that are not iterable will be treated as
-        a static value and each task run will recieve the same value.
+        a static value and each task run will receive the same value.
 
         Will create as many task runs as the length of the iterable(s) in the
         backing API and submit the task runs to the flow's task runner. This
diff --git a/src/prefect/testing/cli.py b/src/prefect/testing/cli.py
index 7aa5d2d179d6..ee2a2460463e 100644
--- a/src/prefect/testing/cli.py
+++ b/src/prefect/testing/cli.py
@@ -36,11 +36,11 @@ def check_contains(cli_result: Result, content: str, should_contain: bool):
     if should_contain:
         assert (
             content in output
-        ), f"Desired contents {display_content!r} not found in CLI ouput"
+        ), f"Desired contents {display_content!r} not found in CLI output"
     else:
         assert (
             content not in output
-        ), f"Undesired contents {display_content!r} found in CLI ouput"
+        ), f"Undesired contents {display_content!r} found in CLI output"
 
 
 def invoke_and_assert(
diff --git a/src/prefect/testing/fixtures.py b/src/prefect/testing/fixtures.py
index 518382ddf977..5ca7681e72b0 100644
--- a/src/prefect/testing/fixtures.py
+++ b/src/prefect/testing/fixtures.py
@@ -125,7 +125,7 @@ def mock_anyio_sleep(monkeypatch):
     Mock sleep used to not actually sleep but to set the current time to now + sleep
     delay seconds while still yielding to other tasks in the event loop.
 
-    Provides "assert_sleeps_for" context manager which asserts a sleep time occured
+    Provides "assert_sleeps_for" context manager which asserts a sleep time occurred
     within the context while using the actual runtime of the context as a tolerance.
     """
     original_now = pendulum.now
diff --git a/src/prefect/testing/standard_test_suites/task_runners.py b/src/prefect/testing/standard_test_suites/task_runners.py
index 2da36a032247..8d197eddd2c4 100644
--- a/src/prefect/testing/standard_test_suites/task_runners.py
+++ b/src/prefect/testing/standard_test_suites/task_runners.py
@@ -172,7 +172,7 @@ def test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type(
         @task
         def foo():
             time.sleep(self.get_sleep_time())
-            # Yield again in case the sleep started before the other thread was aready
+            # Yield again in case the sleep started before the other thread was already
             time.sleep(0)
             tmp_file.write_text("foo")
 
diff --git a/src/prefect/testing/utilities.py b/src/prefect/testing/utilities.py
index 498981d1d533..68fccaa949c7 100644
--- a/src/prefect/testing/utilities.py
+++ b/src/prefect/testing/utilities.py
@@ -71,7 +71,7 @@ def _mock_set_magics(self):
 
         def _get_child_mock(self, **kw):
             """Patch to return async mocks for async methods"""
-            # This implemetation copied from unittest in Python 3.8
+            # This implementation copied from unittest in Python 3.8
             _new_name = kw.get("_new_name")
             if _new_name in self.__dict__.get("_spec_asyncs", {}):
                 return AsyncMock(**kw)
diff --git a/src/prefect/utilities/asyncutils.py b/src/prefect/utilities/asyncutils.py
index ebe740061595..92c7d31798f6 100644
--- a/src/prefect/utilities/asyncutils.py
+++ b/src/prefect/utilities/asyncutils.py
@@ -408,7 +408,7 @@ async def gather(*calls: Callable[[], Coroutine[Any, Any, T]]) -> List[T]:
     """
     Run calls concurrently and gather their results.
 
-    Unlike `asyncio.gather` this expects to receieve _callables_ not _coroutines_.
+    Unlike `asyncio.gather` this expects to receive _callables_ not _coroutines_.
     This matches `anyio` semantics.
     """
     keys = []
diff --git a/src/prefect/utilities/callables.py b/src/prefect/utilities/callables.py
index eaee34e2725f..d59685c5a449 100644
--- a/src/prefect/utilities/callables.py
+++ b/src/prefect/utilities/callables.py
@@ -29,7 +29,7 @@ def get_call_parameters(
 ) -> Dict[str, Any]:
     """
     Bind a call to a function to get parameter/value mapping. Default values on the
-    signature will be included if not overriden.
+    signature will be included if not overridden.
 
     Raises a ParameterBindError if the arguments/kwargs are not valid for the function
     """
diff --git a/src/prefect/utilities/collections.py b/src/prefect/utilities/collections.py
index 80fd0f279209..d040d612b23b 100644
--- a/src/prefect/utilities/collections.py
+++ b/src/prefect/utilities/collections.py
@@ -232,7 +232,7 @@ def visit_collection(
     `visit_fn` can be used to alter the element if `return_data` is set.
 
     Note that when using `return_data` a copy of each collection is created to avoid
-    mutating the original object. This may have significant performance penalities and
+    mutating the original object. This may have significant performance penalties and
     should only be used if you intend to transform the collection.
 
     Supported types:
diff --git a/src/prefect/utilities/dockerutils.py b/src/prefect/utilities/dockerutils.py
index 09b942583fcf..42509ba5da5e 100644
--- a/src/prefect/utilities/dockerutils.py
+++ b/src/prefect/utilities/dockerutils.py
@@ -502,7 +502,7 @@ def parse_image_tag(name: str) -> Tuple[str, Optional[str]]:
 def format_outlier_version_name(version: str):
     """
     Formats outlier docker version names to pass `packaging.version.parse` validation
-    - Current cases are simple, but creats stub for more complicated formatting if eventually needed.
+    - Current cases are simple, but creates stub for more complicated formatting if eventually needed.
     - Example outlier versions that throw a parsing exception:
       - "20.10.0-ce" (variant of community edition label)
       - "20.10.0-ee" (variant of enterprise edition label)
diff --git a/src/prefect/utilities/services.py b/src/prefect/utilities/services.py
index 9e76b346f4b8..ec71409b1920 100644
--- a/src/prefect/utilities/services.py
+++ b/src/prefect/utilities/services.py
@@ -90,13 +90,13 @@ async def critical_service_loop(
         # distribution of errors. We are trying to determine which of those two worlds
         # we are currently experiencing.  We compare the likelihood that we'd draw N
         # consecutive errors from each.  In the everything-is-fine distribution, that
-        # would be a very low-probability occurrance, but in the everything-is-on-fire
-        # distribution, that is a high-probability occurrance.
+        # would be a very low-probability occurrence, but in the everything-is-on-fire
+        # distribution, that is a high-probability occurrence.
         #
         # Remarkably, we only need to look back for a small number of consecutive
         # errors to have reasonable confidence that this is indeed an anomaly.
         # @anticorrelator and @chrisguidry estimated that we should only need to look
-        # back for 3 consectutive errors.
+        # back for 3 consecutive errors.
         if not any(track_record):
             # We've failed enough times to be sure something is wrong, the writing is
             # on the wall.  Let's explain what we've seen and exit.
diff --git a/tests/_internal/concurrency/test_waiters.py b/tests/_internal/concurrency/test_waiters.py
index c6abb203ddb8..18f9ca719930 100644
--- a/tests/_internal/concurrency/test_waiters.py
+++ b/tests/_internal/concurrency/test_waiters.py
@@ -118,7 +118,7 @@ def test_async_waiter_done_callbacks():
 
 def test_sync_waiter_timeout_in_worker_thread():
     """
-    In this test, a timeout is raised due to a slow call that is occuring on the worker
+    In this test, a timeout is raised due to a slow call that is occurring on the worker
     thread.
     """
     done_callback = Call.new(identity, 1)
diff --git a/tests/agent/test_agent_run_submission.py b/tests/agent/test_agent_run_submission.py
index 50f1335c3751..56279e03ac0a 100644
--- a/tests/agent/test_agent_run_submission.py
+++ b/tests/agent/test_agent_run_submission.py
@@ -781,7 +781,7 @@ def raise_value_error():
             ).dict()
         )
         agent.logger.exception.assert_called_once_with(
-            f"An error occured while monitoring flow run '{flow_run.id}'. "
+            f"An error occurred while monitoring flow run '{flow_run.id}'. "
             "The flow run will not be marked as failed, but an issue may have "
             "occurred."
         )
diff --git a/tests/cli/deployment/test_deployment_build.py b/tests/cli/deployment/test_deployment_build.py
index 312de726c3e4..fa7790abd94d 100644
--- a/tests/cli/deployment/test_deployment_build.py
+++ b/tests/cli/deployment/test_deployment_build.py
@@ -414,7 +414,7 @@ def test_mixing_parameter_overrides(self, patch_import, tmp_path):
                 "--params",
                 '{"which": "parameter"}',
                 "--param",
-                "shouldbe:used",
+                "should-be:used",
             ],
             expected_code=1,
             temp_dir=tmp_path,
diff --git a/tests/cli/test_cloud.py b/tests/cli/test_cloud.py
index 41a8efc22a58..76d8516ee913 100644
--- a/tests/cli/test_cloud.py
+++ b/tests/cli/test_cloud.py
@@ -1349,7 +1349,7 @@ def test_update_webhook(respx_mock):
     new_webhook_name = "wowza-webhooks"
     existing_webhook = {
         "name": "this will change",
-        "description": "this wont change",
+        "description": "this won't change",
         "template": "neither will this",
     }
     respx_mock.get(f"{foo_workspace.api_url()}/webhooks/{webhook_id}").mock(
diff --git a/tests/cli/test_flow_run.py b/tests/cli/test_flow_run.py
index 67b40163339a..bc71c11677c2 100644
--- a/tests/cli/test_flow_run.py
+++ b/tests/cli/test_flow_run.py
@@ -244,7 +244,7 @@ async def test_non_terminal_states_set_to_cancelling(self, prefect_client, state
             ],
             expected_code=0,
             expected_output_contains=(
-                f"Flow run '{before.id}' was succcessfully scheduled for cancellation."
+                f"Flow run '{before.id}' was successfully scheduled for cancellation."
             ),
         )
         after = await prefect_client.read_flow_run(before.id)
@@ -274,7 +274,7 @@ async def test_scheduled_states_set_to_cancelled(self, prefect_client, state):
             ],
             expected_code=0,
             expected_output_contains=(
-                f"Flow run '{before.id}' was succcessfully scheduled for cancellation."
+                f"Flow run '{before.id}' was successfully scheduled for cancellation."
             ),
         )
         after = await prefect_client.read_flow_run(before.id)
diff --git a/tests/events/conftest.py b/tests/events/conftest.py
index 9b1d185352ec..30360ee28aab 100644
--- a/tests/events/conftest.py
+++ b/tests/events/conftest.py
@@ -26,7 +26,7 @@ def example_event_1() -> Event:
 @pytest.fixture
 def example_event_2() -> Event:
     return Event(
-        event="wonderous.things.happened",
+        event="wondrous.things.happened",
         resource={"prefect.resource.id": "something-valuable"},
     )
 
diff --git a/tests/fixtures/logging.py b/tests/fixtures/logging.py
index 8a837f4e33ac..d70cab06ec02 100644
--- a/tests/fixtures/logging.py
+++ b/tests/fixtures/logging.py
@@ -22,7 +22,7 @@ def enable_api_log_handler_if_marked(request):
     The `APILogHandler` is disabled during testing by default to reduce overhead.
 
     Test functions or classes can be marked with `@pytest.mark.enable_api_log_handler`
-    to indicate that they need the handler to be reenabled because they are testing
+    to indicate that they need the handler to be re-enabled because they are testing
     its functionality.
     """
     marker = request.node.get_closest_marker("enable_api_log_handler")
diff --git a/tests/infrastructure/test_docker_container.py b/tests/infrastructure/test_docker_container.py
index 03e416f1194c..f5463c1c6e21 100644
--- a/tests/infrastructure/test_docker_container.py
+++ b/tests/infrastructure/test_docker_container.py
@@ -792,7 +792,7 @@ def test_container_metadata(docker: "DockerClient"):
 
 @pytest.mark.service("docker")
 def test_container_name_collision(docker: "DockerClient"):
-    # Generate a unique base name to avoid collissions with existing images
+    # Generate a unique base name to avoid collisions with existing images
     base_name = uuid.uuid4().hex
 
     container = DockerContainer(
@@ -831,7 +831,7 @@ def test_stream_container_logs(capsys, mock_docker_client):
     mock_container = mock_docker_client.containers.get.return_value
     mock_container.logs = MagicMock(return_value=[b"hello", b"world"])
 
-    DockerContainer(command=["doesnt", "matter"]).run()
+    DockerContainer(command=["doesn't", "matter"]).run()
 
     captured = capsys.readouterr()
     assert "hello\nworld\n" in captured.out
@@ -846,7 +846,7 @@ def test_logs_warning_when_container_marked_for_removal(caplog, mock_docker_clie
     mock_container.logs = MagicMock(side_effect=docker.errors.APIError(warning))
 
     DockerContainer(
-        command=["doesnt", "matter"],
+        command=["doesn't", "matter"],
     ).run()
 
     assert "Docker container fake-name was marked for removal" in caplog.text
@@ -857,11 +857,11 @@ def test_logs_when_unexpected_docker_error(caplog, mock_docker_client):
     mock_container.logs = MagicMock(side_effect=docker.errors.APIError("..."))
 
     DockerContainer(
-        command=["doesnt", "matter"],
+        command=["doesn't", "matter"],
     ).run()
 
     assert (
-        "An unexpected Docker API error occured while streaming output from container"
+        "An unexpected Docker API error occurred while streaming output from container"
         " fake-name."
         in caplog.text
     )
diff --git a/tests/infrastructure/test_kubernetes_job.py b/tests/infrastructure/test_kubernetes_job.py
index 086b638ebee1..b70ef1560781 100644
--- a/tests/infrastructure/test_kubernetes_job.py
+++ b/tests/infrastructure/test_kubernetes_job.py
@@ -1054,7 +1054,7 @@ def mock_log_stream(*args, **kwargs):
         assert f"test {i}" not in stdout
 
 
-@pytest.mark.flaky  # Rarely, the sleep times we check for do not fit within the tolerences
+@pytest.mark.flaky  # Rarely, the sleep times we check for do not fit within the tolerances
 def test_watch_timeout_is_restarted_until_job_is_complete(
     mock_k8s_client,
     mock_watch,
@@ -1313,7 +1313,7 @@ def test_validates_for_a_job_with_incompatible_values(self):
             {
                 "loc": ("job",),
                 "msg": (
-                    "Job has incompatble values for the following attributes: "
+                    "Job has incompatible values for the following attributes: "
                     "/apiVersion must have value 'batch/v1', "
                     "/kind must have value 'Job'"
                 ),
diff --git a/tests/results/test_task_results.py b/tests/results/test_task_results.py
index afe2d9aa2df8..1d3e8dffdc74 100644
--- a/tests/results/test_task_results.py
+++ b/tests/results/test_task_results.py
@@ -161,7 +161,7 @@ def bar():
     ).state
     assert not api_state.data.has_cached_object()
     assert await api_state.result() == 1
-    # After retrieval from the API, the "cache_result_in_memory" settig is dropped
+    # After retrieval from the API, the "cache_result_in_memory" setting is dropped
     # and caching is enabled by default
     assert api_state.data.has_cached_object()
     assert await api_state.result() == 1
diff --git a/tests/server/api/test_flow_runs.py b/tests/server/api/test_flow_runs.py
index 6a0c5fca4b02..c9931f9f6dba 100644
--- a/tests/server/api/test_flow_runs.py
+++ b/tests/server/api/test_flow_runs.py
@@ -20,7 +20,7 @@ async def test_create_flow_run(self, flow, client, session):
             "/flow_runs/",
             json=actions.FlowRunCreate(
                 flow_id=flow.id,
-                name="orange you glad i didnt say yellow salamander",
+                name="orange you glad i didn't say yellow salamander",
                 state=states.Pending(),
             ).dict(json_compatible=True),
         )
@@ -29,7 +29,7 @@ async def test_create_flow_run(self, flow, client, session):
         assert response.json()["id"]
         assert response.json()["state"]["type"] == "PENDING"
         assert (
-            response.json()["name"] == "orange you glad i didnt say yellow salamander"
+            response.json()["name"] == "orange you glad i didn't say yellow salamander"
         )
 
         flow_run = await models.flow_runs.read_flow_run(
diff --git a/tests/server/database/test_queries.py b/tests/server/database/test_queries.py
index 870d3a1af0fd..819c10bc6520 100644
--- a/tests/server/database/test_queries.py
+++ b/tests/server/database/test_queries.py
@@ -127,7 +127,7 @@ async def test_get_runs_in_queue_limit_sorts_correctly(
         self, session, db, deployment_1
     ):
         """
-        Tests that the query sorts by scheduled time correctly; the unit tests with a small nubmer of runs
+        Tests that the query sorts by scheduled time correctly; the unit tests with a small number of runs
         can return the correct order even though no sort is applied.
 
         https://github.com/PrefectHQ/prefect/pull/7457
@@ -207,7 +207,7 @@ async def test_use_query_to_filter_deployments(
     async def test_query_skips_locked(self, db):
         """Concurrent queries should not both receive runs"""
         if db.database_config.connection_url.startswith("sqlite"):
-            pytest.skip("FOR UDPATE SKIP LOCKED is not supported on SQLite")
+            pytest.skip("FOR UPDATE SKIP LOCKED is not supported on SQLite")
 
         query = db.queries.get_scheduled_flow_runs_from_work_queues(db=db)
 
@@ -692,7 +692,7 @@ async def test_concurrent_reads(self, db):
         before the limit is applied.
         """
         if db.database_config.connection_url.startswith("sqlite"):
-            pytest.skip("FOR UDPATE SKIP LOCKED is not supported on SQLite")
+            pytest.skip("FOR UPDATE SKIP LOCKED is not supported on SQLite")
 
         db.queries.get_scheduled_flow_runs_from_work_queues(db=db)
 
@@ -715,7 +715,7 @@ async def test_concurrent_reads_with_overlap(self, db):
         sees 18 runs and the second query sees the others.
         """
         if db.database_config.connection_url.startswith("sqlite"):
-            pytest.skip("FOR UDPATE SKIP LOCKED is not supported on SQLite")
+            pytest.skip("FOR UPDATE SKIP LOCKED is not supported on SQLite")
 
         db.queries.get_scheduled_flow_runs_from_work_queues(db=db)
 
diff --git a/tests/server/models/deprecated/test_work_queues.py b/tests/server/models/deprecated/test_work_queues.py
index 39d1bafd9e81..04c38ee1e9f3 100644
--- a/tests/server/models/deprecated/test_work_queues.py
+++ b/tests/server/models/deprecated/test_work_queues.py
@@ -376,7 +376,7 @@ async def test_get_runs_in_work_queue_filters_on_deployment_ids(
         bad_deployment_work_queue = await models.work_queues.create_work_queue(
             session=session,
             work_queue=schemas.actions.WorkQueueCreate(
-                name="Work Queue for Deployment that doesnt exist",
+                name="Work Queue for Deployment that doesn't exist",
                 filter=schemas.core.QueueFilter(deployment_ids=[uuid4()]),
             ),
         )
@@ -393,7 +393,7 @@ async def test_get_runs_in_work_queue_uses_union_of_filter_criteria(self, sessio
         conflicting_filter_work_queue = await models.work_queues.create_work_queue(
             session=session,
             work_queue=schemas.actions.WorkQueueCreate(
-                name="Work Queue for Deployment that doesnt exist",
+                name="Work Queue for Deployment that doesn't exist",
                 filter=schemas.core.QueueFilter(
                     deployment_ids=[uuid4()], tags=["tb12"]
                 ),
diff --git a/tests/server/models/test_deployments.py b/tests/server/models/test_deployments.py
index 41fa3a85751f..f569d4aa86e7 100644
--- a/tests/server/models/test_deployments.py
+++ b/tests/server/models/test_deployments.py
@@ -910,7 +910,7 @@ async def test_run_details_are_applied_to_scheduled_runs(self, deployment, sessi
     async def test_scheduling_multiple_batches_correctly_updates_runs(
         self, session, deployment, flow_function, flow, db
     ):
-        # ensures that updating flow run states works correctly and doesnt set
+        # ensures that updating flow run states works correctly and doesn't set
         # any to None inadvertently
         deployment_2 = await models.deployments.create_deployment(
             session=session,
diff --git a/tests/server/models/test_filters.py b/tests/server/models/test_filters.py
index e6360b840cb0..a26871b9f7f4 100644
--- a/tests/server/models/test_filters.py
+++ b/tests/server/models/test_filters.py
@@ -409,7 +409,7 @@ async def test_models_read(self, session, kwargs, expected):
     async def test_api_count(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
 
-        repsonse = await client.post(
+        response = await client.post(
             "/flows/count",
             json=json.loads(
                 json.dumps(
@@ -418,13 +418,13 @@ async def test_api_count(self, client, kwargs, expected):
                 )
             ),
         )
-        assert repsonse.json() == expected
+        assert response.json() == expected
 
     @pytest.mark.parametrize("kwargs,expected", params)
     async def test_api_read(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
 
-        repsonse = await client.post(
+        response = await client.post(
             "/flows/filter",
             json=json.loads(
                 json.dumps(
@@ -433,7 +433,7 @@ async def test_api_read(self, client, kwargs, expected):
                 )
             ),
         )
-        assert len({r["id"] for r in repsonse.json()}) == expected
+        assert len({r["id"] for r in response.json()}) == expected
 
 
 class TestCountFlowRunModels:
@@ -632,19 +632,19 @@ async def test_models_read(self, session, kwargs, expected):
     async def test_api_count(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
 
-        repsonse = await client.post(
+        response = await client.post(
             "/flow_runs/count",
             json=json.loads(
                 json.dumps(adjusted_kwargs, default=pydantic.json.pydantic_encoder)
             ),
         )
-        assert repsonse.json() == expected
+        assert response.json() == expected
 
     @pytest.mark.parametrize("kwargs,expected", params)
     async def test_api_read(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
 
-        repsonse = await client.post(
+        response = await client.post(
             "/flow_runs/filter",
             json=json.loads(
                 json.dumps(
@@ -653,7 +653,7 @@ async def test_api_read(self, client, kwargs, expected):
                 )
             ),
         )
-        assert len({r["id"] for r in repsonse.json()}) == expected
+        assert len({r["id"] for r in response.json()}) == expected
 
 
 class TestCountTaskRunsModels:
@@ -795,19 +795,19 @@ async def test_models_read(self, session, kwargs, expected):
     @pytest.mark.parametrize("kwargs,expected", params)
     async def test_api_count(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
-        repsonse = await client.post(
+        response = await client.post(
             "/task_runs/count",
             json=json.loads(
                 json.dumps(adjusted_kwargs, default=pydantic.json.pydantic_encoder)
             ),
         )
-        assert repsonse.json() == expected
+        assert response.json() == expected
 
     @pytest.mark.parametrize("kwargs,expected", params)
     async def test_api_read(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
 
-        repsonse = await client.post(
+        response = await client.post(
             "/task_runs/filter",
             json=json.loads(
                 json.dumps(
@@ -816,7 +816,7 @@ async def test_api_read(self, client, kwargs, expected):
                 )
             ),
         )
-        assert len({r["id"] for r in repsonse.json()}) == expected
+        assert len({r["id"] for r in response.json()}) == expected
 
 
 class TestCountDeploymentModels:
@@ -973,7 +973,7 @@ async def test_models_read(self, session, kwargs, expected):
     async def test_api_count(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
 
-        repsonse = await client.post(
+        response = await client.post(
             "/deployments/count",
             json=json.loads(
                 json.dumps(
@@ -982,13 +982,13 @@ async def test_api_count(self, client, kwargs, expected):
                 )
             ),
         )
-        assert repsonse.json() == expected
+        assert response.json() == expected
 
     @pytest.mark.parametrize("kwargs,expected", params)
     async def test_api_read(self, client, kwargs, expected):
         adjusted_kwargs = adjust_kwargs_for_client(kwargs)
 
-        repsonse = await client.post(
+        response = await client.post(
             "/deployments/filter",
             json=json.loads(
                 json.dumps(
@@ -997,4 +997,4 @@ async def test_api_read(self, client, kwargs, expected):
                 )
             ),
         )
-        assert len({r["id"] for r in repsonse.json()}) == expected
+        assert len({r["id"] for r in response.json()}) == expected
diff --git a/tests/server/models/test_flow_run_states.py b/tests/server/models/test_flow_run_states.py
index c108432acdf7..f0e2d9ecbbd9 100644
--- a/tests/server/models/test_flow_run_states.py
+++ b/tests/server/models/test_flow_run_states.py
@@ -38,7 +38,7 @@ async def test_locks_row_for_update(self, db, flow_run):
 
             - Set to SCHEDULED
             - Set to PENDING
-                - Before commiting, wait for session 2 to begin orchestrating
+                - Before committing, wait for session 2 to begin orchestrating
 
         Session 2:
 
diff --git a/tests/server/models/test_flow_runs.py b/tests/server/models/test_flow_runs.py
index d937ecbbc6a8..f91fea9a44c5 100644
--- a/tests/server/models/test_flow_runs.py
+++ b/tests/server/models/test_flow_runs.py
@@ -110,7 +110,7 @@ async def test_create_multiple_flow_runs(self, flow, session):
     # the sqlalchemy session will (correctly) recognize that a new object was
     # added to it with the same primary key as an existing object, and emit a
     # warning. Because that is the situation we want to test for, we filter the
-    # warning to avoid unecessary noise.
+    # warning to avoid unnecessary noise.
     @pytest.mark.filterwarnings(
         "ignore: New instance .* conflicts with persistent instance"
     )
diff --git a/tests/server/orchestration/test_rules.py b/tests/server/orchestration/test_rules.py
index 2f99295825e8..68754fb04cdf 100644
--- a/tests/server/orchestration/test_rules.py
+++ b/tests/server/orchestration/test_rules.py
@@ -773,7 +773,7 @@ async def cleanup(self, initial_state, validated_state, context):
             proposed_state=proposed_state,
         )
 
-        # an ExitStack is a python builtin contstruction that allows us to
+        # an ExitStack is a python builtin construction that allows us to
         # nest an arbitrary number of contexts (and therefore, rules), in this test
         # we'll enter the contexts one by one so we can follow what's happening
         async with contextlib.AsyncExitStack() as stack:
@@ -921,7 +921,7 @@ async def cleanup(self, initial_state, validated_state, context):
             proposed_state=proposed_state,
         )
 
-        # an ExitStack is a python builtin contstruction that allows us to
+        # an ExitStack is a python builtin construction that allows us to
         # nest an arbitrary number of contexts (and therefore, rules), in this test
         # we'll enter the contexts one by one so we can follow what's happening
         async with contextlib.AsyncExitStack() as stack:
diff --git a/tests/server/services/test_scheduler.py b/tests/server/services/test_scheduler.py
index fb3f4e3052b4..fe7c570972fa 100644
--- a/tests/server/services/test_scheduler.py
+++ b/tests/server/services/test_scheduler.py
@@ -268,7 +268,7 @@ async def test_schedules_runs_for_recently_created_deployments(
     async def test_schedules_runs_for_recently_updated_deployments(
         self, deployment, session, db
     ):
-        # artifically move the created time back (updated time will still be recent)
+        # artificially move the created time back (updated time will still be recent)
         await session.execute(
             sa.update(db.Deployment)
             .where(db.Deployment.id == deployment.id)
@@ -294,7 +294,7 @@ async def test_schedules_runs_for_recently_updated_deployments(
     async def test_schedules_no_runs_for_deployments_updated_a_while_ago(
         self, deployment, session, db
     ):
-        # artifically move the updated time back
+        # artificially move the updated time back
         await session.execute(
             sa.update(db.Deployment)
             .where(db.Deployment.id == deployment.id)
diff --git a/tests/test_context.py b/tests/test_context.py
index 0547273e144e..6817487dc114 100644
--- a/tests/test_context.py
+++ b/tests/test_context.py
@@ -184,7 +184,7 @@ def bar():
                 ),
                 parameters={"foo": "bar"},
             ) as task_ctx:
-                assert get_run_context() is task_ctx, "Task context takes precendence"
+                assert get_run_context() is task_ctx, "Task context takes precedence"
 
             assert (
                 get_run_context() is flow_ctx
diff --git a/tests/test_deployments.py b/tests/test_deployments.py
index 725714e1903e..288b3ba24ad7 100644
--- a/tests/test_deployments.py
+++ b/tests/test_deployments.py
@@ -348,7 +348,7 @@ async def test_build_from_flow_manages_docker_specific_path(self, flow_function)
         )
         assert d.path == "/opt/prefect/flows"
 
-        # can be overriden
+        # can be overridden
         d = await Deployment.build_from_flow(
             flow=flow_function,
             name="foo",
diff --git a/tests/test_engine.py b/tests/test_engine.py
index abb331c0b98b..9e8d6d5bb842 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -787,7 +787,7 @@ def flaky_function():
         )
 
         # Actually run the task
-        # this task should sleep for a total of 17 seconds across all conifgured retries
+        # this task should sleep for a total of 17 seconds across all configured retries
         with mock_anyio_sleep.assert_sleeps_for(17):
             state = await orchestrate_task_run(
                 task=flaky_function,
diff --git a/tests/test_flows.py b/tests/test_flows.py
index 4c74f2ab0033..8b4b473dada9 100644
--- a/tests/test_flows.py
+++ b/tests/test_flows.py
@@ -2025,7 +2025,7 @@ def parent_flow():
         assert parent_flow() == "hello"
         assert flow_run_count == 2
         assert child_flow_run_count == 2, "Child flow should run again"
-        assert child_task_run_count == 2, "Child taks should run again with child flow"
+        assert child_task_run_count == 2, "Child tasks should run again with child flow"
 
     def test_flow_retry_with_error_in_flow_and_one_failed_task_with_retries(self):
         task_run_retry_count = 0
diff --git a/tests/test_logging.py b/tests/test_logging.py
index 8786b6a3f9b9..62d23a43695b 100644
--- a/tests/test_logging.py
+++ b/tests/test_logging.py
@@ -602,7 +602,7 @@ def test_does_not_raise_when_logger_outside_of_run_context_with_warn_setting(
             updates={PREFECT_LOGGING_TO_API_WHEN_MISSING_FLOW: "warn"},
         ):
             # NOTE: We use `raises` instead of `warns` because pytest will otherwise
-            #       capture the warning call and skip checing that we use it correctly
+            #       capture the warning call and skip checking that we use it correctly
             #       See https://github.com/pytest-dev/pytest/issues/9288
             with pytest.raises(
                 UserWarning,
@@ -783,7 +783,7 @@ async def test_logs_are_sent_immediately_when_stopped(
 
         assert (
             end_time - start_time
-        ) < 5  # An arbitary time less than the 10s interval
+        ) < 5  # An arbitrary time less than the 10s interval
 
         logs = await prefect_client.read_logs()
         assert len(logs) == 2
@@ -801,7 +801,7 @@ async def test_logs_are_sent_immediately_when_flushed(
 
         assert (
             end_time - start_time
-        ) < 5  # An arbitary time less than the 10s interval
+        ) < 5  # An arbitrary time less than the 10s interval
 
         logs = await prefect_client.read_logs()
         assert len(logs) == 2
@@ -1259,7 +1259,7 @@ def test_disable_run_logger(caplog):
     @task
     def task_with_run_logger():
         logger = get_run_logger()
-        logger.critical("wont show")
+        logger.critical("won't show")
         return 42
 
     flow_run_logger = get_logger("prefect.flow_run")
@@ -1274,7 +1274,7 @@ def task_with_run_logger():
 
     assert not flow_run_logger.disabled
     assert task_run_logger.disabled  # was already disabled beforehand
-    assert caplog.record_tuples == [("null", logging.CRITICAL, "wont show")]
+    assert caplog.record_tuples == [("null", logging.CRITICAL, "won't show")]
 
 
 def test_patch_print_writes_to_stdout_without_run_context(caplog, capsys):
diff --git a/tests/utilities/test_importtools.py b/tests/utilities/test_importtools.py
index 76c1531e4534..202e3259708e 100644
--- a/tests/utilities/test_importtools.py
+++ b/tests/utilities/test_importtools.py
@@ -76,11 +76,11 @@ def test_to_and_from_qualified_name_roundtrip(obj):
 @pytest.fixture
 def pop_docker_module():
     # Allows testing of `lazy_import` on a clean sys
-    orginal = sys.modules.pop("docker")
+    original = sys.modules.pop("docker")
     try:
         yield
     finally:
-        sys.modules["docker"] = orginal
+        sys.modules["docker"] = original
 
 
 @pytest.mark.usefixtures("pop_docker_module")
diff --git a/versioneer.py b/versioneer.py
index e283ecbf1a62..ac6490806e5a 100644
--- a/versioneer.py
+++ b/versioneer.py
@@ -728,7 +728,7 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, runner=run_command):
         # TAG-NUM-gHEX
         mo = re.search(r'^(.+)-(\d+)-g([0-9a-f]+)$', git_describe)
         if not mo:
-            # unparseable. Maybe git-describe is misbehaving?
+            # unparsable. Maybe git-describe is misbehaving?
             pieces["error"] = ("unable to parse git-describe output: '%%s'"
                                %% describe_out)
             return pieces
@@ -1239,7 +1239,7 @@ def git_pieces_from_vcs(tag_prefix, root, verbose, runner=run_command):
         # TAG-NUM-gHEX
         mo = re.search(r"^(.+)-(\d+)-g([0-9a-f]+)$", git_describe)
         if not mo:
-            # unparseable. Maybe git-describe is misbehaving?
+            # unparsable. Maybe git-describe is misbehaving?
             pieces["error"] = "unable to parse git-describe output: '%s'" % describe_out
             return pieces