Spaces:
Runtime error
Runtime error
| import requests | |
| from huggingface_hub import HfApi, login, whoami, Repository | |
| import json | |
| import time | |
| import os | |
| from typing import Tuple, Dict, Any, Optional | |
| def authenticate_hf(token: str) -> Tuple[bool, str, Optional[Dict]]: | |
| """ | |
| Authenticate with Hugging Face using the provided token. | |
| Args: | |
| token: Hugging Face access token | |
| Returns: | |
| Tuple of (success, message, user_info) | |
| """ | |
| try: | |
| login(token=token) | |
| user_info = whoami() | |
| return True, f"✅ Authenticated as {user_info['name']}", user_info | |
| except Exception as e: | |
| return False, f"❌ Authentication failed: {str(e)}", None | |
| def deploy_model(model_id: str, task: str = "text-generation", hardware: str = "cpu") -> Tuple[bool, str, Optional[str]]: | |
| """ | |
| Deploy a model to Hugging Face Inference Endpoints. | |
| Args: | |
| model_id: Model repository ID | |
| task: Task type for the model | |
| hardware: Hardware accelerator type | |
| Returns: | |
| Tuple of (success, message, endpoint_url) | |
| """ | |
| try: | |
| api = HfApi() | |
| endpoint_name = f"{model_id.split('/')[-1]}-endpoint".replace(".", "-") | |
| # Check if endpoint already exists | |
| try: | |
| existing_endpoints = list(api.list_inference_endpoints()) | |
| for endpoint in existing_endpoints: | |
| if endpoint.name == endpoint_name: | |
| return True, f"✅ Endpoint already exists: {endpoint.url}", endpoint.url | |
| except: | |
| pass | |
| # Create new endpoint | |
| endpoint = api.create_inference_endpoint( | |
| name=endpoint_name, | |
| model=model_id, | |
| task=task, | |
| accelerator=hardware, | |
| type="public", | |
| framework="pytorch" | |
| ) | |
| return True, f"✅ Deployment initiated! URL: {endpoint.url}", endpoint.url | |
| except Exception as e: | |
| return False, f"❌ Deployment failed: {str(e)}", None | |
| def get_user_models(username: str, token: str, limit: int = 20) -> list: | |
| """ | |
| Get list of models for a user. | |
| Args: | |
| username: Hugging Face username | |
| token: Access token | |
| limit: Maximum number of models to fetch | |
| Returns: | |
| List of model IDs | |
| """ | |
| try: | |
| api = HfApi(token=token) | |
| models = list(api.list_models(author=username, limit=limit)) | |
| return [model.id for model in models] | |
| except Exception as e: | |
| print(f"Error fetching models: {e}") | |
| return [] | |
| def check_deployment_status(endpoint_name: str, token: str) -> Dict[str, Any]: | |
| """ | |
| Check the status of a deployed endpoint. | |
| Args: | |
| endpoint_name: Name of the endpoint | |
| token: Access token | |
| Returns: | |
| Dictionary with endpoint status information | |
| """ | |
| try: | |
| api = HfApi(token=token) | |
| endpoint = api.get_inference_endpoint(endpoint_name) | |
| return { | |
| "name": endpoint.name, | |
| "status": endpoint.status, | |
| "url": endpoint.url, | |
| "model": endpoint.model, | |
| "created_at": endpoint.created_at, | |
| "updated_at": endpoint.updated_at | |
| } | |
| except Exception as e: | |
| return {"error": str(e)} | |
| def get_inference_endpoint(endpoint_url: str, prompt: str, token: str, max_tokens: int = 100) -> Optional[str]: | |
| """ | |
| Get inference from a deployed endpoint. | |
| Args: | |
| endpoint_url: URL of the inference endpoint | |
| prompt: Input prompt for the model | |
| token: Access token | |
| max_tokens: Maximum tokens to generate | |
| Returns: | |
| Generated text or None if error | |
| """ | |
| try: | |
| headers = {"Authorization": f"Bearer {token}"} | |
| data = { | |
| "inputs": prompt, | |
| "parameters": { | |
| "max_new_tokens": max_tokens, | |
| "temperature": 0.7, | |
| "do_sample": True | |
| } | |
| } | |
| response = requests.post(endpoint_url, headers=headers, json=data, timeout=30) | |
| if response.status_code == 200: | |
| result = response.json() | |
| if isinstance(result, list) and len(result) > 0: | |
| return result[0].get("generated_text", "") | |
| return str(result) | |
| else: | |
| print(f"API Error: {response.status_code} - {response.text}") | |
| return None | |
| except Exception as e: | |
| print(f"Inference error: {e}") | |
| return None | |
| def create_model_repository(model_name: str, username: str, token: str, private: bool = False) -> Tuple[bool, str]: | |
| """ | |
| Create a new model repository on Hugging Face. | |
| Args: | |
| model_name: Name for the new model | |
| username: Hugging Face username | |
| token: Access token | |
| private: Whether the repository should be private | |
| Returns: | |
| Tuple of (success, message) | |
| """ | |
| try: | |
| api = HfApi(token=token) | |
| repo_id = f"{username}/{model_name}" | |
| # Check if repository already exists | |
| try: | |
| api.repo_info(repo_id=repo_id, repo_type="model") | |
| return False, f"❌ Repository {repo_id} already exists" | |
| except: | |
| pass | |
| # Create new repository | |
| api.create_repo( | |
| repo_id=repo_id, | |
| repo_type="model", | |
| private=private, | |
| exist_ok=False | |
| ) | |
| return True, f"✅ Repository {repo_id} created successfully" | |
| except Exception as e: | |
| return False, f"❌ Failed to create repository: {str(e)}" | |
| def upload_model_to_repo(local_path: str, repo_id: str, token: str, commit_message: str = "Upload model") -> Tuple[bool, str]: | |
| """ | |
| Upload model files to a Hugging Face repository. | |
| Args: | |
| local_path: Local path to model files | |
| repo_id: Repository ID | |
| token: Access token | |
| commit_message: Commit message for the upload | |
| Returns: | |
| Tuple of (success, message) | |
| """ | |
| try: | |
| api = HfApi(token=token) | |
| # Upload all files in the directory | |
| api.upload_folder( | |
| folder_path=local_path, | |
| repo_id=repo_id, | |
| repo_type="model", | |
| commit_message=commit_message | |
| ) | |
| return True, f"✅ Model uploaded to {repo_id}" | |
| except Exception as e: | |
| return False, f"❌ Upload failed: {str(e)}" | |
| def list_available_endpoints(token: str) -> list: | |
| """ | |
| List all available inference endpoints. | |
| Args: | |
| token: Access token | |
| Returns: | |
| List of endpoint information | |
| """ | |
| try: | |
| api = HfApi(token=token) | |
| endpoints = list(api.list_inference_endpoints()) | |
| return [ | |
| { | |
| "name": endpoint.name, | |
| "url": endpoint.url, | |
| "model": endpoint.model, | |
| "status": endpoint.status | |
| } | |
| for endpoint in endpoints | |
| ] | |
| except Exception as e: | |
| print(f"Error listing endpoints: {e}") | |
| return [] | |
| def delete_endpoint(endpoint_name: str, token: str) -> Tuple[bool, str]: | |
| """ | |
| Delete an inference endpoint. | |
| Args: | |
| endpoint_name: Name of the endpoint to delete | |
| token: Access token | |
| Returns: | |
| Tuple of (success, message) | |
| """ | |
| try: | |
| api = HfApi(token=token) | |
| api.delete_inference_endpoint(endpoint_name) | |
| return True, f"✅ Endpoint {endpoint_name} deleted successfully" | |
| except Exception as e: | |
| return False, f"❌ Failed to delete endpoint: {str(e)}" | |
| def get_model_info(model_id: str, token: str) -> Dict[str, Any]: | |
| """ | |
| Get detailed information about a model. | |
| Args: | |
| model_id: Model repository ID | |
| token: Access token | |
| Returns: | |
| Dictionary with model information | |
| """ | |
| try: | |
| api = HfApi(token=token) | |
| model_info = api.repo_info(repo_id=model_id, repo_type="model") | |
| return { | |
| "id": model_info.id, | |
| "author": model_info.author, | |
| "sha": model_info.sha, | |
| "last_modified": model_info.last_modified, | |
| "tags": model_info.tags, | |
| "pipeline_tag": model_info.pipeline_tag, | |
| "downloads": model_info.downloads, | |
| "likes": model_info.likes, | |
| "private": model_info.private, | |
| "siblings": [file.rfilename for file in model_info.siblings] | |
| } | |
| except Exception as e: | |
| return {"error": str(e)} |