Skip to content
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.

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

chore: Define Auth type alias in types module #2586

Merged
merged 4 commits into from
Aug 5, 2024
Merged
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
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,8 @@
{%- if cookiecutter.auth_method in ("OAuth2", "JWT") %}
from functools import cached_property
{%- endif %}
from typing import TYPE_CHECKING, Any, Callable, Iterable
from typing import TYPE_CHECKING, Any, Iterable

import requests
{% if cookiecutter.auth_method == "API Key" -%}
from singer_sdk.authenticators import APIKeyAuthenticator
from singer_sdk.helpers.jsonpath import extract_jsonpath
Expand Down Expand Up @@ -47,11 +46,14 @@
import importlib_resources

if TYPE_CHECKING:
import requests
{%- if cookiecutter.auth_method in ("OAuth2", "JWT") %}
from singer_sdk.helpers.types import Auth, Context
{%- else %}
from singer_sdk.helpers.types import Context
{%- endif %}


_Auth = Callable[[requests.PreparedRequest], requests.PreparedRequest]

# TODO: Delete this is if not using json files for schema definition
SCHEMAS_DIR = importlib_resources.files(__package__) / "schemas"

Expand All @@ -74,7 +76,7 @@ def url_base(self) -> str:
{%- if cookiecutter.auth_method in ("OAuth2", "JWT") %}

@cached_property
def authenticator(self) -> _Auth:
def authenticator(self) -> Auth:
"""Return a new authenticator object.

Returns:
Expand Down
2 changes: 2 additions & 0 deletions noxfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@

RUFF_OVERRIDES = """\
extend = "./pyproject.toml"

[lint]
extend-ignore = ["TD002", "TD003", "FIX002"]
"""

Expand Down
4 changes: 4 additions & 0 deletions singer_sdk/helpers/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import sys
import typing as t

import requests

if sys.version_info < (3, 9):
from typing import Mapping # noqa: ICN003
else:
Expand All @@ -15,10 +17,12 @@
else:
from typing import TypeAlias # noqa: ICN003


__all__ = [
"Context",
"Record",
]

Context: TypeAlias = Mapping[str, t.Any]
Record: TypeAlias = t.Dict[str, t.Any]
Auth: TypeAlias = t.Callable[[requests.PreparedRequest], requests.PreparedRequest]
11 changes: 2 additions & 9 deletions singer_sdk/streams/rest.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,25 +26,18 @@
from singer_sdk.streams.core import Stream

if t.TYPE_CHECKING:
import sys
from datetime import datetime

from backoff.types import Details

from singer_sdk._singerlib import Schema
from singer_sdk.helpers.types import Context
from singer_sdk.helpers.types import Auth, Context
from singer_sdk.tap_base import Tap

if sys.version_info >= (3, 10):
from typing import TypeAlias # noqa: ICN003
else:
from typing_extensions import TypeAlias

DEFAULT_PAGE_SIZE = 1000
DEFAULT_REQUEST_TIMEOUT = 300 # 5 minutes

_TToken = t.TypeVar("_TToken")
_Auth: TypeAlias = t.Callable[[requests.PreparedRequest], requests.PreparedRequest]


class RESTStream(Stream, t.Generic[_TToken], metaclass=abc.ABCMeta): # noqa: PLR0904
Expand Down Expand Up @@ -610,7 +603,7 @@ def parse_response(self, response: requests.Response) -> t.Iterable[dict]:
# Abstract methods:

@property
def authenticator(self) -> _Auth:
def authenticator(self) -> Auth:
"""Return or set the authenticator for managing HTTP auth headers.

If an authenticator is not specified, REST-based taps will simply pass
Expand Down
Loading