Community help

OmkarIndulkar · ‎08-07-2024

DAPClientError: malformed HTTP response occurs at least twice in a month. and was unable to identify the exact issue.

DAPClientError                            Traceback (most recent call last)
Cell In [35], line 3
      1 if __name__ == '__main__':
      2     loop = asyncio.get_event_loop()
----> 3     res = loop.run_until_complete(main())
      4     try:
      5         loop.close()    

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/nest_asyncio.py:90, in _patch_loop.<locals>.run_until_complete(self, future)
     87 if not f.done():
     88     raise RuntimeError(
     89         'Event loop stopped before Future completed.')
---> 90 return f.result()

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/futures.py:201, in Future.result(self)
    199 self.__log_traceback = False
    200 if self._exception is not None:
--> 201     raise self._exception.with_traceback(self._exception_tb)
    202 return self._result

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/tasks.py:232, in Task.__step(***failed resolving arguments***)
    228 try:
    229     if exc is None:
    230         # We use the send method directly, because coroutines
    231         # don't have __iter__ and __next__ methods.
--> 232         result = coro.send(None)
    233     else:
    234         result = coro.throw(exc)

Cell In [31], line 3, in main()
      1 async def main():
      2     async with ClientSession() as session:
----> 3         data_d = await asyncio.gather(*[download_and_upload_to_blob(table, session) for table in valid_tables])

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/tasks.py:304, in Task.__wakeup(self, future)
    302 def __wakeup(self, future):
    303     try:
--> 304         future.result()
    305     except BaseException as exc:
    306         # This may also be a cancellation.
    307         self.__step(exc)

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/tasks.py:232, in Task.__step(***failed resolving arguments***)
    228 try:
    229     if exc is None:
    230         # We use the send method directly, because coroutines
    231         # don't have __iter__ and __next__ methods.
--> 232         result = coro.send(None)
    233     else:
    234         result = coro.throw(exc)

Cell In [27], line 27, in download_and_upload_to_blob(table, session)
     25             await asyncio.sleep(BACKOFF_FACTOR ** attempt)
     26         else:
---> 27             raise e
     31 # Create a BlobServiceClient
     32 blob_service_client = BlobServiceClient(
     33     account_url=account_url,
     34     credential=account_key
     35 )

Cell In [27], line 20, in download_and_upload_to_blob(table, session)
     13     async with DAPClient() as session:
     14         query = IncrementalQuery(
     15             format=Format.Parquet,
     16             mode=None,
     17             since=last_seen,
     18             until=None,
     19         )
---> 20         await session.download_table_data("canvas", table, query, output_directory)
     21     break
     22 except Exception as e:

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:676, in DAPSession.download_table_data(self, namespace, table, query, output_directory, decompress)
    673 objects = await self.get_objects(job.id)
    674 directory = os.path.join(output_directory, f"job_{job.id}")
--> 676 downloaded_files = await self.download_objects(objects, directory, decompress)
    677 logger.info(f"Files from server downloaded to folder: {directory}")
    679 if isinstance(job, CompleteSnapshotJob):

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:566, in DAPSession.download_objects(self, objects, output_directory, decompress)
    553 """
    554 Save data stored remotely into a local directory.
    555 
   (...)
    559 :returns: A list of paths to files saved in the local file system.
    560 """
    562 downloads = [
    563     self.download_object(object, output_directory, decompress)
    564     for object in objects
    565 ]
--> 566 local_files = await gather_n(downloads, concurrency=DOWNLOAD_CONCURRENCY)
    568 return local_files

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/concurrency.py:186, in gather_n(coroutines, concurrency, return_exceptions)
    169 async def gather_n(
    170     coroutines: Iterable[Invokable[T]],
    171     *,
    172     concurrency: int,
    173     return_exceptions: bool = False
    174 ) -> Iterable[T]:
    175     """
    176     Runs awaitable objects, with at most the specified degree of concurrency.
    177 
   (...)
    183     :raises asyncio.CancelledError: Raised when one of the tasks in cancelled.
    184     """
--> 186     results = await _gather_n(
    187         coroutines, concurrency=concurrency, return_exceptions=return_exceptions
    188     )
    190     if isinstance(coroutines, list):
    191         return list(results)

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/concurrency.py:111, in _gather_n(coroutines, concurrency, return_exceptions)
    109     exc = task.exception()
    110     if exc and not return_exceptions:
--> 111         raise exc
    113 # schedule some tasks up to degree of concurrency
    114 for _, coroutine in zip(range(len(done)), iterator):

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/tasks.py:232, in Task.__step(***failed resolving arguments***)
    228 try:
    229     if exc is None:
    230         # We use the send method directly, because coroutines
    231         # don't have __iter__ and __next__ methods.
--> 232         result = coro.send(None)
    233     else:
    234         result = coro.throw(exc)

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:582, in DAPSession.download_object(self, object, output_directory, decompress)
    570 async def download_object(
    571     self, object: Object, output_directory: str, decompress: bool = False
    572 ) -> str:
    573     """
    574     Save a single remote file to a local directory.
    575 
   (...)
    579     :returns: A path of the file saved in the local file system.
    580     """
--> 582     resource = (await self.get_resources([object]))[object.id]
    584     return await self.download_resource(resource, output_directory, decompress)

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:490, in DAPSession.get_resources(self, objects)
    487 logger.debug("Retrieve resource URLs for objects:")
    488 logger.debug([o.id for o in objects])
--> 490 response = await self._post("/dap/object/url", objects, ResourceResult)
    492 return response.urls

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:311, in DAPSession._post(self, path, request_data, response_type)
    304 logger.debug(f"POST request payload:\n{repr(request_payload)}")
    306 async with self._session.post(
    307     f"{self._base_url}{path}",
    308     data=json_dump_string(request_payload),
    309     headers={"Content-Type": "application/json"},
    310 ) as response:
--> 311     return await self._process(response, response_type)

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:348, in DAPSession._process(self, response, response_type, suppress_output)
    345     if response_text:
    346         logger.error(f"malformed HTTP response:\n{response_text}")
--> 348     raise DAPClientError("malformed HTTP response")
    350 if not suppress_output:
    351     logger.debug(f"GET/POST response payload:\n{repr(response_payload)}")

DAPClientError: malformed HTTP response

and from Today this new error occurred 'ProcessingError'

Operation on target Incremental Load failed: ---------------------------------------------------------------------------
ProcessingError                           Traceback (most recent call last)
Cell In [51], line 3
      1 if __name__ == '__main__':
      2     loop = asyncio.get_event_loop()
----> 3     res = loop.run_until_complete(main())
      4     try:
      5         loop.close()    

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/nest_asyncio.py:90, in _patch_loop.<locals>.run_until_complete(self, future)
     87 if not f.done():
     88     raise RuntimeError(
     89         'Event loop stopped before Future completed.')
---> 90 return f.result()

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/futures.py:201, in Future.result(self)
    199 self.__log_traceback = False
    200 if self._exception is not None:
--> 201     raise self._exception.with_traceback(self._exception_tb)
    202 return self._result

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/tasks.py:232, in Task.__step(***failed resolving arguments***)
    228 try:
    229     if exc is None:
    230         # We use the `send` method directly, because coroutines
    231         # don't have `__iter__` and `__next__` methods.
--> 232         result = coro.send(None)
    233     else:
    234         result = coro.throw(exc)

Cell In [47], line 3, in main()
      1 async def main():
      2     async with ClientSession() as session:
----> 3         data_d = await asyncio.gather(*[download_and_upload_to_blob_logs(table, session) for table in valid_tables_logs])

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/tasks.py:304, in Task.__wakeup(self, future)
    302 def __wakeup(self, future):
    303     try:
--> 304         future.result()
    305     except BaseException as exc:
    306         # This may also be a cancellation.
    307         self.__step(exc)

File ~/cluster-env/clonedenv/lib/python3.10/asyncio/tasks.py:232, in Task.__step(***failed resolving arguments***)
    228 try:
    229     if exc is None:
    230         # We use the `send` method directly, because coroutines
    231         # don't have `__iter__` and `__next__` methods.
--> 232         result = coro.send(None)
    233     else:
    234         result = coro.throw(exc)

Cell In [45], line 23, in download_and_upload_to_blob_logs(table, session)
     21             await asyncio.sleep(BACKOFF_FACTOR ** attempt)
     22         else:
---> 23             raise e
     25 # Create a BlobServiceClient
     26 blob_service_client = BlobServiceClient(
     27     account_url=account_url,
     28     credential=account_key
     29 )

Cell In [45], line 16, in download_and_upload_to_blob_logs(table, session)
      9     async with DAPClient() as session:
     10         query = IncrementalQuery(
     11             format=Format.Parquet,
     12             mode=None,
     13             since=last_seen,
     14             until=None,
     15         )
---> 16         await session.download_table_data("canvas_logs", table, query, output_directory)
     17     break
     18 except Exception as e:

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:668, in DAPSession.download_table_data(self, namespace, table, query, output_directory, decompress)
    665 # fail early if output directory does not exist and cannot be created
    666 os.makedirs(output_directory, exist_ok=True)
--> 668 job = await self.execute_job(namespace, table, query)
    670 if job.status is not JobStatus.Complete:
    671     raise DAPClientError(f"query job ended with status: {job.status.value}")

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:636, in DAPSession.execute_job(self, namespace, table, query)
    634     job = await self.query_snapshot(namespace, table, query)
    635 elif isinstance(query, IncrementalQuery):
--> 636     job = await self.query_incremental(namespace, table, query)
    637 else:
    638     raise TypeError(f"type mismatch for parameter `query`: {type(query)}")

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:404, in DAPSession.query_incremental(self, namespace, table, query)
    399 """
    400 Starts an incremental query.
    401 """
    403 logger.debug(f"Query updates for table: {table}")
--> 404 job = await self._post(f"/dap/query/{namespace}/table/{table}/data", query, Job)  # type: ignore
    405 return job

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:311, in DAPSession._post(self, path, request_data, response_type)
    304 logger.debug(f"POST request payload:\n{repr(request_payload)}")
    306 async with self._session.post(
    307     f"{self._base_url}{path}",
    308     data=json_dump_string(request_payload),
    309     headers={"Content-Type": "application/json"},
    310 ) as response:
--> 311     return await self._process(response, response_type)

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/dap/api.py:358, in DAPSession._process(self, response, response_type, suppress_output)
    356     error_object = self._map_to_error_type(response.status, response_payload)
    357     logger.warning(f"Received error in response: {error_object}")
--> 358     raise error_object
    359 else:
    360     response_object = json_to_object(response_type, response_payload)

ProcessingError:

dapclienterror malformed http response and processingerror occured while downloading the cd2 data

cd2 dap

DAP client library

EDUCAUSE Insights: Data and Decision-Making

Problems with GraphQL in User Pageviews reports

Metrics Easy Button

Analytics API / Metrics calculation logic

CD2: Enhanced Rubrics

Inconsistencies in user submission attempts (new q...

Seeking advice on CD2, ETL and presentation proces...

Sample Data

Canvas Data Access Platform (DAP) Python Client Li...

Seeking Advice: Integrating CD2 Data for Student &...

You're signed out

dapclienterror malformed http response and processingerror occured while downloading the cd2 data

Community help

View our top guides and resources: