Spaces:
Running
Running
File size: 3,520 Bytes
9734b71 4106e0f 9734b71 2eec8c3 9734b71 2eec8c3 9734b71 4106e0f 9734b71 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | from __future__ import annotations
import os
import sys
from pathlib import Path
from typing import Any, Callable, Mapping, MutableMapping, Protocol
from dotenv import load_dotenv
from backend.runtime_utils import (
can_bind_tcp_port,
find_free_tcp_port,
prepare_runtime_environment,
)
DEFAULT_HUGGINGFACE_PORT = 7860
ADMIN_TOKEN_ENV_KEY = "ADMIN_TOKEN"
class LoggerLike(Protocol):
def warning(self, msg: str, *args: Any, **kwargs: Any) -> None: ...
def is_huggingface_space(env: Mapping[str, str] | None = None) -> bool:
runtime_env = env or os.environ
return bool(runtime_env.get("SPACE_ID") or runtime_env.get("SPACE_HOST"))
def bootstrap_runtime_port(
env: MutableMapping[str, str] | None = None,
*,
huggingface_port: int = DEFAULT_HUGGINGFACE_PORT,
) -> None:
runtime_env = env or os.environ
if is_huggingface_space(runtime_env) and not runtime_env.get("PORT", "").strip():
runtime_env["PORT"] = str(huggingface_port)
def load_runtime_env(
env_file: Path,
*,
override: bool = False,
) -> bool:
if not env_file.exists():
return False
load_dotenv(env_file, override=override)
return True
def is_admin_token_configured(
env: Mapping[str, str] | None = None,
*,
token_env_key: str = ADMIN_TOKEN_ENV_KEY,
) -> bool:
runtime_env = os.environ if env is None else env
return bool(runtime_env.get(token_env_key, "").strip())
def require_admin_token_configured(
env: Mapping[str, str] | None = None,
*,
token_env_key: str = ADMIN_TOKEN_ENV_KEY,
) -> None:
if is_admin_token_configured(env, token_env_key=token_env_key):
return
raise RuntimeError(
"ADMIN_TOKEN chua duoc cau hinh. Tu choi khoi dong runtime de tranh fallback auth mac dinh."
)
def load_fastapi_app(
project_root: Path,
*,
module_name: str = "backend.main",
attr_name: str = "app",
) -> Any:
prepare_runtime_environment(project_root)
project_root_str = str(project_root)
if project_root_str not in sys.path:
sys.path.insert(0, project_root_str)
module = __import__(module_name, fromlist=[attr_name])
return getattr(module, attr_name)
def resolve_server_port(
host: str,
*,
env: MutableMapping[str, str] | None = None,
logger: LoggerLike | None = None,
huggingface_port: int = DEFAULT_HUGGINGFACE_PORT,
port_checker: Callable[[str, int], bool] = can_bind_tcp_port,
free_port_finder: Callable[[str], int] = find_free_tcp_port,
) -> int:
runtime_env = env or os.environ
raw_port = runtime_env.get("PORT", "").strip()
if raw_port:
try:
configured_port = int(raw_port)
except ValueError as exc:
raise ValueError(f"Invalid PORT value: {raw_port}") from exc
if is_huggingface_space(runtime_env) or port_checker(host, configured_port):
return configured_port
fallback_port = free_port_finder(host)
runtime_env["PORT"] = str(fallback_port)
if logger is not None:
logger.warning(
"PORT %s dang ban tren %s, tu dong chuyen sang cong %s",
configured_port,
host,
fallback_port,
)
return fallback_port
if is_huggingface_space(runtime_env):
runtime_env["PORT"] = str(huggingface_port)
return huggingface_port
fallback_port = free_port_finder(host)
runtime_env["PORT"] = str(fallback_port)
return fallback_port
|