diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c35f0a8..f54e44a 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -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] diff --git a/generate_schema.py b/generate_schema.py index e49ac6d..7cda05d 100644 --- a/generate_schema.py +++ b/generate_schema.py @@ -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.""" @@ -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", ) diff --git a/pyproject.toml b/pyproject.toml index bccb458..8346e8c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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] @@ -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 diff --git a/tap_adp/authenticator.py b/tap_adp/authenticator.py index ebfdf13..f0a3251 100644 --- a/tap_adp/authenticator.py +++ b/tap_adp/authenticator.py @@ -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 @@ -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.""" @@ -46,7 +66,7 @@ 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", @@ -54,49 +74,56 @@ def oauth_request_body(self) -> dict: "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.") diff --git a/tap_adp/client.py b/tap_adp/client.py index 0eb79a9..59ebf5f 100644 --- a/tap_adp/client.py +++ b/tap_adp/client.py @@ -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 @@ -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 = "$[*]" @@ -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: @@ -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) @@ -103,7 +112,7 @@ 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, @@ -111,15 +120,15 @@ def get_url_params( 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. @@ -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 diff --git a/tap_adp/streams.py b/tap_adp/streams.py index 4d244f4..0919e49 100644 --- a/tap_adp/streams.py +++ b/tap_adp/streams.py @@ -5,9 +5,9 @@ from __future__ import annotations import sys -import typing as t from datetime import datetime, timedelta from http import HTTPStatus +from typing import TYPE_CHECKING, Any import requests @@ -18,7 +18,9 @@ else: from typing_extensions import override -if t.TYPE_CHECKING: +if TYPE_CHECKING: + from collections.abc import Iterable + from singer_sdk.helpers.types import Context, Record @@ -36,13 +38,13 @@ class WorkersStream(PaginatedADPStream): @override @property - def http_headers(self) -> dict: + def http_headers(self) -> dict[str, str]: headers = super().http_headers headers["Accept"] = "application/json;masked=false" return headers @override - def get_child_context(self, record: Record, context: Context | None) -> dict: + def get_child_context(self, record: Record, context: Context | None) -> Context | None: return {"_sdc_worker_aoid": record["associateOID"]} @@ -141,7 +143,7 @@ class USTaxProfileStream(ADPStream): parent_stream_type = WorkersStream @override - def parse_response(self, response: requests.Response) -> t.Iterable[dict]: + def parse_response(self, response: requests.Response) -> Iterable[dict[str, Any]]: if response.status_code == HTTPStatus.NOT_FOUND: return iter([]) return super().parse_response(response) @@ -184,7 +186,7 @@ class JobRequisitionStream(PaginatedADPStream): records_jsonpath = "$.jobRequisitions[*]" @override - def get_child_context(self, record: Record, context: Context | None) -> dict: + def get_child_context(self, record: Record, context: Context | None) -> Context | None: return {"_sdc_requisition_id": record["itemID"]} @@ -207,9 +209,7 @@ class QuestionnaireStream(ADPStream): """ name = "questionnaire" - path = ( - "/staffing/v3/work-fulfillment/recruiting-questionnaires/{_sdc_requisition_id}" - ) + path = "/staffing/v3/work-fulfillment/recruiting-questionnaires/{_sdc_requisition_id}" primary_keys = ("questionnaireID",) records_jsonpath = "$" parent_stream_type = JobRequisitionStream @@ -253,15 +253,15 @@ class PayrollOutputStream(ADPStream): records_jsonpath = "$.payrollOutputs[*]" # There's a root level processMessages key that has metaData about the corresponding payroll(s) might be useful, ignoring for now to move forward quickly # noqa: E501 @override - def get_child_context(self, record: Record, context: Context | None) -> dict: + def get_child_context(self, record: Record, context: Context | None) -> Context | None: return {"_sdc_payroll_item_id": record["itemID"]} @override def get_url_params( self, context: Context | None, - next_page_token: t.Any | None, - ) -> dict[str, t.Any] | str: + next_page_token: Any | None, + ) -> dict[str, Any] | str: params = {} # Date 30 days ago if date := self.get_starting_timestamp(context): @@ -273,7 +273,7 @@ def get_url_params( @override def post_process( self, - record: Record, + row: Record, context: Context | None = None, ) -> Record | None: # We subtract 30 days as recent payrolls are not available to pull @@ -281,14 +281,14 @@ def post_process( # payrolls that havne't been completed yet so we want to play it safe and try # to get them all. # This gives us a good chance of pulling all the most recent payrolls - record["_sdc_modified_schedule_entry_id"] = ( + row["_sdc_modified_schedule_entry_id"] = ( datetime.strptime( # noqa: DTZ007 - record["payrollScheduleReference"]["scheduleEntryID"][:8], + row["payrollScheduleReference"]["scheduleEntryID"][:8], "%Y%m%d", ) - timedelta(days=30) ) - return record + return row class PayrollOutputAccStream(ADPStream): @@ -304,8 +304,8 @@ class PayrollOutputAccStream(ADPStream): def get_url_params( self, context: Context | None, - next_page_token: t.Any | None, - ) -> dict[str, t.Any] | str: + next_page_token: Any | None, + ) -> dict[str, Any] | str: # Today's date assert context is not None # noqa: S101 @@ -321,13 +321,9 @@ def validate_response(self, response: requests.Response) -> None: if response.status_code == HTTPStatus.NOT_FOUND: response_json = response.json() if response_json.get("confirmMessage", {}).get("processMessages"): - process_messages = response_json.get("confirmMessage", {}).get( - "processMessages" - ) + process_messages = response_json.get("confirmMessage", {}).get("processMessages") for process_message in process_messages: - dev_message = process_message.get("developerMessage", {}).get( - "messageTxt", "" - ) + dev_message = process_message.get("developerMessage", {}).get("messageTxt", "") code_value = process_message.get( "developerMessage", {} ).get( @@ -343,9 +339,7 @@ def validate_response(self, response: requests.Response) -> None: if response.status_code == HTTPStatus.BAD_REQUEST and response.json().get( "confirmMessage", {} ).get("processMessages"): - process_messages = ( - response.json().get("confirmMessage", {}).get("processMessages") - ) + process_messages = response.json().get("confirmMessage", {}).get("processMessages") for process_message in process_messages: dev_message = process_message["developerMessage"]["messageTxt"] code_value = process_message["developerMessage"]["codeValue"] @@ -356,14 +350,16 @@ def validate_response(self, response: requests.Response) -> None: if ( code_value == "PAYGEN00030" ): # The payroll job id provided was in an invalid state (EDL, DAT, PVE, NER, EER, etc). # noqa: E501 - exception_message = f"The payroll job id provided was in an invalid state ({dev_message})." # noqa: E501 + exception_message = ( + f"The payroll job id provided was in an invalid state ({dev_message})." + ) self.logger.warning(exception_message) raise SkippableAPIError(exception_message) # Default handling if this isn't hit super().validate_response(response) @override - def get_records(self, context: Context | None) -> t.Iterable[Record]: + def get_records(self, context: Context | None) -> Iterable[Record]: """Return a generator of record-type dictionary objects. Each record emitted should be a dictionary of property names to their values. diff --git a/tap_adp/tap.py b/tap_adp/tap.py index 812f323..bee2e5e 100644 --- a/tap_adp/tap.py +++ b/tap_adp/tap.py @@ -2,11 +2,16 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from singer_sdk import Tap from singer_sdk import typing as th # JSON schema typing helpers from tap_adp import streams +if TYPE_CHECKING: + from tap_adp.client import ADPStream + class TapADP(Tap): """ADP tap class.""" @@ -55,7 +60,7 @@ class TapADP(Tap): ), ).to_dict() - def discover_streams(self) -> list[streams.ADPStream]: + def discover_streams(self) -> list[ADPStream]: """Return a list of discovered streams. Returns: diff --git a/uv.lock b/uv.lock index e58a51e..0557682 100644 --- a/uv.lock +++ b/uv.lock @@ -1549,7 +1549,7 @@ source = { editable = "." } dependencies = [ { name = "requests" }, { name = "singer-sdk", extra = ["faker"] }, - { name = "typing-extensions", marker = "python_full_version < '3.12'" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] [package.optional-dependencies] @@ -1572,7 +1572,7 @@ requires-dist = [ { name = "requests", specifier = "~=2.33.0" }, { name = "s3fs", marker = "extra == 's3'", specifier = "~=2026.3.0" }, { name = "singer-sdk", extras = ["faker"], specifier = "~=0.53.2" }, - { name = "typing-extensions", marker = "python_full_version < '3.12'", specifier = ">=4.15.0" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'", specifier = ">=4.15.0" }, ] provides-extras = ["s3"]