def on_request_body()

in callouts/python/extproc/example/e2e_tests/observability_server.py [0:0]

• 7 lines of code
• 3 McCabe index (conditional complexity)

  def on_request_body(self, body: service_pb2.HttpBody,
                      context: ServicerContext) -> BodyResponse:
    """Custom processor on the request body."""
    logging.info('on_request_body %s', body)
    with lock:
      if (not body.end_of_stream or body.body):
        counters['request_body_count'] += 1
    return BodyResponse()