class_docstr: Hook for Apache Livy through the REST API.
  
  :param livy_conn_id: reference to a pre-defined Livy Connection.
  :param extra_options: A dictionary of options passed to Livy.
  :param extra_headers: A dictionary of headers passed to the HTTP request to livy.
  :param auth_type: The auth type for the service.
  
  .. seealso::
      For more details refer to the Apache Livy API reference:
      https://livy.apache.org/docs/latest/rest-api.html
class_docstr_tok: ["Hook","for","Apache","Livy","through","the","REST","API",".",":","param","livy_conn_id",":","reference","to","a","pre-defined","Livy","Connection",".",":","param","extra_options",":","A","dictionary","of","options","passed","to","Livy",".",":","param","extra_headers",":","A","dictionary","of","headers","passed","to","the","HTTP","request","to","livy",".",":","param","auth_type",":","The","auth","type","for","the","service",".","..","seealso",":",":","For","more","details","refer","to","the","Apache","Livy","API","reference",":","https",":","\/\/livy.apache.org\/docs\/latest\/rest-api.html"]
docstr: Get the session logs for a specified batch.
  
  :param session_id: identifier of the batch sessions
  :param log_start_position: Position from where to pull the logs
  :param log_batch_size: Number of lines to pull in one batch
  
  :return: response body
docstr_tok: ["Get","the","session","logs","for","a","specified","batch",".",":","param","session_id",":","identifier","of","the","batch","sessions",":","param","log_start_position",":","Position","from","where","to","pull","the","logs",":","param","log_batch_size",":","Number","of","lines","to","pull","in","one","batch",":","return",":","response","body"]
code: def get_batch_logs(
      self, session_id: int | str, log_start_position, log_batch_size
  ) -> dict:
      """
      Get the session logs for a specified batch.
  
      :param session_id: identifier of the batch sessions
      :param log_start_position: Position from where to pull the logs
      :param log_batch_size: Number of lines to pull in one batch
  
      :return: response body
      """
      self._validate_session_id(session_id)
      log_params = {"from": log_start_position, "size": log_batch_size}
      response = self.run_method(
          endpoint=f"/batches/{session_id}/log",
          data=log_params,
          headers=self.extra_headers,
      )
      try:
          response.raise_for_status()
      except requests.exceptions.HTTPError as err:
          self.log.warning(
              "Got status code %d for session %s",
              err.response.status_code,
              session_id,
          )
          raise AirflowException(
              f"Could not fetch the logs for batch with session id: {session_id}. "
              f"Message: {err.response.text}"
          )
      return response.json()
  
imports: from __future__ import annotations
  import asyncio
  import json
  import re
  from enum import Enum
  from typing import TYPE_CHECKING, Any, Sequence
  import aiohttp
  import requests
  from aiohttp import ClientResponseError
  from asgiref.sync import sync_to_async
  from airflow.exceptions import AirflowException
  from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
  from airflow.utils.log.logging_mixin import LoggingMixin
label_desc: Use image node_id 8 for calling the LivyHook obj's underlying member method code with example usage: obj.get_batch_logs(session_id, log_start_position, log_batch_size) and returns: response