Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/server.py: 53%
568 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 02:04 +0000
1"""
2Defines the Prefect REST API FastAPI app.
3"""
5from __future__ import annotations
7import asyncio
8import atexit
9import base64
10import contextlib
11import gc
12import hmac
13import importlib.metadata
14import logging
15import mimetypes
16import os
17import random
18import shutil
19import socket
20import sqlite3
21import subprocess
22import sys
23import time
24from contextlib import AsyncExitStack, asynccontextmanager
25from datetime import timedelta
26from functools import wraps
27from hashlib import sha256
28from typing import Any, AsyncGenerator, Awaitable, Callable, Optional
30import anyio
31import asyncpg
32import httpx
33import sqlalchemy as sa
34import sqlalchemy.exc
35import sqlalchemy.orm.exc
36from docket import Docket
37from fastapi import Depends, FastAPI, Request, Response
38from fastapi.encoders import jsonable_encoder
39from fastapi.exceptions import RequestValidationError
40from fastapi.middleware.cors import CORSMiddleware
41from fastapi.middleware.gzip import GZipMiddleware
42from fastapi.openapi.utils import get_openapi
43from fastapi.responses import JSONResponse, RedirectResponse
44from fastapi.staticfiles import StaticFiles
45from packaging.version import Version
46from typing_extensions import Self
48import prefect
49import prefect.server.api as api
50import prefect.settings
51from prefect._internal.compatibility.starlette import status
52from prefect._internal.observability import configure_logfire
53from prefect.client.constants import SERVER_API_VERSION
54from prefect.locking._filelock import FileLock
55from prefect.logging import get_logger
56from prefect.server.api._ui_static import UIBundle, UIVersion, log_ui_static_copy_error
57from prefect.server.api.background_workers import background_worker
58from prefect.server.api.dependencies import EnforceMinimumAPIVersion
59from prefect.server.exceptions import ObjectNotFoundError
60from prefect.server.schemas.ui import UISettings
61from prefect.server.services.base import RunInEphemeralServers, RunInWebservers, Service
62from prefect.server.utilities.database import get_dialect
63from prefect.settings import (
64 PREFECT_API_DATABASE_CONNECTION_URL,
65 PREFECT_API_LOG_RETRYABLE_ERRORS,
66 PREFECT_DEBUG_MODE,
67 PREFECT_MEMO_STORE_PATH,
68 PREFECT_MEMOIZE_BLOCK_AUTO_REGISTRATION,
69 PREFECT_SERVER_API_BASE_PATH,
70 PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS,
71 PREFECT_UI_SERVE_BASE,
72 get_current_settings,
73)
74from prefect.utilities.hashing import hash_objects
76logfire: Any | None = configure_logfire()
78# FastAPI < 0.137 copies routes when including a router; 0.137+ keeps a
79# reference to the original via _IncludedRouter.
80_FASTAPI_COPIES_ROUTES_ON_INCLUDE: bool = Version(
81 importlib.metadata.version("fastapi")
82) < Version("0.137.0")
84TITLE = "Prefect Server"
85API_TITLE = "Prefect REST API"
86UI_TITLE = "Prefect REST API UI"
87API_VERSION: str = prefect.__version__
88# migrations should run only once per app start; the ephemeral API can potentially
89# create multiple apps in a single process
90LIFESPAN_RAN_FOR_APP: set[Any] = set()
92logger: "logging.Logger" = get_logger("server")
94enforce_minimum_version: EnforceMinimumAPIVersion = EnforceMinimumAPIVersion(
95 # this should be <= SERVER_API_VERSION; clients that send
96 # a version header under this value will be rejected
97 minimum_api_version="0.8.0",
98 logger=logger,
99)
102API_ROUTERS = (
103 api.flows.router,
104 api.flow_runs.router,
105 api.task_runs.router,
106 api.flow_run_states.router,
107 api.task_run_states.router,
108 api.deployments.router,
109 api.saved_searches.router,
110 api.logs.router,
111 api.concurrency_limits.router,
112 api.concurrency_limits_v2.router,
113 api.block_types.router,
114 api.block_documents.router,
115 api.workers.router,
116 api.task_workers.router,
117 api.work_queues.router,
118 api.artifacts.router,
119 api.block_schemas.router,
120 api.block_capabilities.router,
121 api.collections.router,
122 api.variables.router,
123 api.csrf_token.router,
124 api.events.router,
125 api.automations.router,
126 api.templates.router,
127 api.ui.flows.router,
128 api.ui.flow_runs.router,
129 api.ui.schemas.router,
130 api.ui.task_runs.router,
131 api.admin.router,
132 api.root.router,
133)
135SQLITE_LOCKED_MSG = "database is locked"
136UI_VERSION_COOKIE_NAME = "prefect_ui_version"
137UI_VERSION_PATH_SEGMENT = "v2"
138UI_STATIC_REFERENCE_FILE_NAME = "UI_SERVE_BASE"
141class _SQLiteLockedOperationalErrorFilter(logging.Filter):
142 """Filter uvicorn error logs for retryable SQLite lock failures."""
144 def filter(self, record: logging.LogRecord) -> bool:
145 exc: BaseException | None = record.exc_info[1] if record.exc_info else None
147 if not isinstance(exc, sqlalchemy.exc.OperationalError):
148 return True
150 orig_exc = getattr(exc, "orig", None)
151 if not isinstance(orig_exc, sqlite3.OperationalError): 151 ↛ anywhereline 151 didn't jump anywhere: it always raised an exception.
152 return True
154 if getattr(orig_exc, "sqlite_errorname", None) in { 154 ↛ 160line 154 didn't jump to line 160 because the condition on line 154 was always true
155 "SQLITE_BUSY",
156 "SQLITE_BUSY_SNAPSHOT",
157 } or SQLITE_LOCKED_MSG in getattr(orig_exc, "args", []):
158 return get_current_settings().server.log_retryable_errors
160 return True
163_SQLITE_LOCKED_LOG_FILTER: _SQLiteLockedOperationalErrorFilter | None = None
166def _install_sqlite_locked_log_filter() -> None:
167 global _SQLITE_LOCKED_LOG_FILTER
169 if _SQLITE_LOCKED_LOG_FILTER is not None:
170 return
172 filter_ = _SQLiteLockedOperationalErrorFilter()
173 logging.getLogger("uvicorn.error").addFilter(filter_)
174 logging.getLogger("docket.worker").addFilter(filter_)
175 _SQLITE_LOCKED_LOG_FILTER = filter_
178def _normalize_ui_base_url(base_url: str) -> str:
179 if not base_url: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 return "/"
182 if not base_url.startswith("/"): 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 base_url = f"/{base_url}"
185 if base_url != "/": 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true
186 base_url = base_url.rstrip("/")
188 return base_url or "/"
191def _join_ui_path(base_url: str, suffix: str) -> str:
192 normalized_base_url = _normalize_ui_base_url(base_url)
193 normalized_suffix = f"/{suffix.lstrip('/')}"
195 if normalized_base_url == "/": 195 ↛ 198line 195 didn't jump to line 198 because the condition on line 195 was always true
196 return normalized_suffix
198 return f"{normalized_base_url}{normalized_suffix}"
201def _resolve_ui_base_urls(configured_base_url: str) -> tuple[str, str]:
202 normalized_base_url = _normalize_ui_base_url(configured_base_url)
203 v2_suffix = f"/{UI_VERSION_PATH_SEGMENT}"
205 if normalized_base_url == v2_suffix:
206 return "/", normalized_base_url
208 if normalized_base_url.endswith(v2_suffix): 208 ↛ 209line 208 didn't jump to line 209 because the condition on line 208 was never true
209 v1_base_url = normalized_base_url[: -len(v2_suffix)] or "/"
210 return v1_base_url, normalized_base_url
212 return normalized_base_url, _join_ui_path(
213 normalized_base_url, UI_VERSION_PATH_SEGMENT
214 )
217def _path_targets_ui(path: str, base_url: str) -> bool:
218 if base_url == "/":
219 return path.startswith("/")
221 return path == base_url or path.startswith(f"{base_url}/")
224def _relative_ui_path(path: str, base_url: str) -> str:
225 if base_url == "/":
226 return path or "/"
228 relative_path = path.removeprefix(base_url)
229 return relative_path or "/"
232def _build_ui_path(base_url: str, relative_path: str) -> str:
233 normalized_base_url = _normalize_ui_base_url(base_url)
234 normalized_relative_path = (
235 relative_path if relative_path.startswith("/") else f"/{relative_path}"
236 )
238 if normalized_base_url == "/":
239 return normalized_relative_path
241 if normalized_relative_path == "/":
242 # Preserve the trailing slash so the redirect target falls inside
243 # the bundle's mount. Starlette's `Mount` only routes requests
244 # whose path starts with `{mount}/`; a bare `{mount}` falls through
245 # to whatever else matches, which in practice is the V1 SPA mount
246 # at "/" — that returns V1's index.html under the `/v2` URL and
247 # the V1 router can't resolve the route.
248 return f"{normalized_base_url}/"
250 return f"{normalized_base_url}{normalized_relative_path}"
253def _looks_like_ui_static_bundle(static_dir: str) -> bool:
254 return os.path.exists(os.path.join(static_dir, UI_STATIC_REFERENCE_FILE_NAME))
257def _is_html_navigation(request: Request) -> bool:
258 if request.method not in {"GET", "HEAD"}:
259 return False
261 sec_fetch_mode = request.headers.get("sec-fetch-mode")
262 sec_fetch_dest = request.headers.get("sec-fetch-dest")
263 if sec_fetch_mode == "navigate" or sec_fetch_dest in {"document", "iframe"}:
264 return True
266 accept = request.headers.get("accept", "").lower()
267 return "text/html" in accept
270class RequestLimitMiddleware:
271 """
272 A middleware that limits the number of concurrent requests handled by the API.
274 This is a blunt tool for limiting SQLite concurrent writes which will cause failures
275 at high volume. Ideally, we would only apply the limit to routes that perform
276 writes.
277 """
279 def __init__(self, app: Any, limit: float):
280 self.app = app
281 self._limiter = anyio.CapacityLimiter(limit)
283 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
284 async with self._limiter:
285 await self.app(scope, receive, send)
288async def validation_exception_handler(
289 request: Request, exc: RequestValidationError
290) -> JSONResponse:
291 """Provide a detailed message for request validation errors."""
292 return JSONResponse(
293 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
294 content=jsonable_encoder(
295 {
296 "exception_message": "Invalid request received.",
297 "exception_detail": exc.errors(),
298 "request_body": exc.body,
299 }
300 ),
301 )
304async def integrity_exception_handler(request: Request, exc: Exception) -> JSONResponse:
305 """Capture database integrity errors."""
306 logger.error("Encountered exception in request:", exc_info=True)
307 return JSONResponse(
308 content={
309 "detail": (
310 "Data integrity conflict. This usually means a "
311 "unique or foreign key constraint was violated. "
312 "See server logs for details."
313 )
314 },
315 status_code=status.HTTP_409_CONFLICT,
316 )
319def is_client_retryable_exception(exc: Exception) -> bool:
320 if isinstance(exc, sqlalchemy.exc.OperationalError) and isinstance( 320 ↛ 323line 320 didn't jump to line 323 because the condition on line 320 was never true
321 exc.orig, sqlite3.OperationalError
322 ):
323 if getattr(exc.orig, "sqlite_errorname", None) in {
324 "SQLITE_BUSY",
325 "SQLITE_BUSY_SNAPSHOT",
326 } or SQLITE_LOCKED_MSG in getattr(exc.orig, "args", []):
327 return True
328 else:
329 # Avoid falling through to the generic `DBAPIError` case below
330 return False
332 if isinstance(
333 exc,
334 (
335 sqlalchemy.exc.DBAPIError,
336 asyncpg.exceptions.QueryCanceledError,
337 asyncpg.exceptions.ConnectionDoesNotExistError,
338 asyncpg.exceptions.CannotConnectNowError,
339 sqlalchemy.exc.InvalidRequestError,
340 sqlalchemy.orm.exc.DetachedInstanceError,
341 ),
342 ):
343 return True
345 return False
348def replace_placeholder_string_in_files(
349 directory: str,
350 placeholder: str,
351 replacement: str,
352 allowed_extensions: list[str] | None = None,
353) -> None:
354 """
355 Recursively loops through all files in the given directory and replaces
356 a placeholder string.
357 """
358 if allowed_extensions is None: 358 ↛ 361line 358 didn't jump to line 361 because the condition on line 358 was always true
359 allowed_extensions = [".txt", ".html", ".css", ".js", ".json", ".txt"]
361 for root, _, files in os.walk(directory):
362 for file in files:
363 if any(file.endswith(ext) for ext in allowed_extensions):
364 file_path = os.path.join(root, file)
366 with open(file_path, "r", encoding="utf-8") as file:
367 file_data = file.read()
369 file_data = file_data.replace(placeholder, replacement)
371 with open(file_path, "w", encoding="utf-8") as file:
372 file.write(file_data)
375def copy_directory(directory: str, path: str) -> None:
376 os.makedirs(path, exist_ok=True)
377 for item in os.listdir(directory):
378 source = os.path.join(directory, item)
379 destination = os.path.join(path, item)
381 if os.path.isdir(source):
382 if os.path.exists(destination): 382 ↛ 383line 382 didn't jump to line 383 because the condition on line 382 was never true
383 shutil.rmtree(destination)
384 shutil.copytree(source, destination, symlinks=True)
385 # ensure copied files are writeable
386 for root, _, files in os.walk(destination):
387 for f in files:
388 os.chmod(os.path.join(root, f), 0o700)
389 else:
390 shutil.copy2(source, destination)
391 # Ensure copied file is writeable
392 os.chmod(destination, 0o700)
395async def custom_internal_exception_handler(
396 request: Request, exc: Exception
397) -> JSONResponse:
398 """
399 Log a detailed exception for internal server errors before returning.
401 Send 503 for errors clients can retry on.
402 """
403 if is_client_retryable_exception(exc):
404 if PREFECT_API_LOG_RETRYABLE_ERRORS.value(): 404 ↛ 405line 404 didn't jump to line 405 because the condition on line 404 was never true
405 logger.error("Encountered retryable exception in request:", exc_info=True)
407 return JSONResponse(
408 content={"exception_message": "Service Unavailable"},
409 status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
410 )
412 logger.error("Encountered exception in request:", exc_info=True)
414 return JSONResponse(
415 content={"exception_message": "Internal Server Error"},
416 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
417 )
420async def prefect_object_not_found_exception_handler(
421 request: Request, exc: ObjectNotFoundError
422) -> JSONResponse:
423 """Return 404 status code on object not found exceptions."""
424 return JSONResponse(
425 content={"exception_message": str(exc)}, status_code=status.HTTP_404_NOT_FOUND
426 )
429API_APP_CACHE: dict[tuple[str, str | None], FastAPI] = {}
432def create_api_app(
433 dependencies: list[Any] | None = None,
434 health_check_path: str = "/health",
435 version_check_path: str = "/version",
436 fast_api_app_kwargs: dict[str, Any] | None = None,
437 final: bool = False,
438 ignore_cache: bool = False,
439) -> FastAPI:
440 """
441 Create a FastAPI app that includes the Prefect REST API
443 Args:
444 dependencies: a list of global dependencies to add to each Prefect REST API router
445 health_check_path: the health check route path
446 fast_api_app_kwargs: kwargs to pass to the FastAPI constructor
447 final: whether this will be the last instance of the Prefect server to be
448 created in this process, so that additional optimizations may be applied
449 ignore_cache: if set, a new app will be created even if the settings and fast_api_app_kwargs match
450 an existing app in the cache
452 Returns:
453 a FastAPI app that serves the Prefect REST API
454 """
455 cache_key = (
456 prefect.settings.get_current_settings().hash_key(),
457 hash_objects(fast_api_app_kwargs) if fast_api_app_kwargs else None,
458 )
460 if cache_key in API_APP_CACHE and not ignore_cache: 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true
461 return API_APP_CACHE[cache_key]
463 fast_api_app_kwargs = fast_api_app_kwargs or {}
464 api_app = FastAPI(title=API_TITLE, **fast_api_app_kwargs)
466 if logfire: 466 ↛ 467line 466 didn't jump to line 467 because the condition on line 466 was never true
467 logfire.instrument_fastapi(api_app) # pyright: ignore
469 api_app.add_middleware(GZipMiddleware)
471 @api_app.get(health_check_path, tags=["Root"])
472 async def health_check() -> bool: # type: ignore[reportUnusedFunction]
473 return True
475 @api_app.get(version_check_path, tags=["Root"])
476 async def server_version() -> str: # type: ignore[reportUnusedFunction]
477 return SERVER_API_VERSION
479 # always include version checking
480 if dependencies is None: 480 ↛ 483line 480 didn't jump to line 483 because the condition on line 480 was always true
481 dependencies = [Depends(enforce_minimum_version)]
482 else:
483 dependencies.append(Depends(enforce_minimum_version))
485 for router in API_ROUTERS:
486 api_app.include_router(router, dependencies=dependencies)
487 if final and _FASTAPI_COPIES_ROUTES_ON_INCLUDE: 487 ↛ 497line 487 didn't jump to line 497 because the condition on line 487 was never true
488 # When including a router, older versions of FastAPI (< 0.137) copy
489 # the routes and build entirely new Pydantic models. Since Prefect
490 # does not reuse routers, we can delete the originals to reclaim
491 # ~50-55 MB of memory.
492 #
493 # FastAPI 0.137+ wraps included routers in an _IncludedRouter that
494 # references the original router for request matching, so the
495 # originals must be kept. The duplication no longer occurs in this
496 # case either, so the optimisation is unnecessary.
497 del router.routes
499 if final: 499 ↛ 502line 499 didn't jump to line 502 because the condition on line 499 was always true
500 gc.collect()
502 @api_app.middleware("http")
503 async def default_content_type(request: Request, call_next: Any): # type: ignore[reportUnusedFunction]
504 # Older Prefect clients (<3.6.19) sent JSON bodies via httpx's
505 # content= parameter, which omits the Content-Type header.
506 # FastAPI >=0.132.0 requires Content-Type: application/json to
507 # parse request bodies. Default it here for backward compat.
508 if ( 508 ↛ 513line 508 didn't jump to line 513 because the condition on line 508 was never true
509 request.method in {"POST", "PUT", "PATCH"}
510 and "content-type" not in request.headers
511 and int(request.headers.get("content-length", "0")) > 0
512 ):
513 request.scope["headers"] = [
514 *request.scope["headers"],
515 (b"content-type", b"application/json"),
516 ]
517 return await call_next(request)
519 auth_string = prefect.settings.PREFECT_SERVER_API_AUTH_STRING.value()
521 if auth_string is not None: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true
522 health_check_paths = {health_check_path, "/ready"}
524 @api_app.middleware("http")
525 async def token_validation(request: Request, call_next: Any): # type: ignore[reportUnusedFunction]
526 header_token = request.headers.get("Authorization")
528 # Allow unauthenticated health/ready probes (e.g. k8s).
529 # Use scope["path"] (not request.url.path) because url.path
530 # can be spoofed via Host header manipulation. Use exact path
531 # matching (not suffix matching) to prevent auth bypass via
532 # crafted paths like /variables/name/system-health.
533 scope = request.scope
534 app_path = scope["path"].removeprefix(scope.get("root_path", ""))
535 if app_path in health_check_paths and request.method.upper() == "GET":
536 return await call_next(request)
537 try:
538 if header_token is None:
539 return JSONResponse(
540 status_code=status.HTTP_401_UNAUTHORIZED,
541 content={"exception_message": "Unauthorized"},
542 )
543 scheme, creds = header_token.split()
544 assert scheme == "Basic"
545 decoded = base64.b64decode(creds).decode("utf-8")
546 except Exception:
547 return JSONResponse(
548 status_code=status.HTTP_401_UNAUTHORIZED,
549 content={"exception_message": "Unauthorized"},
550 )
551 if not hmac.compare_digest(decoded, auth_string):
552 return JSONResponse(
553 status_code=status.HTTP_401_UNAUTHORIZED,
554 content={"exception_message": "Unauthorized"},
555 )
556 return await call_next(request)
558 API_APP_CACHE[cache_key] = api_app
560 return api_app
563def create_ui_app(ephemeral: bool) -> FastAPI:
564 ui_app = FastAPI(title=UI_TITLE)
565 configured_default_ui: UIVersion = (
566 "v2" if prefect.settings.get_current_settings().server.ui.v2_enabled else "v1"
567 )
568 v1_base_url, v2_base_url = _resolve_ui_base_urls(PREFECT_UI_SERVE_BASE.value())
569 ui_settings_path = _join_ui_path(v1_base_url, "ui-settings")
570 static_directory_root = prefect.settings.PREFECT_UI_STATIC_DIRECTORY.value()
571 mounted_bundles: dict[UIVersion, UIBundle] = {}
573 if os.name == "nt": 573 ↛ 575line 573 didn't jump to line 575 because the condition on line 573 was never true
574 # Windows defaults to text/plain for .js files
575 mimetypes.init()
576 mimetypes.add_type("application/javascript", ".js")
578 def available_ui_versions() -> list[UIVersion]:
579 return [version for version in ("v1", "v2") if version in mounted_bundles]
581 def default_ui() -> UIVersion:
582 available_uis = available_ui_versions()
583 if configured_default_ui in available_uis:
584 return configured_default_ui
585 if available_uis:
586 return available_uis[0]
587 return configured_default_ui
589 def build_static_dir(version: UIVersion, default_subpath: str) -> str:
590 if static_directory_root: 590 ↛ 591line 590 didn't jump to line 591 because the condition on line 590 was never true
591 if version == "v1" and _looks_like_ui_static_bundle(static_directory_root):
592 return static_directory_root
593 return os.path.join(static_directory_root, version)
594 return default_subpath
596 bundles = [
597 UIBundle(
598 version="v1",
599 source_static_path=str(prefect.__ui_static_path__),
600 static_dir=build_static_dir("v1", str(prefect.__ui_static_subpath__)),
601 base_url=v1_base_url,
602 cache_key=f"v1:{prefect.__version__}:{v1_base_url}",
603 ),
604 UIBundle(
605 version="v2",
606 source_static_path=str(prefect.__ui_v2_static_path__),
607 static_dir=build_static_dir("v2", str(prefect.__ui_v2_static_subpath__)),
608 base_url=v2_base_url,
609 cache_key=f"v2:{prefect.__version__}:{v2_base_url}",
610 ),
611 ]
613 @ui_app.get(ui_settings_path)
614 def ui_settings() -> UISettings: # type: ignore[reportUnusedFunction]
615 return UISettings(
616 api_url=prefect.settings.PREFECT_UI_API_URL.value(),
617 csrf_enabled=prefect.settings.PREFECT_SERVER_CSRF_PROTECTION_ENABLED.value(),
618 auth="BASIC"
619 if prefect.settings.PREFECT_SERVER_API_AUTH_STRING.value()
620 else None,
621 flags=[],
622 default_ui=default_ui(),
623 available_uis=available_ui_versions(),
624 v1_base_url=v1_base_url if "v1" in mounted_bundles else None,
625 v2_base_url=v2_base_url if "v2" in mounted_bundles else None,
626 )
628 @ui_app.middleware("http")
629 async def redirect_to_preferred_ui(request: Request, call_next: Any): # type: ignore[reportUnusedFunction]
630 scope_path = request.scope["path"]
631 path = scope_path.removeprefix(request.scope.get("root_path", "")) or "/"
633 if (
634 not _is_html_navigation(request)
635 or not available_ui_versions()
636 or path.endswith("/ui-settings")
637 ):
638 return await call_next(request)
640 requested_ui: UIVersion | None = None
641 relative_path: str | None = None
642 if _path_targets_ui(path, v2_base_url):
643 requested_ui = "v2"
644 relative_path = _relative_ui_path(path, v2_base_url)
645 elif _path_targets_ui(path, v1_base_url): 645 ↛ 649line 645 didn't jump to line 649 because the condition on line 645 was always true
646 requested_ui = "v1"
647 relative_path = _relative_ui_path(path, v1_base_url)
649 if requested_ui is None or relative_path is None:
650 return await call_next(request)
652 if requested_ui == "v2":
653 if requested_ui in mounted_bundles:
654 return await call_next(request)
655 elif relative_path != "/":
656 return await call_next(request)
658 requested_cookie_ui = request.cookies.get(UI_VERSION_COOKIE_NAME)
659 preferred_ui = (
660 requested_cookie_ui
661 if requested_cookie_ui in available_ui_versions()
662 else default_ui()
663 )
665 if preferred_ui == requested_ui:
666 return await call_next(request)
668 redirect_path = _build_ui_path(
669 mounted_bundles[preferred_ui].base_url,
670 relative_path,
671 )
672 scope_prefix = (
673 scope_path[: -len(path)]
674 if scope_path.endswith(path) and path != "/"
675 else request.scope.get("root_path", "")
676 )
677 redirect_url = f"{scope_prefix}{redirect_path}"
678 if request.url.query:
679 redirect_url = f"{redirect_url}?{request.url.query}"
680 return RedirectResponse(
681 url=redirect_url,
682 status_code=status.HTTP_307_TEMPORARY_REDIRECT,
683 )
685 def reference_file_matches_cache_key(bundle: UIBundle) -> bool:
686 reference_file_path = os.path.join(
687 bundle.static_dir, UI_STATIC_REFERENCE_FILE_NAME
688 )
690 if os.path.exists(bundle.static_dir): 690 ↛ 691line 690 didn't jump to line 691 because the condition on line 690 was never true
691 try:
692 with open(reference_file_path, "r") as f:
693 return f.read() == bundle.cache_key
694 except FileNotFoundError:
695 return False
696 else:
697 return False
699 def create_ui_static_subpath(bundle: UIBundle) -> None:
700 if os.path.isdir(bundle.static_dir): 700 ↛ 701line 700 didn't jump to line 701 because the condition on line 700 was never true
701 shutil.rmtree(bundle.static_dir)
703 os.makedirs(bundle.static_dir)
704 copy_directory(bundle.source_static_path, bundle.static_dir)
705 replace_placeholder_string_in_files(
706 bundle.static_dir,
707 "/PREFECT_UI_SERVE_BASE_REPLACE_PLACEHOLDER",
708 "" if bundle.base_url == "/" else bundle.base_url,
709 )
711 # Create a file to indicate that the static files have been copied
712 # This is used to determine if the static files need to be copied again
713 # when the server is restarted
714 with open(
715 os.path.join(bundle.static_dir, UI_STATIC_REFERENCE_FILE_NAME), "w"
716 ) as f:
717 f.write(bundle.cache_key)
719 ui_app.add_middleware(GZipMiddleware)
721 if prefect.settings.PREFECT_UI_ENABLED.value() and not ephemeral: 721 ↛ 758line 721 didn't jump to line 758 because the condition on line 721 was always true
722 for bundle in bundles:
723 if not os.path.exists(bundle.source_static_path): 723 ↛ 724line 723 didn't jump to line 724 because the condition on line 723 was never true
724 continue
726 if not reference_file_matches_cache_key(bundle): 726 ↛ 745line 726 didn't jump to line 745 because the condition on line 726 was always true
727 lock_path = os.path.join(
728 os.path.dirname(bundle.static_dir),
729 f".{bundle.version}_ui_static.lock",
730 )
731 lock = FileLock(lock_path, timeout=90)
732 try:
733 lock.acquire()
734 try:
735 # Re-check after acquiring the lock; another worker
736 # may have completed the copy while we waited.
737 if not reference_file_matches_cache_key(bundle): 737 ↛ 740line 737 didn't jump to line 740 because the condition on line 737 was always true
738 create_ui_static_subpath(bundle)
739 finally:
740 lock.release()
741 except OSError as exc:
742 log_ui_static_copy_error(bundle, exc, logger)
743 continue
745 mounted_bundles[bundle.version] = bundle
747 for version in ("v2", "v1"):
748 bundle = mounted_bundles.get(version)
749 if bundle is None: 749 ↛ 750line 749 didn't jump to line 750 because the condition on line 749 was never true
750 continue
752 ui_app.frontend(
753 bundle.base_url,
754 directory=bundle.static_dir,
755 fallback="index.html",
756 )
758 return ui_app
761APP_CACHE: dict[tuple[prefect.settings.Settings, bool], FastAPI] = {}
764def _memoize_block_auto_registration(
765 fn: Callable[[], Awaitable[None]],
766) -> Callable[[], Awaitable[None]]:
767 """
768 Decorator to handle skipping the wrapped function if the block registry has
769 not changed since the last invocation
770 """
771 import toml
773 import prefect.plugins
774 from prefect._internal.compatibility.backports import tomllib
775 from prefect.blocks.core import Block
776 from prefect.server.models.block_registration import _load_collection_blocks_data
777 from prefect.utilities.dispatch import get_registry_for_type
779 @wraps(fn)
780 async def wrapper(*args: Any, **kwargs: Any) -> None:
781 if not PREFECT_MEMOIZE_BLOCK_AUTO_REGISTRATION.value(): 781 ↛ 782line 781 didn't jump to line 782 because the condition on line 781 was never true
782 await fn(*args, **kwargs)
783 return
785 # Ensure collections are imported and have the opportunity to register types
786 # before loading the registry
787 prefect.plugins.load_prefect_collections()
789 blocks_registry = get_registry_for_type(Block)
790 collection_blocks_data = await _load_collection_blocks_data()
791 current_blocks_loading_hash = hash_objects(
792 blocks_registry,
793 collection_blocks_data,
794 PREFECT_API_DATABASE_CONNECTION_URL.value(),
795 hash_algo=sha256,
796 )
798 memo_store_path = PREFECT_MEMO_STORE_PATH.value()
799 try:
800 if memo_store_path.exists(): 800 ↛ 801line 800 didn't jump to line 801 because the condition on line 800 was never true
801 saved_blocks_loading_hash = tomllib.loads(
802 memo_store_path.read_text(encoding="utf-8")
803 ).get("block_auto_registration")
804 if (
805 saved_blocks_loading_hash is not None
806 and current_blocks_loading_hash == saved_blocks_loading_hash
807 ):
808 if PREFECT_DEBUG_MODE.value():
809 logger.debug(
810 "Skipping block loading due to matching hash for block "
811 "auto-registration found in memo store."
812 )
813 return
814 except Exception as exc:
815 logger.warning(
816 ""
817 f"Unable to read memo_store.toml from {PREFECT_MEMO_STORE_PATH} during "
818 f"block auto-registration: {exc!r}.\n"
819 "All blocks will be registered."
820 )
822 await fn(*args, **kwargs)
824 if current_blocks_loading_hash is not None: 824 ↛ exitline 824 didn't return from function 'wrapper' because the condition on line 824 was always true
825 try:
826 if not memo_store_path.exists(): 826 ↛ 829line 826 didn't jump to line 829 because the condition on line 826 was always true
827 memo_store_path.touch(mode=0o0600)
829 memo_store_path.write_text(
830 toml.dumps({"block_auto_registration": current_blocks_loading_hash})
831 )
832 except Exception as exc:
833 logger.warning(
834 "Unable to write to memo_store.toml at"
835 f" {PREFECT_MEMO_STORE_PATH} after block auto-registration:"
836 f" {exc!r}.\n Subsequent server start ups will perform block"
837 " auto-registration, which may result in slower server startup."
838 )
840 return wrapper
843def _log_worker_channel_config() -> None:
844 """Log worker channel queue backend and key configuration at startup."""
845 wc_settings = prefect.settings.get_current_settings().server.worker_channel
846 logger.debug(
847 "Worker channel configuration: "
848 "cleanup_queue_storage=%s "
849 "cleanup_lease_seconds=%s "
850 "cleanup_max_delivery_attempts=%s "
851 "cleanup_completed_idempotency_retention_seconds=%s",
852 wc_settings.cleanup_queue_storage,
853 wc_settings.cleanup_lease_seconds,
854 wc_settings.cleanup_max_delivery_attempts,
855 wc_settings.cleanup_completed_idempotency_retention_seconds,
856 )
859def create_app(
860 settings: Optional[prefect.settings.Settings] = None,
861 ephemeral: bool = False,
862 webserver_only: bool = False,
863 final: bool = False,
864 ignore_cache: bool = False,
865) -> FastAPI:
866 """
867 Create a FastAPI app that includes the Prefect REST API and UI
869 Args:
870 settings: The settings to use to create the app. If not set, settings are pulled
871 from the context.
872 ephemeral: If set, the application will be treated as ephemeral. The UI
873 and services will be disabled.
874 webserver_only: If set, the webserver and UI will be available but all background
875 services will be disabled.
876 final: whether this will be the last instance of the Prefect server to be
877 created in this process, so that additional optimizations may be applied
878 ignore_cache: If set, a new application will be created even if the settings
879 match. Otherwise, an application is returned from the cache.
880 """
881 settings = settings or prefect.settings.get_current_settings()
882 cache_key = (settings.hash_key(), ephemeral, webserver_only)
883 ephemeral = ephemeral or bool(os.getenv("PREFECT__SERVER_EPHEMERAL"))
884 webserver_only = webserver_only or bool(os.getenv("PREFECT__SERVER_WEBSERVER_ONLY"))
885 final = final or bool(os.getenv("PREFECT__SERVER_FINAL"))
887 from prefect.logging.configuration import setup_logging
889 setup_logging()
891 if cache_key in APP_CACHE and not ignore_cache:
892 return APP_CACHE[cache_key]
894 # TODO: Move these startup functions out of this closure into the top-level or
895 # another dedicated location
896 async def run_migrations():
897 """Ensure the database is created and up to date with the current migrations"""
898 if prefect.settings.PREFECT_API_DATABASE_MIGRATE_ON_START: 898 ↛ exitline 898 didn't return from function 'run_migrations' because the condition on line 898 was always true
899 from prefect.server.database import provide_database_interface
901 db = provide_database_interface()
902 await db.create_db()
904 @_memoize_block_auto_registration
905 async def add_block_types():
906 """Add all registered blocks to the database"""
907 if not prefect.settings.PREFECT_API_BLOCKS_REGISTER_ON_START: 907 ↛ 908line 907 didn't jump to line 908 because the condition on line 907 was never true
908 return
910 from prefect.server.database import provide_database_interface
911 from prefect.server.models.block_registration import run_block_auto_registration
913 db = provide_database_interface()
914 session = await db.session()
916 async with session:
917 await run_block_auto_registration(session=session)
919 @asynccontextmanager
920 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
921 if app in LIFESPAN_RAN_FOR_APP: 921 ↛ 922line 921 didn't jump to line 922 because the condition on line 921 was never true
922 yield
923 return
925 await run_migrations()
926 await add_block_types()
928 _log_worker_channel_config()
930 Services: type[Service] | None = (
931 RunInWebservers
932 if webserver_only
933 else RunInEphemeralServers
934 if ephemeral
935 else Service
936 )
938 async with AsyncExitStack() as stack:
939 docket = await stack.enter_async_context(
940 Docket(
941 name=settings.server.docket.name,
942 url=settings.server.docket.url,
943 execution_ttl=timedelta(0),
944 )
945 )
946 await stack.enter_async_context(
947 background_worker(
948 docket, ephemeral=ephemeral, webserver_only=webserver_only
949 )
950 )
951 api_app.state.docket = docket
952 if Services: 952 ↛ 954line 952 didn't jump to line 954 because the condition on line 952 was always true
953 await stack.enter_async_context(Services.running())
954 LIFESPAN_RAN_FOR_APP.add(app)
955 yield
957 def on_service_exit(service: Service, task: asyncio.Task[None]) -> None:
958 """
959 Added as a callback for completion of services to log exit
960 """
961 try:
962 # Retrieving the result will raise the exception
963 task.result()
964 except Exception:
965 logger.error(f"{service.name} service failed!", exc_info=True)
966 else:
967 logger.info(f"{service.name} service stopped!")
969 app = FastAPI(
970 title=TITLE,
971 version=API_VERSION,
972 lifespan=lifespan,
973 )
974 api_app = create_api_app(
975 fast_api_app_kwargs={
976 "exception_handlers": {
977 # NOTE: FastAPI special cases the generic `Exception` handler and
978 # registers it as a separate middleware from the others
979 Exception: custom_internal_exception_handler,
980 RequestValidationError: validation_exception_handler,
981 sa.exc.IntegrityError: integrity_exception_handler,
982 ObjectNotFoundError: prefect_object_not_found_exception_handler,
983 }
984 },
985 final=final,
986 ignore_cache=ignore_cache,
987 )
988 ui_app = create_ui_app(ephemeral)
990 # middleware
991 app.add_middleware(
992 CORSMiddleware,
993 allow_origins=prefect.settings.PREFECT_SERVER_CORS_ALLOWED_ORIGINS.value().split(
994 ","
995 ),
996 allow_methods=prefect.settings.PREFECT_SERVER_CORS_ALLOWED_METHODS.value().split(
997 ","
998 ),
999 allow_headers=prefect.settings.PREFECT_SERVER_CORS_ALLOWED_HEADERS.value().split(
1000 ","
1001 ),
1002 )
1004 # Limit the number of concurrent requests when using a SQLite database to reduce
1005 # chance of errors where the database cannot be opened due to a high number of
1006 # concurrent writes
1007 if ( 1007 ↛ 1011line 1007 didn't jump to line 1011 because the condition on line 1007 was never true
1008 get_dialect(prefect.settings.PREFECT_API_DATABASE_CONNECTION_URL.value()).name
1009 == "sqlite"
1010 ):
1011 _install_sqlite_locked_log_filter()
1012 app.add_middleware(RequestLimitMiddleware, limit=100)
1014 if prefect.settings.PREFECT_SERVER_CSRF_PROTECTION_ENABLED.value(): 1014 ↛ 1015line 1014 didn't jump to line 1015 because the condition on line 1014 was never true
1015 app.add_middleware(api.middleware.CsrfMiddleware)
1017 if prefect.settings.PREFECT_API_ENABLE_METRICS: 1017 ↛ 1018line 1017 didn't jump to line 1018 because the condition on line 1017 was never true
1018 from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
1020 @api_app.get("/metrics")
1021 async def metrics() -> Response: # type: ignore[reportUnusedFunction]
1022 return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)
1024 api_app.mount(
1025 "/static",
1026 StaticFiles(
1027 directory=os.path.join(
1028 os.path.dirname(os.path.realpath(__file__)), "static"
1029 )
1030 ),
1031 name="static",
1032 )
1033 app.api_app = api_app
1034 if PREFECT_SERVER_API_BASE_PATH: 1034 ↛ 1035line 1034 didn't jump to line 1035 because the condition on line 1034 was never true
1035 app.mount(PREFECT_SERVER_API_BASE_PATH.value(), app=api_app, name="api")
1036 else:
1037 app.mount("/api", app=api_app, name="api")
1038 app.mount("/", app=ui_app, name="ui")
1040 def openapi():
1041 """
1042 Convenience method for extracting the user facing OpenAPI schema from the API app.
1044 This method is attached to the global public app for easy access.
1045 """
1046 partial_schema = get_openapi(
1047 title=API_TITLE,
1048 version=API_VERSION,
1049 routes=api_app.routes,
1050 )
1051 new_schema = partial_schema.copy()
1053 new_schema["info"]["x-logo"] = {"url": "static/prefect-logo-mark-gradient.png"}
1054 return new_schema
1056 app.openapi = openapi
1058 APP_CACHE[cache_key] = app
1059 return app
1062subprocess_server_logger: "logging.Logger" = get_logger()
1065class SubprocessASGIServer:
1066 _instances: dict[int | None, "SubprocessASGIServer"] = {}
1067 _port_range: range = range(8000, 9000)
1069 def __new__(cls, port: int | None = None, *args: Any, **kwargs: Any) -> Self:
1070 """
1071 Return an instance of the server associated with the provided port.
1072 Prevents multiple instances from being created for the same port.
1073 """
1074 if port not in cls._instances:
1075 instance = super().__new__(cls)
1076 cls._instances[port] = instance
1077 return cls._instances[port]
1079 def __init__(self, port: Optional[int] = None):
1080 # This ensures initialization happens only once
1081 if not hasattr(self, "_initialized"):
1082 self._instance_key: Optional[int] = port
1083 self.port: Optional[int] = port
1084 self.server_process: subprocess.Popen[Any] | None = None
1085 self.running: bool = False
1086 self._initialized = True
1088 def find_available_port(self) -> int:
1089 max_attempts = 10
1090 for _ in range(max_attempts):
1091 port = random.choice(self._port_range)
1092 if self.is_port_available(port):
1093 return port
1094 time.sleep(random.uniform(0.1, 0.5)) # Random backoff
1095 raise RuntimeError("Unable to find an available port after multiple attempts")
1097 @staticmethod
1098 def is_port_available(port: int) -> bool:
1099 with contextlib.closing(
1100 socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1101 ) as sock:
1102 try:
1103 sock.bind(("127.0.0.1", port))
1104 return True
1105 except socket.error:
1106 return False
1108 @property
1109 def address(self) -> str:
1110 return f"http://127.0.0.1:{self.port}"
1112 @property
1113 def api_url(self) -> str:
1114 return f"{self.address}/api"
1116 def start(self, timeout: Optional[int] = None) -> None:
1117 """
1118 Start the server in a separate process. Safe to call multiple times; only starts
1119 the server once.
1121 Args:
1122 timeout: The maximum time to wait for the server to start
1123 """
1124 if not self.running:
1125 if self.port is None:
1126 self.port = self.find_available_port()
1127 assert self.port is not None, "Port must be provided or available"
1128 help_message = (
1129 f"Starting temporary server on {self.address}\nSee "
1130 "https://docs.prefect.io/v3/concepts/server#how-to-guides "
1131 "for more information on running a dedicated Prefect server."
1132 )
1133 subprocess_server_logger.info(help_message)
1134 try:
1135 self.running = True
1136 self.server_process = self._run_uvicorn_command()
1137 atexit.register(self.stop)
1138 with httpx.Client() as client:
1139 response = None
1140 elapsed_time = 0
1141 max_wait_time = (
1142 timeout
1143 or PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS.value()
1144 )
1145 while elapsed_time < max_wait_time:
1146 if self.server_process.poll() == 3:
1147 self.port = self.find_available_port()
1148 self.server_process = self._run_uvicorn_command()
1149 continue
1150 try:
1151 response = client.get(f"{self.api_url}/health")
1152 except httpx.ConnectError:
1153 pass
1154 else:
1155 if response.status_code == 200:
1156 break
1157 time.sleep(0.1)
1158 elapsed_time += 0.1
1159 if response:
1160 response.raise_for_status()
1161 if not response:
1162 error_message = "Timed out while attempting to connect to ephemeral Prefect API server."
1163 if self.server_process.poll() is not None:
1164 error_message += f" Ephemeral server process exited with code {self.server_process.returncode}."
1165 if self.server_process.stdout:
1166 error_message += (
1167 f" stdout: {self.server_process.stdout.read()}"
1168 )
1169 if self.server_process.stderr:
1170 error_message += (
1171 f" stderr: {self.server_process.stderr.read()}"
1172 )
1173 raise RuntimeError(error_message)
1174 except Exception:
1175 self.running = False
1176 raise
1178 def _run_uvicorn_command(self) -> subprocess.Popen[Any]:
1179 # used to turn off serving the UI
1180 server_env = {
1181 "PREFECT_UI_ENABLED": "0",
1182 "PREFECT__SERVER_EPHEMERAL": "1",
1183 "PREFECT__SERVER_FINAL": "1",
1184 }
1185 return subprocess.Popen(
1186 args=[
1187 sys.executable,
1188 "-m",
1189 "uvicorn",
1190 "--app-dir",
1191 str(prefect.__module_path__.parent),
1192 "--factory",
1193 "prefect.server.api.server:create_app",
1194 "--host",
1195 "127.0.0.1",
1196 "--port",
1197 str(self.port),
1198 "--log-level",
1199 "error",
1200 "--lifespan",
1201 "on",
1202 ],
1203 env={
1204 **os.environ,
1205 **server_env,
1206 **get_current_settings().to_environment_variables(exclude_unset=True),
1207 },
1208 )
1210 def stop(self) -> None:
1211 if self.server_process:
1212 subprocess_server_logger.info(
1213 f"Stopping temporary server on {self.address}"
1214 )
1215 self.server_process.terminate()
1216 try:
1217 self.server_process.wait(timeout=5)
1218 except subprocess.TimeoutExpired:
1219 self.server_process.kill()
1220 # Ensure the process is reaped to avoid ResourceWarning
1221 self.server_process.wait()
1222 finally:
1223 self.server_process = None
1224 # Use _instance_key (the original port passed to __new__) for cleanup,
1225 # since self.port may have changed during start() when an available port
1226 # was assigned.
1227 instance_key = getattr(self, "_instance_key", self.port)
1228 if instance_key in self._instances:
1229 del self._instances[instance_key]
1230 if self.running:
1231 self.running = False
1233 def __enter__(self) -> Self:
1234 self.start()
1235 return self
1237 def __exit__(self, *args: Any) -> None:
1238 self.stop()