from airflow.configuration import conf
from airflow.decorators.base import get_unique_task_id
from airflow.models.xcom_arg import XComArg

from astro.sql.operators.append import AppendOperator, append
from astro.sql.operators.cleanup import CleanupOperator, cleanup
from astro.sql.operators.dataframe import DataframeOperator, dataframe
from astro.sql.operators.drop import DropTableOperator, drop_table
from astro.sql.operators.export_file import ExportFileOperator, export_file
from astro.sql.operators.load_file import LoadFileOperator, load_file
from astro.sql.operators.merge import MergeOperator, merge
from astro.sql.operators.raw_sql import RawSQLOperator, run_raw_sql
from astro.sql.operators.transform import TransformOperator, transform, transform_file
from astro.table import Metadata, Table

__all__ = [
    "AppendOperator",
    "append",
    "CleanupOperator",
    "cleanup",
    "DataframeOperator",
    "dataframe",
    "DropTableOperator",
    "drop_table",
    "ExportFileOperator",
    "export_file",
    "LoadFileOperator",
    "load_file",
    "MergeOperator",
    "merge",
    "Metadata",
    "run_raw_sql",
    "Table",
    "TransformOperator",
    "transform_file",
    "transform",
]


def get_value_list(sql: str, conn_id: str, **kwargs) -> XComArg:
    """
    Execute a sql statement and return the result.
    By default, the response size is less than equal to value of ``max_map_length`` conf.
    You can call a callable handler to alter the response by default it call ``fetchall`` on database result set.


    :param sql: sql query to execute.
        If the sql query will return huge number of row then it can overload the XCOM.
        also, If you are using output of this method to expand a task using dynamic task map then
        it can create lots of parallel task. So it is advisable to limit your sql query statement.
    :param conn_id: Airflow connection id. This connection id will be used to identify the database client
        and connect with it at runtime
    """
    handler = kwargs.get("handler") or (lambda result_set: result_set.fetchall())
    max_map_length = int(conf.get(section="core", key="max_map_length"))
    op_kwargs = {
        "handler": handler,
        "response_limit": max_map_length,
    }
    task_id = kwargs.get("task_id") or get_unique_task_id(
        "get_value_list", dag=kwargs.get("dag"), task_group=kwargs.get("task_group")
    )
    kwargs.update({"task_id": task_id})
    return RawSQLOperator(
        sql=sql, conn_id=conn_id, op_kwargs=op_kwargs, python_callable=(lambda *args: None), **kwargs
    ).output

python-sdk/src/astro/sql/__init__.py
from airflow.configuration import conf
from airflow.decorators.base import get_unique_task_id
from airflow.models.xcom_arg import XComArg

from astro.sql.operators.append import AppendOperator, append
from astro.sql.operators.cleanup import CleanupOperator, cleanup
from astro.sql.operators.dataframe import DataframeOperator, dataframe
from astro.sql.operators.drop import DropTableOperator, drop_table
from astro.sql.operators.export_file import ExportFileOperator, export_file
from astro.sql.operators.load_file import LoadFileOperator, load_file
from astro.sql.operators.merge import MergeOperator, merge
from astro.sql.operators.raw_sql import RawSQLOperator, run_raw_sql
from astro.sql.operators.transform import TransformOperator, transform, transform_file
from astro.table import Metadata, Table

__all__ = [
    "AppendOperator",
    "append",
    "CleanupOperator",
    "cleanup",
    "DataframeOperator",
    "dataframe",
    "DropTableOperator",
    "drop_table",
    "ExportFileOperator",
    "export_file",
    "get_value_list",
    "LoadFileOperator",
    "load_file",
    "MergeOperator",
    "merge",
    "Metadata",
    "run_raw_sql",
    "Table",
    "TransformOperator",
    "transform_file",
    "transform",
]


def get_value_list(sql: str, conn_id: str, **kwargs) -> XComArg:
    """
    Execute a sql statement and return the result.
    By default, the response size is less than equal to value of ``max_map_length`` conf.
    You can call a callable handler to alter the response by default it call ``fetchall`` on database result set.


    :param sql: sql query to execute.
        If the sql query will return huge number of row then it can overload the XCOM.
        also, If you are using output of this method to expand a task using dynamic task map then
        it can create lots of parallel task. So it is advisable to limit your sql query statement.
    :param conn_id: Airflow connection id. This connection id will be used to identify the database client
        and connect with it at runtime
    """
    handler = kwargs.get("handler") or (lambda result_set: result_set.fetchall())
    max_map_length = int(conf.get(section="core", key="max_map_length"))
    op_kwargs = {
        "handler": handler,
        "response_limit": max_map_length,
    }
    task_id = kwargs.get("task_id") or get_unique_task_id(
        "get_value_list", dag=kwargs.get("dag"), task_group=kwargs.get("task_group")
    )
    kwargs.update({"task_id": task_id})
    return RawSQLOperator(
        sql=sql, conn_id=conn_id, op_kwargs=op_kwargs, python_callable=(lambda *args: None), **kwargs
    ).output

python-sdk/src/astro/sql/__init__.py
diff --git a/python-sdk/docs/astro/files/operators/get_file_list.rst b/python-sdk/docs/astro/files/operators/get_file_list.rst
index 540e9dd6de..c210ca283d 100644
--- a/python-sdk/docs/astro/files/operators/get_file_list.rst
+++ b/python-sdk/docs/astro/files/operators/get_file_list.rst
@@ -1,8 +1,8 @@
 .. _get_file_list:
 
-=============
-get_file_list
-=============
+====================================================
+:py:func:`get_file_list <astro.files.get_file_list>`
+====================================================
 
 When to use the ``get_file_list`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/python-sdk/docs/astro/sql/operators/append.rst b/python-sdk/docs/astro/sql/operators/append.rst
index ca26662103..360ac29352 100644
--- a/python-sdk/docs/astro/sql/operators/append.rst
+++ b/python-sdk/docs/astro/sql/operators/append.rst
@@ -1,8 +1,8 @@
 .. _append_operator:
 
-================
-append operator
-================
+======================================================
+:py:mod:`append operator <astro.sql.operators.append>`
+======================================================
 When to use the ``append`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 The ``append`` operator allows you to append data from a source table to a target table.
diff --git a/python-sdk/docs/astro/sql/operators/cleanup.rst b/python-sdk/docs/astro/sql/operators/cleanup.rst
index ebe4dfc714..3c4556dd24 100644
--- a/python-sdk/docs/astro/sql/operators/cleanup.rst
+++ b/python-sdk/docs/astro/sql/operators/cleanup.rst
@@ -1,6 +1,6 @@
-======================================
-cleanup operator
-======================================
+========================================================
+:py:mod:`cleanup operator <astro.sql.operators.cleanup>`
+========================================================
 
 .. _cleanup_operator:
 
diff --git a/python-sdk/docs/astro/sql/operators/dataframe.rst b/python-sdk/docs/astro/sql/operators/dataframe.rst
index 9d532ec7a0..ad4fda0f93 100644
--- a/python-sdk/docs/astro/sql/operators/dataframe.rst
+++ b/python-sdk/docs/astro/sql/operators/dataframe.rst
@@ -1,8 +1,8 @@
 .. dataframe_operator:
 
-======================================
-dataframe operator
-======================================
+============================================================
+:py:mod:`dataframe operator <astro.sql.operators.dataframe>`
+============================================================
 
 When to use the ``dataframe`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/python-sdk/docs/astro/sql/operators/drop_table.rst b/python-sdk/docs/astro/sql/operators/drop_table.rst
index 00347172aa..74606cfc56 100644
--- a/python-sdk/docs/astro/sql/operators/drop_table.rst
+++ b/python-sdk/docs/astro/sql/operators/drop_table.rst
@@ -1,8 +1,8 @@
 .. _drop_table_operator:
 
-======================================
-drop table operator
-======================================
+========================================================
+:py:mod:`drop table operator <astro.sql.operators.drop>`
+========================================================
 
 When to use the ``drop_table`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/python-sdk/docs/astro/sql/operators/export.rst b/python-sdk/docs/astro/sql/operators/export.rst
index d7e863f8ca..734f93f324 100644
--- a/python-sdk/docs/astro/sql/operators/export.rst
+++ b/python-sdk/docs/astro/sql/operators/export.rst
@@ -1,8 +1,8 @@
 .. _export_file:
 
-====================
-export_file operator
-====================
+================================================================
+:py:mod:`export_file operator <astro.sql.operators.export_file>`
+================================================================
 
 .. _export_file_operator:
 
diff --git a/python-sdk/docs/astro/sql/operators/get_value_list.rst b/python-sdk/docs/astro/sql/operators/get_value_list.rst
index b17731af26..d1770e7311 100644
--- a/python-sdk/docs/astro/sql/operators/get_value_list.rst
+++ b/python-sdk/docs/astro/sql/operators/get_value_list.rst
@@ -1,8 +1,8 @@
 .. _get_value_list:
 
-==============
-get_value_list
-==============
+====================================================
+:py:func:`get_value_list <astro.sql.get_value_list>`
+====================================================
 
 When to use the ``get_value_list`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/python-sdk/docs/astro/sql/operators/load_file.rst b/python-sdk/docs/astro/sql/operators/load_file.rst
index 6cdc6b6d21..d06584907f 100644
--- a/python-sdk/docs/astro/sql/operators/load_file.rst
+++ b/python-sdk/docs/astro/sql/operators/load_file.rst
@@ -1,8 +1,8 @@
 .. _load_file:
 
-==================
-load_file operator
-==================
+============================================================
+:py:mod:`load_file operator <astro.sql.operators.load_file>`
+============================================================
 
 When to use the ``load_file`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/python-sdk/docs/astro/sql/operators/merge.rst b/python-sdk/docs/astro/sql/operators/merge.rst
index b184469fed..08f3f96d89 100644
--- a/python-sdk/docs/astro/sql/operators/merge.rst
+++ b/python-sdk/docs/astro/sql/operators/merge.rst
@@ -1,8 +1,8 @@
 .. _merge_operator:
 
-===============
-merge operator
-===============
+====================================================
+:py:mod:`merge operator <astro.sql.operators.merge>`
+====================================================
 
 When to use the ``merge`` operator
 ----------------------------------
diff --git a/python-sdk/docs/astro/sql/operators/raw_sql.rst b/python-sdk/docs/astro/sql/operators/raw_sql.rst
index e7ad1974c5..170318e2bb 100644
--- a/python-sdk/docs/astro/sql/operators/raw_sql.rst
+++ b/python-sdk/docs/astro/sql/operators/raw_sql.rst
@@ -1,8 +1,8 @@
 .. _run_raw_sql:
 
-====================
-run_raw_sql operator
-====================
+============================================================
+:py:mod:`run_raw_sql operator <astro.sql.operators.raw_sql>`
+============================================================
 
 When to use the ``run_raw_sql`` operator
 -----------------------------------------
diff --git a/python-sdk/docs/astro/sql/operators/transform.rst b/python-sdk/docs/astro/sql/operators/transform.rst
index 32aca41053..b1e9b7851c 100644
--- a/python-sdk/docs/astro/sql/operators/transform.rst
+++ b/python-sdk/docs/astro/sql/operators/transform.rst
@@ -1,8 +1,8 @@
 .. _transform_operator:
 
-==================
-transform operator
-==================
+============================================================
+:py:mod:`transform operator <astro.sql.operators.transform>`
+============================================================
 
 When to use the ``transform`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/python-sdk/docs/astro/sql/operators/transform_file.rst b/python-sdk/docs/astro/sql/operators/transform_file.rst
index e3f7a947cc..127d2dc32a 100644
--- a/python-sdk/docs/astro/sql/operators/transform_file.rst
+++ b/python-sdk/docs/astro/sql/operators/transform_file.rst
@@ -1,8 +1,8 @@
 .. _transform_file_operator:
 
-=======================
-transform_file operator
-=======================
+=================================================================================
+:py:func:`transform_file operator <astro.sql.operators.transform.transform_file>`
+=================================================================================
 
 When to use the ``transform_file`` operator
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
diff --git a/python-sdk/src/astro/sql/__init__.py b/python-sdk/src/astro/sql/__init__.py
index c8b12fc510..ffa1daef35 100644
--- a/python-sdk/src/astro/sql/__init__.py
+++ b/python-sdk/src/astro/sql/__init__.py
@@ -24,6 +24,7 @@
     "drop_table",
     "ExportFileOperator",
     "export_file",
+    "get_value_list",
     "LoadFileOperator",
     "load_file",
     "MergeOperator",

Doc: Cross link to API reference page from Operators page
Currently there is no way to jump to Func/Operator definition from https://astro-sdk-python.readthedocs.io/en/stable/astro/sql/operators/get_value_list.html (and other operators listed on https://astro-sdk-python.readthedocs.io/en/stable/guides/operators.html) to their definition https://astro-sdk-python.readthedocs.io/en/stable/autoapi/index.html

We should cross-link them using https://www.sphinx-doc.org/en/master/usage/restructuredtext/domains.html#cross-referencing-syntax
