def _process()

in glean-core/python/glean/net/ping_upload_worker.py [0:0]

• 42 lines of code
• 12 McCabe index (conditional complexity)

def _process(data_dir: Path, application_id: str, configuration) -> bool:
    # Import here to avoid cyclical import
    from ..glean import Glean

    if not Glean.is_initialized():
        # We don't want to send pings or otherwise update the database during
        # initialization in a subprocess, so we use
        # `glean_initialize_for_subprocess` rather than `glean_initialize` here.
        cfg = InternalConfiguration(
            data_path=str(data_dir),
            application_id=application_id,
            language_binding_name="python",
            # Set upload enabled to False. The subprocess should not record
            # telemetry. In the special `glean_initialize_for_subprocess` mode,
            # this does not have any side effects like resetting the client_id
            # or sending a deletion-request ping.
            upload_enabled=False,
            max_events=configuration.max_events,
            delay_ping_lifetime_io=False,
            use_core_mps=False,
            app_build="",
            trim_data_to_registered_pings=False,
            log_level=None,
            rate_limit=None,
            enable_event_timestamps=False,
            experimentation_id=None,
            enable_internal_pings=False,
            ping_schedule={},
            ping_lifetime_threshold=0,
            ping_lifetime_max_time=0,
        )
        if not glean_initialize_for_subprocess(cfg):
            log.error("Couldn't initialize Glean in subprocess")
            sys.exit(1)

    # Limits are enforced by glean-core to avoid an inifinite loop here.
    # Whenever a limit is reached, this binding will receive `UploadTaskTag.DONE` and step out.
    while True:
        task = glean_get_upload_task()

        if isinstance(task, PingUploadTask.UPLOAD):
            # Ping data is available for upload: parse the structure but make
            # sure to let Rust free the memory.
            request = task.request

            capable_request = CapablePingUploadRequest(
                PingUploadRequest(request, configuration.server_endpoint)
            )

            # Delegate the upload to the uploader.
            upload_result = configuration.ping_uploader.do_upload(capable_request)

            # Process the response.
            action = glean_process_ping_upload_response(request.document_id, upload_result)
            if action == UploadTaskAction.NEXT:
                continue
            else:  # action == UploadTaskAction.END
                return True

        elif isinstance(task, PingUploadTask.WAIT):
            time.sleep(task.time / 1000)
        elif task.is_done():
            return True