class_docstr: Allows a DAG to "branch" or follow a specified path based on the results of a SQL query.
  
  :param sql: The SQL code to be executed, should return true or false (templated)
     Template reference are recognized by str ending in '.sql'.
     Expected SQL query to return a boolean (True/False), integer (0 = False, Otherwise = 1)
     or string (true/y/yes/1/on/false/n/no/0/off).
  :param follow_task_ids_if_true: task id or task ids to follow if query returns true
  :param follow_task_ids_if_false: task id or task ids to follow if query returns false
  :param conn_id: the connection ID used to connect to the database.
  :param database: name of database which overwrite the defined one in connection
  :param parameters: (optional) the parameters to render the SQL query with.
class_docstr_tok: ["Allows","a","DAG","to","``","branch","''","or","follow","a","specified","path","based","on","the","results","of","a","SQL","query",".",":","param","sql",":","The","SQL","code","to","be","executed",",","should","return","true","or","false","(","templated",")","Template","reference","are","recognized","by","str","ending","in","'.sql","'",".","Expected","SQL","query","to","return","a","boolean","(","True\/False",")",",","integer","(","0","=","False",",","Otherwise","=","1",")","or","string","(","true\/y\/yes\/1\/on\/false\/n\/no\/0\/off",")",".",":","param","follow_task_ids_if_true",":","task","id","or","task","ids","to","follow","if","query","returns","true",":","param","follow_task_ids_if_false",":","task","id","or","task","ids","to","follow","if","query","returns","false",":","param","conn_id",":","the","connection","ID","used","to","connect","to","the","database",".",":","param","database",":","name","of","database","which","overwrite","the","defined","one","in","connection",":","param","parameters",":","(","optional",")","the","parameters","to","render","the","SQL","query","with","."]
code: def execute(self, context: Context):
      self.log.info(
          "Executing: %s (with parameters %s) with connection: %s",
          self.sql,
          self.parameters,
          self.conn_id,
      )
      record = self.get_db_hook().get_first(self.sql, self.parameters)
      if not record:
          raise AirflowException(
              "No rows returned from sql query. Operator expected True or False return value."
          )
  
      if isinstance(record, list):
          if isinstance(record[0], list):
              query_result = record[0][0]
          else:
              query_result = record[0]
      elif isinstance(record, tuple):
          query_result = record[0]
      else:
          query_result = record
  
      self.log.info(
          "Query returns %s, type '%s'",
          query_result,
          type(query_result),
      )
  
      follow_branch = None
      try:
          if isinstance(query_result, bool):
              if query_result:
                  follow_branch = self.follow_task_ids_if_true
          elif isinstance(query_result, str):
              # return result is not Boolean, try to convert from String to Boolean
              if _parse_boolean(query_result):
                  follow_branch = self.follow_task_ids_if_true
          elif isinstance(query_result, int):
              if bool(query_result):
                  follow_branch = self.follow_task_ids_if_true
          else:
              raise AirflowException(
                  f"Unexpected query return result '{query_result}' type '{type(query_result)}'"
              )
  
          if follow_branch is None:
              follow_branch = self.follow_task_ids_if_false
      except ValueError:
          raise AirflowException(
              f"Unexpected query return result '{query_result}' type '{type(query_result)}'"
          )
  
      self.skip_all_except(context["ti"], follow_branch)
  
imports: from __future__ import annotations
  import ast
  import re
  from functools import cached_property
  from typing import TYPE_CHECKING, Any, Callable, Iterable, Mapping, NoReturn, Sequence, SupportsAbs
  from airflow.exceptions import AirflowException, AirflowFailException
  from airflow.hooks.base import BaseHook
  from airflow.models import BaseOperator, SkipMixin
  from airflow.providers.common.sql.hooks.sql import DbApiHook, fetch_all_handler, return_single_query_results
  from airflow.utils.helpers import merge_dicts
label_desc: Use image node_id 2 for calling the BranchSQLOperator obj's underlying member method code with example usage: obj.execute(context) without return types