Skip to content

Pass WorkGroup parameter to GetTableMetadata and ListTableMetadata APIs #646

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
laughingman7743 merged 2 commits into from
Jan 16, 2026
Merged

Pass WorkGroup parameter to GetTableMetadata and ListTableMetadata APIs #646

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
58a14c8
Pass WorkGroup parameter to GetTableMetadata and ListTableMetadata APIs
laughingman7743 Jan 16, 2026
60e1a09
Add WorkGroup parameter to ListDatabases API
laughingman7743 Jan 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion pyathena/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,8 @@ def _build_list_table_metadata_request(
request.update({"Expression": expression})
if next_token:
request.update({"NextToken": next_token})
if self._work_group:
request.update({"WorkGroup": self._work_group})
return request

def _build_list_databases_request(
Expand All @@ -306,6 +308,8 @@ def _build_list_databases_request(
}
if next_token:
request.update({"NextToken": next_token})
if self._work_group:
request.update({"WorkGroup": self._work_group})
return request

def _list_databases(
Expand Down Expand Up @@ -359,11 +363,13 @@ def _get_table_metadata(
schema_name: Optional[str] = None,
logging_: bool = True,
) -> AthenaTableMetadata:
request = {
request: Dict[str, Any] = {
"CatalogName": catalog_name if catalog_name else self._catalog_name,
"DatabaseName": schema_name if schema_name else self._schema_name,
"TableName": table_name,
}
if self._work_group:
request.update({"WorkGroup": self._work_group})
try:
response = retry_api_call(
self._connection.client.get_table_metadata,
Expand Down