Usage

Basic usage

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("SELECT * FROM one_row")
print(cursor.description)
print(cursor.fetchall())

Managed query result storage

When using a workgroup with managed query result storage enabled, you don’t need to specify an S3 staging directory.

from pyathena import connect

cursor = connect(work_group="YOUR_MANAGED_WORK_GROUP",
                 region_name="us-west-2").cursor()
cursor.execute("SELECT * FROM one_row")
print(cursor.fetchall())

If the AWS_ATHENA_S3_STAGING_DIR environment variable is set, pass s3_staging_dir="" to explicitly disable the fallback. Otherwise the API will reject the request because ResultConfiguration and ManagedQueryResultsConfiguration cannot be set together.

cursor = connect(work_group="YOUR_MANAGED_WORK_GROUP",
                 s3_staging_dir="",
                 region_name="us-west-2").cursor()

Note

With managed query result storage, query results are retrieved via the GetQueryResults API (1000 rows per request) instead of reading S3 files directly. This may be slower for large result sets. For large datasets, consider using customer-managed storage or the UNLOAD statement.

Cursor iteration

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("SELECT * FROM many_rows LIMIT 10")
for row in cursor:
    print(row)

Query with parameters

Supported DB API paramstyle is only PyFormat. PyFormat only supports named placeholders with old % operator style and parameters specify dictionary format.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("""
               SELECT col_string FROM one_row_complex
               WHERE col_string = %(param)s
               """,
               {"param": "a string"})
print(cursor.fetchall())

if % character is contained in your query, it must be escaped with %% like the following:

SELECT col_string FROM one_row_complex
WHERE col_string = %(param)s OR col_string LIKE 'a%%'

A datetime parameter is rendered as a TIMESTAMP literal that keeps its microseconds. A value without a sub-millisecond part is rendered with three fractional digits as a timestamp(3) literal, and any other value with six as a timestamp(6) literal.

from datetime import datetime

cursor.execute("SELECT %(a)s, %(b)s",
               {"a": datetime(2024, 1, 1, 12, 0, 0, 789000),
                "b": datetime(2024, 1, 1, 12, 0, 0, 789012)})
# SELECT TIMESTAMP '2024-01-01 12:00:00.789', TIMESTAMP '2024-01-01 12:00:00.789012'

Athena pads a timestamp(3) value with zeros when it meets a timestamp(6) value, and truncates a value written to a lower precision. Iceberg TIMESTAMP columns store microseconds, while Hive TIMESTAMP columns store milliseconds and truncate sub-millisecond digits on write. A timestamp(6) value with a sub-millisecond part therefore does not equal the value stored in a Hive column. CREATE TABLE AS SELECT into a Hive table and UNLOAD with Parquet output, including the UNLOAD that cursors run with unload=True, reject a timestamp(6) output column with NOT_SUPPORTED: Incorrect timestamp precision for timestamp(6); the value can still be used in a WHERE clause. Cast such a column to TIMESTAMP(3) to write it with millisecond precision:

SELECT CAST(%(param)s AS TIMESTAMP(3)) AS col_timestamp

Use parameterized queries

If you want to use Athena’s parameterized queries, you can do so by changing the paramstyle to qmark as follows.

import pyathena
from pyathena import connect

pyathena.paramstyle = "qmark"
cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("""
               SELECT col_string FROM one_row_complex
               WHERE col_string = ?
               """,
               ["'a string'"])
print(cursor.fetchall())

You can also specify the paramstyle using the execute method when executing a query.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("""
               SELECT col_string FROM one_row_complex
               WHERE col_string = ?
               """,
               ["'a string'"],
               paramstyle="qmark")
print(cursor.fetchall())

You can find more information about the considerations and limitations of parameterized queries in the official documentation.

Execution options

The execute() method of every SQL cursor (Cursor, AsyncCursor, the aio cursors, and their pandas/arrow/polars/s3fs variants) accepts the same set of shared keyword arguments, such as work_group, s3_staging_dir, cache_size, cache_expiration_time, result_reuse_enable, result_reuse_minutes, paramstyle, and result_set_type_hints. The synchronous and aio cursors also accept on_start_query_execution. AsyncCursor and its variants return the query ID from execute() instead; they do not take on_start_query_execution as a keyword argument and ignore it in options. The Spark cursors execute calculations instead of SQL queries and do not accept these arguments.

These arguments can also be passed together as an ExecuteOptions instance using the options keyword argument.

from pyathena import ExecuteOptions, connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
options = ExecuteOptions(work_group="YOUR_WORK_GROUP",
                         result_reuse_enable=True,
                         result_reuse_minutes=60)
cursor.execute("SELECT * FROM one_row", options=options)

When both options and individual keyword arguments are specified, the individual keyword arguments take precedence.

# Executes with work_group="ANOTHER_WORK_GROUP"; the other fields of options still apply.
cursor.execute("SELECT * FROM one_row",
               options=options,
               work_group="ANOTHER_WORK_GROUP")

Passing None for an individual keyword argument is treated as “not specified” and does not reset the corresponding options field. To run a single query without a field set on options, construct a new instance without that field.

ExecuteOptions is immutable. To create a variation of an existing instance, use the merge() method, which returns a new instance with the specified fields applied.

adhoc_options = options.merge(work_group="ANOTHER_WORK_GROUP")

The on_start_query_execution field is invoked by the synchronous and aio cursors; AsyncCursor-based cursors return the query ID directly through their execution model and do not invoke it.

Quickly re-run queries

Result reuse configuration

Athena engine version 3 allows you to reuse the results of previous queries.

It is available by specifying the arguments result_reuse_enable and result_reuse_minutes in the connection object.

from pyathena import connect

conn = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
               region_name="us-west-2",
               work_group="YOUR_WORK_GROUP",
               result_reuse_enable=True,
               result_reuse_minutes=60)

You can also specify result_reuse_enable and result_reuse_minutes when executing a query.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("SELECT * FROM one_row",
               work_group="YOUR_WORK_GROUP",
               result_reuse_enable=True,
               result_reuse_minutes=60)

If the following error occurs, please use a workgroup configured with Athena engine version 3.

pyathena.error.DatabaseError: An error occurred (InvalidRequestException) when calling the StartQueryExecution operation: This functionality is not enabled in the selected engine version. Please check the engine version settings or contact AWS support for further assistance.

If for some reason you cannot use the reuse feature of Athena engine version 3, please use the Cache configuration implemented by PyAthena.

Note

Semantic-equivalence matching (November 2025)

As of November 2025, Athena’s result reuse no longer requires byte-for-byte identical query text. Result reuse is now triggered when a query is semantically equivalent to a previously executed query — cosmetic differences such as whitespace, comments, or keyword casing no longer prevent a cache hit.

This behavior change happens server-side; PyAthena does not need to be updated, and existing code using result_reuse_enable continues to work unchanged. Users who relied on exact-string matching for cache invalidation should be aware that more queries may now reuse prior results than before.

See the Athena documentation for the latest rules on what counts as semantically equivalent.

Cache configuration

Please use the Result reuse configuration.

You can attempt to re-use the results from a previously executed query to help save time and money in the cases where your underlying data isn’t changing. Set the cache_size or cache_expiration_time parameter of cursor.execute() to a number larger than 0 to enable caching. cache_size is the number of the most recent query executions in the work group to search, including executions by other clients of the same work group. In a busy work group, a previous execution may no longer be among them.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("SELECT * FROM one_row")  # run once
print(cursor.query_id)
cursor.execute("SELECT * FROM one_row", cache_size=10)  # re-use earlier results
print(cursor.query_id)  # You should expect to see the same Query ID

The unit of cache_expiration_time is seconds. To use the results of queries executed up to one hour ago, specify like the following.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("SELECT * FROM one_row", cache_expiration_time=3600)  # Use queries executed within 1 hour as cache.

If cache_size is not specified, the value of sys.maxsize will be automatically set and all query results executed up to one hour ago will be checked. Therefore, it is recommended to specify cache_expiration_time together with cache_size like the following.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute("SELECT * FROM one_row", cache_size=100, cache_expiration_time=3600)  # Use the last 100 queries within 1 hour as cache.

Results will only be re-used from a succeeded DML query (the assumption being that you always want to re-run queries like CREATE TABLE and DROP TABLE) whose query string (with pyformat parameters substituted) matches exactly, and that ran with the same schema and catalog as the cursor. The cache is not used for a qmark query with parameters. With unload=True on the pandas, Arrow, and Polars cursors, a query that is wrapped in UNLOAD is written to a new location each time, so the cache never matches it.

The S3 staging directory is not checked, so it’s possible that the location of the results is not in your provided s3_staging_dir.

Federated passthrough queries

Athena’s federated query feature lets you query data in external sources (PostgreSQL, MySQL, Snowflake, etc.) through Lambda-based connectors. By default Athena’s planner translates your SQL into the source dialect, but for complex queries — or when you need a feature only the source supports — you can use passthrough queries to push native SQL directly to the source.

Passthrough queries use the TABLE(<connector>.system.query(query => '...')) syntax. Because PyAthena passes the SQL string through to Athena unchanged, no client-side changes are required — these queries work out of the box.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

cursor.execute("""
    SELECT * FROM TABLE(
        my_postgres_connector.system.query(
            query => 'SELECT id, name FROM public.users WHERE created_at > NOW() - INTERVAL ''7 days'''
        )
    )
""")
print(cursor.fetchall())

The connector (my_postgres_connector above) must be registered in Athena as a data source connector first; see the Athena documentation for the connector setup steps and the list of connectors that support passthrough.

Note

Result columns from a passthrough query come back with the source system’s types, which Athena maps to its own type system. When using PandasCursor or ArrowCursor, verify that the resulting dtypes match your expectations — connector-specific types (e.g., PostgreSQL json, MySQL geometry) may be returned as strings.

Query execution callback

PyAthena provides a callback mechanism that allows you to get immediate access to the query ID as soon as the start_query_execution API call is made, before waiting for query completion. This is useful for monitoring, logging, or cancelling long-running queries from another thread. When cache_size finds a reusable query, no new query starts, and the callback receives the reused query’s ID.

The on_start_query_execution callback can be configured at both the connection level and the execute level. When both are set, both callbacks will be invoked.

Connection-level callback

You can set a default callback for all queries executed through a connection:

from pyathena import connect

def query_callback(query_id):
    print(f"Query started with ID: {query_id}")
    # You can use query_id for monitoring or cancellation

cursor = connect(
    s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
    region_name="us-west-2",
    on_start_query_execution=query_callback
).cursor()

cursor.execute("SELECT * FROM many_rows")  # Callback will be invoked

Execute-level callback

You can also specify a callback for individual query executions:

from pyathena import connect

def specific_callback(query_id):
    print(f"Specific query started: {query_id}")

cursor = connect(
    s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
    region_name="us-west-2"
).cursor()

cursor.execute(
    "SELECT * FROM many_rows",
    on_start_query_execution=specific_callback
)

Query cancellation example

A common use case is to cancel long-running analytical queries after a timeout:

import threading
from pyathena import connect

def cancel_long_running_query():
    """Example: Cancel a complex analytical query after 10 minutes."""
    timeout_minutes = 10

    def track_query_start(query_id):
        print(f"Long-running analysis started: {query_id}")

    def cancel_on_timeout(cursor):
        """Cancel the query that is still running when the timer fires."""
        try:
            cursor.cancel()
            print(f"Query cancelled after {timeout_minutes} minutes timeout")
        except Exception as e:
            print(f"Cancellation failed: {e}")

    cursor = connect(
        s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
        region_name="us-west-2",
        on_start_query_execution=track_query_start
    ).cursor()

    # Complex analytical query that might run for a long time
    long_query = """
    WITH daily_metrics AS (
        SELECT
            date_trunc('day', timestamp_col) as day,
            user_id,
            COUNT(*) as events,
            AVG(duration) as avg_duration
        FROM large_events_table
        WHERE timestamp_col >= current_date - interval '1' year
        GROUP BY 1, 2
    ),
    user_segments AS (
        SELECT
            user_id,
            CASE
                WHEN AVG(events) > 100 THEN 'high_activity'
                WHEN AVG(events) > 10 THEN 'medium_activity'
                ELSE 'low_activity'
            END as segment
        FROM daily_metrics
        GROUP BY user_id
    )
    SELECT
        segment,
        COUNT(DISTINCT dm.user_id) as users,
        AVG(events) as avg_daily_events
    FROM daily_metrics dm
    JOIN user_segments us ON dm.user_id = us.user_id
    GROUP BY segment
    ORDER BY avg_daily_events DESC
    """

    # Cancel the query if it is still running after the timeout
    timer = threading.Timer(timeout_minutes * 60, cancel_on_timeout, args=(cursor,))
    timer.start()
    try:
        print("Starting complex analytical query (10-minute timeout)...")
        cursor.execute(long_query)

        # Process results
        results = cursor.fetchall()
        print(f"Analysis completed successfully: {len(results)} segments found")
        for row in results:
            print(f"  {row[0]}: {row[1]} users, {row[2]:.1f} avg events")

    except Exception as e:
        print(f"Query failed or was cancelled: {e}")
    finally:
        # Stop the timer once the query has finished
        timer.cancel()

# Run the example
cancel_long_running_query()

Multiple callbacks

When both connection-level and execute-level callbacks are specified, both callbacks will be invoked:

from pyathena import connect

def connection_callback(query_id):
    print(f"Connection callback: {query_id}")
    # Log to monitoring system

def execute_callback(query_id):
    print(f"Execute callback: {query_id}")
    # Store for cancellation if needed

cursor = connect(
    s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
    region_name="us-west-2",
    on_start_query_execution=connection_callback
).cursor()

# This will invoke both connection_callback and execute_callback
cursor.execute(
    "SELECT 1",
    on_start_query_execution=execute_callback
)

Supported cursor types

The on_start_query_execution callback is supported by the following cursor types:

• Cursor (default cursor)

• DictCursor

• ArrowCursor

• PandasCursor

• PolarsCursor

• S3FSCursor

• AioCursor, AioDictCursor, AioArrowCursor, AioPandasCursor, AioPolarsCursor, AioS3FSCursor

Note: AsyncCursor and its variants do not support this callback as they already return the query ID immediately through their different execution model.

Query cancellation on interrupt

With kill_on_interrupt enabled, which is the default, a KeyboardInterrupt while execute() waits for the query requests cancellation, waits until the query reaches a terminal state, and then propagates. Cancellation is a best-effort request, so the query can still end as SUCCEEDED or FAILED. The query_id property keeps the ID of the interrupted query. If the cancellation request or that wait fails, the KeyboardInterrupt propagates with the error as its cause.

A KeyboardInterrupt while execute() is still starting the query first waits for the StartQueryExecution request to finish, and then cancels the query it started in the same way. The query_id property returns that query’s ID. If execute() has not begun the request when the interrupt is handled, the request is never sent. AsyncCursor and its variants also stop a query whose start is interrupted in execute(), but they have no query_id property, so the ID of that query is not available. They wait for queries on worker threads, which do not receive KeyboardInterrupt.

A second KeyboardInterrupt during the cancellation request or these waits propagates immediately, and the query can keep running. With kill_on_interrupt=False, the KeyboardInterrupt propagates immediately, and a query that has already started keeps running.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
try:
    cursor.execute("SELECT * FROM many_rows")
except KeyboardInterrupt:
    print(f"Query {cursor.query_id} was interrupted")
    raise

For the native asyncio cursors, see Task cancellation.

Query polling callback

PyAthena provides an on_poll callback that is invoked once per poll iteration with the current query execution object, while PyAthena waits for the query to finish. This is useful for rendering live query progress (state, elapsed time, data scanned) in interactive environments such as Jupyter notebooks.

The callback is optional (None by default), so there is no impact on existing behaviour or performance when it is not used. It must be a synchronous function with the signature Callable[[AthenaQueryExecution], None] (for Spark calculations it receives an AthenaCalculationExecutionStatus). It is invoked on every poll, including the final iteration that observes the terminal state (SUCCEEDED, FAILED, or CANCELLED).

Unlike on_start_query_execution, on_poll is configured at the connection or cursor level only (there is no execute-level override).

Connection-level callback

from pyathena import connect

def on_poll(query_execution):
    print(
        f"State: {query_execution.state}, "
        f"scanned: {query_execution.data_scanned_in_bytes} bytes"
    )

cursor = connect(
    s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
    region_name="us-west-2",
    on_poll=on_poll,
).cursor()

cursor.execute("SELECT * FROM many_rows")  # on_poll is invoked on each poll

Cursor-level callback

from pyathena import connect

def on_poll(query_execution):
    print(f"State: {query_execution.state}")

conn = connect(
    s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
    region_name="us-west-2",
)
cursor = conn.cursor(on_poll=on_poll)
cursor.execute("SELECT * FROM many_rows")

Asynchronous cursors

on_poll also works with the asynchronous cursors. Because polling runs in a background thread (or event loop), the callback runs there too, so keep it lightweight and thread-safe:

from pyathena import connect
from pyathena.pandas.async_cursor import AsyncPandasCursor

def on_poll(query_execution):
    print(f"State: {query_execution.state}")

conn = connect(
    s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
    region_name="us-west-2",
)
cursor = conn.cursor(AsyncPandasCursor, on_poll=on_poll)
query_id, future = cursor.execute("SELECT * FROM many_rows")
result = future.result()

Supported cursor types

on_poll is supported by all cursor types, since polling is shared by the base cursor: the synchronous cursors, the Async* cursors, the native-async Aio* cursors, and the Spark cursors. For Spark cursors the callback receives the per-poll AthenaCalculationExecutionStatus rather than an AthenaQueryExecution.

Type hints for complex types

New in version 3.30.0.

The Athena API does not return element-level type information for complex types (array, map, row/struct). PyAthena parses the string representation returned by Athena, but without type metadata the converter can only apply heuristics — which may produce incorrect Python types for nested values (e.g. integers left as strings inside a struct).

The result_set_type_hints parameter solves this by letting you provide Athena DDL type signatures for specific columns. The converter then uses precise, recursive type-aware conversion instead of heuristics.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()
cursor.execute(
    "SELECT col_array, col_map, col_struct FROM one_row_complex",
    result_set_type_hints={
        "col_array": "array(integer)",
        "col_map": "map(integer, integer)",
        "col_struct": "row(a integer, b integer)",
    },
)
row = cursor.fetchone()
# col_struct values are now integers, not strings:
# {"a": 1, "b": 2} instead of {"a": "1", "b": "2"}

Column name matching is case-insensitive. Type hints support arbitrarily nested types:

cursor.execute(
    """
    SELECT CAST(
      ROW(ROW('2024-01-01', 123), 4.736, 0.583)
      AS ROW(header ROW(stamp VARCHAR, seq INTEGER), x DOUBLE, y DOUBLE)
    ) AS positions
    """,
    result_set_type_hints={
        "positions": "row(header row(stamp varchar, seq integer), x double, y double)",
    },
)
row = cursor.fetchone()
positions = row[0]
# positions["header"]["seq"] == 123 (int, not "123")
# positions["x"] == 4.736 (float, not "4.736")

Hive-style syntax

You can paste type signatures from Hive DDL or DESCRIBE TABLE output directly. Hive-style angle brackets and colons are automatically converted to Trino-style syntax:

# Both are equivalent:
result_set_type_hints={"col": "array(struct(a integer, b varchar))"}   # Trino
result_set_type_hints={"col": "array<struct<a:int,b:varchar>>"}        # Hive

The int alias is also supported and resolves to integer.

Index-based hints for duplicate column names

When a query produces columns with the same alias (e.g. SELECT a AS x, b AS x), name-based hints cannot distinguish between them. Use integer keys to specify hints by zero-based column position:

cursor.execute(
    "SELECT a AS x, b AS x FROM my_table",
    result_set_type_hints={
        0: "array(integer)",   # first "x" column
        1: "map(varchar, integer)",  # second "x" column
    },
)

Integer (index-based) hints take priority over string (name-based) hints for the same column. You can mix both styles in the same dictionary.

Constraints

• Nested arrays in native format — Athena’s native (non-JSON) string representation does not clearly delimit nested arrays. If your query returns nested arrays (e.g. array(array(integer))), use CAST(... AS JSON) in your query to get JSON-formatted output, which is parsed reliably.

• Arrow, Pandas, and Polars cursors — These cursors accept result_set_type_hints but their converters do not currently use the hints because they rely on their own type systems. The parameter is passed through for forward compatibility and for result sets that fall back to the default conversion path.

Breaking change in 3.30.0

Prior to 3.30.0, PyAthena attempted to infer Python types for scalar values inside complex types using heuristics (e.g. "123" → 123). Starting with 3.30.0, values inside complex types are kept as strings unless result_set_type_hints is provided. This change avoids silent misconversion but means existing code that relied on the heuristic behavior may see string values where it previously saw integers or floats.

To restore typed conversion, pass result_set_type_hints with the appropriate type signatures for the affected columns.

Table and database metadata

get_table_metadata(), list_table_metadata(), and list_databases() call the Athena metadata API. Athena applies its metadata API rate limits per account, and they are not listed in Service Quotas. In AwsDataCatalog and S3 Tables catalogs (s3tablescatalog/<table-bucket>), a throttled request is answered from the AWS Glue Data Catalog instead, and a warning is logged. The request goes to Glue on the first throttled response, without waiting for the retry policy. Glue throttling that Athena reports inside a MetadataException counts as throttled. The fallback calls these Glue APIs with the connection’s credentials:

Cursor method	Glue API	IAM action
get_table_metadata()	GetTable	glue:GetTable
list_table_metadata()	GetTables	glue:GetTables
list_databases()	GetDatabases	glue:GetDatabases

Glue’s report that the table does not exist, for get_table_metadata(), raises OperationalError, as Athena’s does. If the Glue request fails for any other reason, for example for lack of permission or because the Glue endpoint cannot be reached, a second warning is logged and the Athena request runs again with the retry policy. A request that cannot connect to Glue, such as from a network with an Athena VPC endpoint but no route to Glue, or that finds no Glue endpoint for the connection’s region and endpoint options, also turns the fallback off for the rest of that connection; a request that cannot connect first waits out the botocore connect timeout and retries of the connection’s config. For list_table_metadata() and list_databases(), Glue’s report that the database or catalog does not exist also returns to the Athena request. Requests in other catalogs use the Athena API with the retry policy only.

The connection builds one Glue client on first use from its session, region, and botocore config, but not its endpoint_url. Pass glue_metadata_fallback=False to connect() to turn the fallback off. The Glue request does not carry the connection’s workgroup; turn the fallback off where access depends on the workgroup, such as a workgroup enabled for IAM Identity Center.

Environment variables

Support Boto3 environment variables.

Additional environment variables

AWS_ATHENA_S3_STAGING_DIR

The S3 location where Athena automatically stores the query results and metadata information. Required if you have not set up workgroups. Not required if a workgroup has been set up. When connecting to a workgroup with managed query result storage, pass s3_staging_dir="" to explicitly disable this environment variable fallback (see Managed query result storage).

AWS_ATHENA_WORK_GROUP

The setting of the workgroup to execute the query.

Credentials

Support Boto3 credentials.

Examples

Passing credentials as parameters

from pyathena import connect

cursor = connect(aws_access_key_id="YOUR_ACCESS_KEY_ID",
                 aws_secret_access_key="YOUR_SECRET_ACCESS_KEY",
                 s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

from pyathena import connect

cursor = connect(aws_access_key_id="YOUR_ACCESS_KEY_ID",
                 aws_secret_access_key="YOUR_SECRET_ACCESS_KEY",
                 aws_session_token="YOUR_SESSION_TOKEN",
                 s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

Multi-factor authentication

You will be prompted to enter the MFA code. The program execution will be blocked until the MFA code is entered.

from pyathena import connect

cursor = connect(duration_seconds=3600,
                 serial_number="arn:aws:iam::ACCOUNT_NUMBER_WITHOUT_HYPHENS:mfa/MFA_DEVICE_ID",
                 s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

Shared credentials file

The shared credentials file has a default location of ~/.aws/credentials.

If you use the default profile, there is no need to specify credential information.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

You can also specify a profile other than the default.

from pyathena import connect

cursor = connect(profile_name="YOUR_PROFILE_NAME",
                 s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

Assume role provider

from pyathena import connect

cursor = connect(role_arn="YOUR_ASSUME_ROLE_ARN",
                 role_session_name="PyAthena-session",
                 duration_seconds=3600,
                 s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

Assume role provider with MFA

You will be prompted to enter the MFA code. The program execution will be blocked until the MFA code is entered.

from pyathena import connect

cursor = connect(role_arn="YOUR_ASSUME_ROLE_ARN",
                 role_session_name="PyAthena-session",
                 duration_seconds=3600,
                 serial_number="arn:aws:iam::ACCOUNT_NUMBER_WITHOUT_HYPHENS:mfa/MFA_DEVICE_ID",
                 s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

Instance profiles

No need to specify credential information.

from pyathena import connect

cursor = connect(s3_staging_dir="s3://YOUR_S3_BUCKET/path/to/",
                 region_name="us-west-2").cursor()

Unsupported: JWT Trusted Identity Propagation

Amazon Athena supports JWT-based Trusted Identity Propagation (TIP) for the official JDBC and ODBC drivers, allowing enterprise SSO identities (Okta, Entra ID, etc.) to be propagated to Athena and Lake Formation for fine-grained access control.

PyAthena does not support JWT TIP, because this auth flow is not exposed through the AWS SDK (boto3 / botocore). PyAthena builds its Athena client via boto3 and therefore relies on standard IAM-based credentials.

If your environment requires JWT TIP, the options are:

• Use the Athena JDBC driver or ODBC driver directly.

• Use IAM Identity Center with role-based access (assume-role flow) — see the Assume role provider examples above. This is not byte-equivalent to TIP but satisfies most SSO-driven access-control requirements.

This is a limitation of the AWS SDK, not of PyAthena. If boto3/botocore adds JWT TIP support in the future, PyAthena will expose it via Connection.