Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
3 changes: 2 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ ignore_missing_imports = false

[tool.ruff]
line-length = 120
include = ["examples/**/*.py", "src/**/*.py", "tests/**/*.py", "tests-integ/**/*.py"]
include = ["examples/**/*.py", "src/**/*.py", "tests/**/*.py", "tests_integ/**/*.py"]
exclude = ["**/*.md"]

[tool.ruff.lint]
Expand All @@ -79,6 +79,7 @@ select = [
"G", # logging format
"I", # isort
"LOG", # logging
"UP", # pyupgrade
]

[tool.ruff.lint.per-file-ignores]
Expand Down
19 changes: 10 additions & 9 deletions src/bedrock_agentcore/identity/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,9 @@
import contextvars
import logging
import os
from collections.abc import Callable
from functools import wraps
from typing import Any, Callable, Dict, List, Literal, Optional
from typing import Any, Literal

import boto3

Expand All @@ -22,14 +23,14 @@ def requires_access_token(
*,
provider_name: str,
into: str = "access_token",
scopes: List[str],
on_auth_url: Optional[Callable[[str], Any]] = None,
scopes: list[str],
on_auth_url: Callable[[str], Any] | None = None,
auth_flow: Literal["M2M", "USER_FEDERATION"],
callback_url: Optional[str] = None,
callback_url: str | None = None,
force_authentication: bool = False,
token_poller: Optional[TokenPoller] = None,
custom_state: Optional[str] = None,
custom_parameters: Optional[Dict[str, str]] = None,
token_poller: TokenPoller | None = None,
custom_state: str | None = None,
custom_parameters: dict[str, str] | None = None,
) -> Callable:
"""Decorator that fetches an OAuth2 access token before calling the decorated function.

Expand Down Expand Up @@ -151,7 +152,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
return decorator


def _get_oauth2_callback_url(user_provided_oauth2_callback_url: Optional[str]):
def _get_oauth2_callback_url(user_provided_oauth2_callback_url: str | None):
if user_provided_oauth2_callback_url:
return user_provided_oauth2_callback_url

Expand Down Expand Up @@ -184,7 +185,7 @@ async def _set_up_local_auth(client: IdentityClient) -> str:
config = {}
if config_path.exists():
try:
with open(config_path, "r", encoding="utf-8") as file:
with open(config_path, encoding="utf-8") as file:
config = json.load(file) or {}
except Exception:
print("Could not find existing workload identity and user id")
Expand Down
Loading