INNER CODE UNIT · Python

get_executor_client

pgalko/BambooAI · web_app/app.py:177

def get_executor_client(user_id: str):
    """Get executor client for specific user"""
    executor_urls = get_dynamic_executor_urls(user_id)
    if executor_urls:
        from bambooai import executor_client
        return executor_client.ExecutorAPIClient(base_url=executor_urls['EXECUTOR_API_BASE_URL'])
    else:
        return None
    
def update_user_activity(user_id: str):
    """Update user activity in orchestrator"""
    try:
        if ORCHESTRATOR_API_URL and GLOBAL_EXECUTION_MODE == 'api' and not DIRECT_EXECUTOR_URL:   # only the orchestrated path has an idle reaper
            response = requests.post(
                f"{ORCHESTRATOR_API_URL}/activity/{user_id}", 
                timeout=3
            )
            if response.status_code == 200:

View source record →