0
$\begingroup$

I'm new to Blender Python programming, and for a personal project, I'm experimenting with creating a Blender Python addon. I aim to execute an asynchronous GET request when I select a button in a panel (custom operator). You can find an example code snippet below. My current issue is that upon selecting the operator, it appears to block any other Blender interactions until the asynchronous call completes. Do you have any suggestions or examples on how I can prevent the operator's logic from blocking? I still want users to be able to continue editing their models, etc., while these API calls/downloads occur.

import bpy
import json
import asyncio
import aiohttp

from bpy.types import Operator, Panel
      
# ------------------------------------------------------------------------
#    Operators
# ------------------------------------------------------------------------                
class ADDON_OT_test(Operator): 
    bl_idname = "addon.search"
    bl_label = "Search" 
                
    def execute(self, context):
        scene = context.scene
        mytool = scene.my_properties
            
             
        # Endpoint Resource Path 
        assets_url = "someurl"
                
        # Request Body
        assets_query_body = { }
    
        # Request Headers
        headers = { }
        
        x = asyncio.run(fetch_data(assets_url, assets_query_body, headers))

        if x is None:
            self.report({'ERROR'}, "something went wrong")
        else:
            beautify_data = json.dumps(x, indent=4)
            print(beautify_data)
            self.report({'INFO'}, "all done")
        
        return {'FINISHED'} 
      
# ------------------------------------------------------------------------
#    Panels
# ------------------------------------------------------------------------

class ADDON_PT_browse_search_panel(Panel):
    bl_idname = "ADDON_PT_Search_Results"
    bl_label = "Search Results"

    bl_space_type = 'VIEW_3D'
    bl_region_type = 'UI'
    
    def draw  (self, context):
        layout = self.layout
        scene = context.scene
        mytool = scene.my_properties
        
        layout.operator("addon.search", text="Search")
        
# ------------------------------------------------------------------------
#    Additional Functionality
# ------------------------------------------------------------------------          
async def fetch_data(url, query_params=None, headers=None):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, params=query_params, headers=headers) as response:
            print("Status:", response.status)
            if response.status == 200:
                print(response.status)
                return await response.json()
            else:
                print(response.status)
                return None    

# ------------------------------------------------------------------------
#    Register/Deregister
# ------------------------------------------------------------------------
classes = [
    ADDON_PT_browse_search_panel,
    ADDON_OT_test
]

def register():
    for cls in classes: 
        bpy.utils.register_class(cls)

def unregister():
    for cls in reversed(classes): 
        bpy.utils.unregister_class(cls)

if __name__ == "__main__":
    register()
$\endgroup$

1 Answer 1

0
$\begingroup$

This is an example of the solution that ended up achieving my goal

Found inspiration from How do I Download an External File from URL and Display Progress in a Panel without blocking the UI?

import bpy
import json
import asyncio
import aiohttp
import threading
import traceback

from bpy.types import Operator, Panel
from bpy.props import PointerProperty
from ..model import WarningText
# ------------------------------------------------------------------------------
#    Operators
# ------------------------------------------------------------------------------
class ADDON_OT_test(Operator):
    """Example Non-Blocking Modal Operator that executes asynchronous API calls.""" 
    bl_idname = "addon.test"
    bl_label = "Test"

    thread = None

    # setup local variables to pass info to context variables
    # (passing context to threads in not advised)
    message = ""
    response_status = 0

    # request variables that contain context info
    endpoint_url = ""
    query_parameters = {}
    headers = {}

    def modal(self, context, event):
        global is_running_search
        wm = context.window_manager
        myprop = context.scene.my_properties

        if event.type == 'TIMER':

            # for UI updates (needs to be first line)
            wm.name = wm.name

            # update every time there are changes in values
            if myprop.test_status != self.message:
                myprop.test_status = self.message

            # handling threads when finishing (successfully or post exception)
            if not self.thread.is_alive():
                self.thread.join()

                # handle any UI updates
                wm.name = wm.name

                # success
                if self.response_status == 200:
                
                    # TODO: pass values stored during API call to properties here

                    print("Test Done")
                else:

                    # TODO: logic if request failed

                    print("Test Interrupted")

                # no longer running
                is_running_search = False

                return {'FINISHED'}

        return {'PASS_THROUGH'}

    def execute(self, context):
        global is_running_search
        wm = context.window_manager
        myprop = context.scene.my_properties

        # ensures we don't re-trigger operator while it's already running
        if not is_running_search:
            is_running_search = True

            # pass endpoint url, headers, query parameters
            self.endpoint_url = f"{myprop.api_url}"
            self.query_parameters = {
                "example": myprop.example
            }
            self.headers = {
                "Content-Type": "application/json",
                "tenant-id": myprop.id,
                "authorization-token": myprop.access_token
            }

            # thread start
            self.thread = threading.Thread(target=request_api_call, args=(self,))
            self.thread.start()

            wm.event_timer_add(0.5, window=context.window)
            wm.modal_handler_add(self)

        return {'RUNNING_MODAL'}

def request_api_call(self):
    """API Call"""

    self.message = "Beginning GET API call..."
    print("Beginning 'GET' API call...")

    try:

        # GET request for asset IDs
        response_status, asset_data = asyncio.run(
            make_api_get_request(
                self.assets_url, 
                self.assets_query_parameters, 
                self.assets_headers
            )
        )

        beautify_data = json.dumps(asset_data, indent=4)
        print(beautify_data)

        # TODO: additional processing logic here

        # everything has passed by this point - assign success response status 
        self.response_status = response_status # 200 (assumption)

    # error handling
    except aiohttp.ClientResponseError as error:
        self.message = f"'{error.status}' {WarningText.EXCEPTION_THROWN}"
        print(error.message)
    except ValueError as error:
        self.message = f"Response Validation {WarningText.EXCEPTION_THROWN}"

        full_error_message = (
            "Response Validation Failed! "
            "Check your JSON response for other field errors:\n"
            f"    {error}"
        )
        print(f"  {full_error_message}")
    except Exception as error: # TODO: WHAT OTHER ERROR SCENARIOS ARE THERE?
        self.message = WarningText.EXCEPTION_THROWN
        print(traceback.format_exc())
      
# ------------------------------------------------------------------------
#    Panels
# ------------------------------------------------------------------------
class ADDON_PT_test_panel(Panel):
    bl_idname = "ADDON_PT_test_panel"
    bl_label = "Test Panel"

    bl_category = 'Test'
    bl_space_type = 'VIEW_3D'
    bl_region_type = 'UI'
    
    def draw  (self, context):
        layout = self.layout
        scene = context.scene
        myprop = scene.my_properties
        
        layout.operator("addon.test", text="Search")

        # TODO: additional UI layout
    
# ------------------------------------------------------------------------
#    Additional Functionality
# ------------------------------------------------------------------------          
async def make_api_get_request(url, query_params=None, headers=None):
    data = {}
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(url, params=query_params, headers=headers) as response:
                data = await response.json()
                response.raise_for_status() # raise exception if status not 2xx

                return response.status, data
        except aiohttp.ClientResponseError as error:
            
            # custom error message
            error.message = custom_http_error_response(error, data)
            raise error.message
# ------------------------------------------------------------------------
#    Register/Deregister
# ------------------------------------------------------------------------
classes = [
    ADDON_PT_test_panel,
    ADDON_OT_test
]

def register():
    for cls in classes: 
        bpy.utils.register_class(cls)

    # stored as property of scene - for accessing via context.scene   
    bpy.types.Scene.my_properties = PointerProperty(type=MyProperties)

def unregister():
    for cls in reversed(classes): 
        bpy.utils.unregister_class(cls)

    del bpy.types.Scene.my_properties

if __name__ == "__main__":
    register()
$\endgroup$

You must log in to answer this question.

Not the answer you're looking for? Browse other questions tagged .