create_huggingface_space.py

from huggingface_hub import HfApi, CommitOperationAdd
import os
import time
import logging
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def create_space(max_retries: int = 3, retry_delay: int = 5) -> Optional[str]:
    """Create and deploy HuggingFace Space with retry logic"""
    
    # Get token from environment
    token = os.environ.get("HUGGINGFACE_TOKEN")
    if not token:
        raise ValueError("HUGGINGFACE_TOKEN not found in environment")
        
    api = HfApi(token=token)
    
    try:
        # Get user info
        whoami = api.whoami()
        username = whoami.get("name")
        if not username:
            raise ValueError("Could not determine HuggingFace username")
            
        # Create new space
        space_name = "python-worker"
        space_id = f"{username}/{space_name}"
        
        logger.info(f"Deploying to space: {space_id}")
        
        # Prepare files for upload
        operations = []
        files_to_upload = {
            "Dockerfile": "Dockerfile",
            "requirements.txt": "requirements.txt",
            "app.py": "app.py",
            "README.md": "README.md"
        }
        
        # Get the base directory for the python worker
        base_dir = os.path.dirname(os.path.abspath(__file__))
        
        for local_path, repo_path in files_to_upload.items():
            file_path = os.path.join(base_dir, local_path)
            with open(file_path, "r") as f:
                content = f.read()
                operations.append(CommitOperationAdd(
                    path_or_fileobj=content.encode(),
                    path_in_repo=repo_path
                ))
        
        # Upload all files in a single commit
        api.create_commit(
            repo_id=space_id,
            repo_type="space",
            operations=operations,
            commit_message="Update automation service files"
        )
        
        space_url = f"https://huggingface.co/spaces/{space_id}"
        logger.info(f"Space updated successfully: {space_url}")
        return space_url
            
    except Exception as e:
        logger.error(f"Deployment failed: {str(e)}")
        raise

if __name__ == "__main__":
    create_space()