qwen_embedding / login.py
SPAL0074's picture
Upload 10 files
4258647 verified
Raw
History Blame Contribute Delete
2.88 kB
# login.py
from __future__ import annotations
import os
import threading
import logging
from typing import Optional
from dotenv import load_dotenv
from huggingface_hub import login as hf_login
_logger = logging.getLogger(__name__)
class HuggingFaceLogin:
"""
Lightweight, idempotent wrapper around Hugging Face Hub login.
- Reads token from environment (.env supported).
- Safe to call authenticate() multiple times (thread-safe).
- Avoids writing to git credential store by default.
"""
_env_loaded = False
_env_lock = threading.Lock()
def __init__(
self,
token: Optional[str] = None,
*,
read_dotenv: bool = True,
add_to_git_credential: bool = False,
) -> None:
# Load env only once per process (thread-safe)
if read_dotenv:
with self._env_lock:
if not self._env_loaded:
load_dotenv()
self.__class__._env_loaded = True
# Accept explicit token or fall back to common env vars
self._token = token or self._read_token_from_env()
if not self._token:
raise ValueError(
"Hugging Face API token not found. "
"Set one of HF_TOKEN / HUGGINGFACEHUB_API_TOKEN / HUGGINGFACE_TOKEN in your environment "
"or pass token=... to HuggingFaceLogin()."
)
self._add_to_git_credential = add_to_git_credential
self._did_authenticate = False
self._auth_lock = threading.Lock()
# -----------------------
# Public API
# -----------------------
def authenticate(self) -> bool:
"""
Perform a one-time login to the Hugging Face Hub.
Safe to call multiple times (no-op after first success).
Returns True if authenticated in this call or previously.
"""
if self._did_authenticate:
return True
with self._auth_lock:
if self._did_authenticate:
return True
# Perform an in-process login; do not persist to git credentials by default
hf_login(
token=self._token,
add_to_git_credential=self._add_to_git_credential,
)
self._did_authenticate = True
_logger.info("✅ Authenticated with Hugging Face Hub.")
return True
@property
def is_authenticated(self) -> bool:
return self._did_authenticate
# -----------------------
# Helpers
# -----------------------
def _read_token_from_env(self) -> Optional[str]:
# Check common env var names in order of preference
for key in ("HF_TOKEN", "HUGGINGFACEHUB_API_TOKEN", "HUGGINGFACE_TOKEN"):
val = os.getenv(key)
if val and val.strip():
return val.strip()
return None