I'm new to using pyathena and also SQLalchemy (or DBAPI in general). We are using pyathena to and SQLalchemy to query the data in our S3 bucket but we need to connect with our work_group rather than aws_access_id or secret key. The pyathena page in pypi gives the example below, but I need to be able to replace conn_str such that we don't use {aws_access_key_id} or {aws_secret_access_key} and use our Athena work group to connect and run queries. I've been searching online but it seems like a very rare issue, does anyone more familiar with Athena/AWS/pyathena/SQLAlchemy have any suggestions?
from urllib.parse import quote_plus
from sqlalchemy.engine import create_engine
from sqlalchemy.sql.expression import select
from sqlalchemy.sql.functions import func
from sqlalchemy.sql.schema import Table, MetaData
conn_str = "awsathena+rest://{aws_access_key_id}:{aws_secret_access_key}@athena.{region_name}.amazonaws.com:443/"\
"{schema_name}?s3_staging_dir={s3_staging_dir}"
engine = create_engine(conn_str.format(
aws_access_key_id=quote_plus("YOUR_ACCESS_KEY_ID"),
aws_secret_access_key=quote_plus("YOUR_SECRET_ACCESS_KEY"),
region_name="us-west-2",
schema_name="default",
s3_staging_dir=quote_plus("s3://YOUR_S3_BUCKET/path/to/")))
with engine.connect() as connection:
many_rows = Table("many_rows", MetaData(), autoload_with=connection)
result = connection.execute(select([func.count("*")], from_obj=many_rows))
print(result.scalar())
