Skip to content
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
5 changes: 3 additions & 2 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,14 @@ repos:
- id: trailing-whitespace

- repo: https://github.com/python-jsonschema/check-jsonschema
rev: 0.36.0
rev: 0.37.1
hooks:
- id: check-dependabot
- id: check-github-workflows
- id: check-meltano

- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.14.13
rev: v0.15.8
hooks:
- id: ruff-check
args: [--fix, --exit-non-zero-on-fix, --show-fixes]
Expand Down
6 changes: 4 additions & 2 deletions generate_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ def make_nullable(schema: dict) -> dict:

def generate_schema(
stream_instance: Stream,
context: Context,
context: Context | None,
output_file: str,
) -> None:
"""Generate a schema for a given stream."""
Expand Down Expand Up @@ -89,7 +89,9 @@ def main() -> None:

for child_name, child_stream in children.items():
generate_schema(
child_stream, context=None, output_file=f"tap_adp/schemas/{child_name}.json"
child_stream,
context=None,
output_file=f"tap_adp/schemas/{child_name}.json",
)


Expand Down
11 changes: 10 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ classifiers = [
dependencies = [
"singer-sdk[faker]~=0.53.2",
"requests~=2.33.0",
"typing-extensions>=4.15.0 ; python_full_version < '3.12'",
"typing-extensions>=4.15.0 ; python_full_version < '3.13'",
]

[project.optional-dependencies]
Expand All @@ -47,11 +47,20 @@ build-backend = "hatchling.build"

[tool.pytest]
addopts = [
"-v",
"-ra",
"--durations=10",
]
filterwarnings = [ "error" ]
log_level = "INFO"
minversion = "9"
strict = true
testpaths = [ "tests" ]

[tool.mypy]
enable_error_code = ["ignore-without-code", "redundant-expr", "truthy-bool"]
strict = true
warn_unreachable = true
warn_unused_configs = true
warn_unused_ignores = true

Expand Down
103 changes: 65 additions & 38 deletions tap_adp/authenticator.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,13 @@
from __future__ import annotations

import os
import ssl
import sys
import tempfile
from typing import Any

import requests
from requests.adapters import HTTPAdapter
from singer_sdk.authenticators import OAuthAuthenticator
from singer_sdk.helpers._util import utc_now

Expand All @@ -20,6 +22,24 @@
AUTH_ENDPOINT = "https://accounts.adp.com/auth/oauth/v2/token"


class _MTLSAdapter(HTTPAdapter):
"""Requests adapter that injects a pre-built SSL context for mTLS.

Works around SSL context caching in requests >=2.32.5 (psf/requests#6767)
that ignores the ``cert=`` parameter when a cached context already exists
for the target host, causing mTLS authentication to silently drop the
client certificate.
"""

def __init__(self, ssl_context: ssl.SSLContext, **kwargs: Any) -> None:
self._ssl_context = ssl_context
super().__init__(**kwargs)

def init_poolmanager(self, *args: Any, **kwargs: Any) -> None:
kwargs["ssl_context"] = self._ssl_context
super().init_poolmanager(*args, **kwargs) # type: ignore[no-untyped-call]


class ADPAuthenticator(OAuthAuthenticator):
"""Authenticator class for ADP."""

Expand All @@ -46,57 +66,64 @@ def __init__(

@override
@property
def oauth_request_body(self) -> dict:
def oauth_request_body(self) -> dict[str, Any]:
"""Define the OAuth request body for ADP."""
return {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
}

def _build_ssl_context(self) -> ssl.SSLContext:
"""Build an SSL context with the client certificate pre-loaded.

Writes the PEM strings to temporary files, loads them into an
``ssl.SSLContext``, then deletes the files before returning. The
context holds the cert in memory so the files are not needed at
request time.
"""
with (
tempfile.NamedTemporaryFile(mode="wb", delete=False, suffix=".pem") as cert_file,
tempfile.NamedTemporaryFile(mode="wb", delete=False, suffix=".pem") as key_file,
):
cert_path = cert_file.name
key_path = key_file.name
cert_file.write(self.cert_public.encode("utf-8"))
key_file.write(self.cert_private.encode("utf-8"))

try:
os.chmod(cert_path, 0o600) # noqa: PTH101
os.chmod(key_path, 0o600) # noqa: PTH101
ctx = ssl.create_default_context()
ctx.load_cert_chain(certfile=cert_path, keyfile=key_path)
finally:
os.unlink(cert_path) # noqa: PTH108
os.unlink(key_path) # noqa: PTH108

return ctx

@override
def update_access_token(self) -> None:
"""Update `access_token` along with `last_refreshed` and `expires_in`."""
request_time = utc_now()

# Create temporary files for the cert and key
with (
tempfile.NamedTemporaryFile(mode="wb+", delete=False) as cert_file,
tempfile.NamedTemporaryFile(mode="wb+", delete=False) as key_file,
):
# Write contents to the temporary files
cert_file.write(self.cert_public.encode("utf-8"))
cert_file.flush()
session = requests.Session()
session.mount("https://", _MTLSAdapter(ssl_context=self._build_ssl_context()))

key_file.write(self.cert_private.encode("utf-8"))
key_file.flush()

# Ensure the files are readable only by the owner (optional)
os.chmod(cert_file.name, 0o600) # noqa: PTH101
os.chmod(key_file.name, 0o600) # noqa: PTH101

# Make the OAuth request
try:
response = requests.post(
self.auth_endpoint,
data=self.oauth_request_body,
headers=self._oauth_headers,
timeout=60,
cert=(cert_file.name, key_file.name),
)
response.raise_for_status()
except requests.HTTPError:
self.logger.warning(
"Failed OAuth login, response was '%s'",
response.text,
)
raise
finally:
# Clean up the temporary files
cert_file.close()
key_file.close()
os.unlink(cert_file.name) # noqa: PTH108
os.unlink(key_file.name) # noqa: PTH108
try:
response = session.post(
self.auth_endpoint,
data=self.oauth_request_body,
headers=self._oauth_headers,
timeout=60,
)
response.raise_for_status()
except requests.HTTPError:
self.logger.warning(
"Failed OAuth login, response was '%s'",
response.text,
)
raise

self.logger.info("OAuth authorization attempt was successful.")

Expand Down
31 changes: 20 additions & 11 deletions tap_adp/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,9 @@

import decimal
import sys
import typing as t
from functools import cached_property
from http import HTTPStatus
from typing import TYPE_CHECKING, Any, Generic, TypeVar

from singer_sdk import SchemaDirectory, StreamSchema
from singer_sdk.helpers._typing import TypeConformanceLevel
Expand All @@ -22,12 +22,21 @@
else:
from typing_extensions import override

if t.TYPE_CHECKING:
if sys.version_info >= (3, 13):
from typing import TypeVar
else:
from typing_extensions import TypeVar

if TYPE_CHECKING:
from collections.abc import Iterable

import requests
from singer_sdk.helpers.types import Context

_T = TypeVar("_T", default=Any)

class ADPStream(RESTStream):

class ADPStream(RESTStream[_T], Generic[_T]):
"""ADP stream class."""

records_jsonpath = "$[*]"
Expand Down Expand Up @@ -56,7 +65,7 @@ def authenticator(self) -> ADPAuthenticator:
)

@override
def parse_response(self, response: requests.Response) -> t.Iterable[dict]:
def parse_response(self, response: requests.Response) -> Iterable[dict[str, Any]]:
"""Parse the response and return an iterator of result records.

Args:
Expand Down Expand Up @@ -90,11 +99,11 @@ def response_error_message(self, response: requests.Response) -> str:
)


class PaginatedADPStream(ADPStream):
class PaginatedADPStream(ADPStream[int]):
"""Paginated ADP stream class."""

@override
def get_new_paginator(self) -> BaseAPIPaginator:
def get_new_paginator(self) -> ADPPaginator:
"""Create a new paginator for ADP API pagination."""
return ADPPaginator(start_value=0, page_size=100)

Expand All @@ -103,23 +112,23 @@ def get_url_params(
self,
context: Context | None,
next_page_token: int | None,
) -> dict[str, t.Any]:
) -> dict[str, Any]:
return {
"$top": 100, # Set the desired page size
"$skip": next_page_token or 0,
}


class ADPPaginator(BaseAPIPaginator[int]):
"""Paginator for ADP API that uses 'top' and 'skip' parameters and stops on 204 response.""" # noqa: E501
"""Paginator for ADP API that uses 'top' and 'skip' parameters and stops on 204 response."""

@override
def __init__(
self,
start_value: int,
page_size: int,
*args: t.Any,
**kwargs: t.Any,
*args: Any,
**kwargs: Any,
) -> None:
"""Initialize the paginator with a starting value and page size.

Expand Down Expand Up @@ -153,5 +162,5 @@ def has_more(self, response: requests.Response) -> bool:

Returns:
`True` if pagination should continue, `False` if a 204 No Content is received.
""" # noqa: E501
"""
return response.status_code != HTTPStatus.NO_CONTENT
Loading