meltano invoke tap-mytap

th.Property(
            "private_key",
            th.StringType,
            default=None,
            description="Private_key",
        ),

authenticator

class JobDivaAuthenticator(OAuthJWTAuthenticator):
    """Authenticator class for JobDiva."""

    @classmethod
    def create_for_stream(
        cls,
        stream,  # noqa: ANN001
    ) -> JobDivaAuthenticator:
        """Instantiate an authenticator for a specific Singer stream.

        Args:
            stream: The Singer stream instance.

        Returns:
            A new authenticator.
        """
        return cls(
            stream=stream,
            auth_endpoint=JOBDIVA_AUTH_ENDPOINT,
            oauth_scopes=JOBDIVA_AUTH_SCOPES,
        )

OAuthJWTAuthenticator

private_key

JobDivaAuthenticator

ouath_request_body

OAuthJWTAuthenticator

OAuthAuthenticator

username

password

class JobDivaAuthenticator(OAuthAuthenticator):
    """Authenticator class for JobDiva."""

    @property
            def oauth_request_body(self) -> dict:
                return {
                    "grant_type": "password",
                    "scope": "<https://api.powerbi.com>",
                    "resource": "<https://analysis.windows.net/powerbi/api>",
                    "client_id": self.config["client_id"],
                    "username": self.config.get("username", self.config["client_id"]),
                    "password": self.config["password"],
                }

    @classmethod
...
...

ERROR    | tap-jobdiva          | Config validation error: 'auth_token' is a required property

tap.py

required=True

JobDivaAuthenticator

auth_endpoint

{'oneOf': [{'type': ['number']}, {'type': ['string']}, {'type': ['boolean']}]}

{'oneOf': [{'type': ['string']}, {'type': 'array', 'items': {'type': ['string']}}]}

WARNING:Could not append type because the JSON schema for the dictionary `{'oneOf': [{'type': ['number']}, {'type': ['string']}, {'type': ['boolean']}]}` appears to be invalid.
WARNING:Could not append type because the JSON schema for the dictionary `{'oneOf': [{'type': ['string']}, {'type': 'array', 'items': {'type': ['string']}}]}` appears to be invalid.

tap-jobdiva\venv\lib\site-packages\requests\models.py", line 1021, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 405 Client Error:  for url: <https://api.jobdiva.com/apiv2/authenticate>

tap-jobdiva\venv\lib\site-packages\simplejson\decoder.py", line 416, in raw_decode
    return self.scan_once(s, idx=_w(s, idx).end())
simplejson.errors.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

405

Method Not Allowed

POST

GET

OAuthAuthenticator

update_access_token

GET

update_access_token

oauth_request_body

@classmethod

meltano config tap-jobdiva list

TAP_JOBDIVA_USERNAME

TAP_JOBDIVA_PASSWORD

TAP_JOBDIVA_CLIENT_ID

.env

TAP_JOBDIVA_USERNAME=<username>
TAP_JOBDIVA_PASSWORD=<password>
TAP_JOBDIVA_CLIENT_ID=<client id>

username

password

client_id

@property
    def oauth_request_body(self) -> dict:
        return {
            "grant_type": "password",
            "scope": JOBDIVA_AUTH_SCOPES,
            "resource": JOBDIVA_AUTH_ENDPOINT,
        }

    def update_access_token(self) -> None:
        """Update `access_token` along with: `last_refreshed` and `expires_in`.

        Raises:
            RuntimeError: When OAuth login fails.
        """
        request_time = utc_now()
        auth_request_payload = self.oauth_request_payload
        token_response = requests.get(
            self.auth_endpoint,
            headers=self._oauth_headers,
            params={
                "clientid": config["client_id"],
                "username": config["username"],
                "password": config["password"],
            },
            data=auth_request_payload,
            timeout=60,
        )
        try:
            token_response.raise_for_status()
        except requests.HTTPError as ex:
            msg = f"Failed OAuth login, response was '{token_response.json()}'. {ex}"
            raise RuntimeError(msg) from ex

        <http://self.logger.info|self.logger.info>("OAuth authorization attempt was successful.")

        token_json = token_response.json()
        self.access_token = token_json["access_token"]
        expiration = token_json.get("expires_in", self._default_expiration)
        self.expires_in = int(expiration) if expiration else None
        if self.expires_in is None:
            self.logger.debug(
                "No expires_in received in OAuth response and no "
                "default_expiration set. Token will be treated as if it never "
                "expires.",
            )
        self.last_refreshed = request_time

tap-jobdiva\venv\lib\site-packages\simplejson\decoder.py", line 416, in raw_decode
    return self.scan_once(s, idx=_w(s, idx).end())
simplejson.errors.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

token_json = token_response.json()
        self.access_token = token_json["access_token"]
        expiration = token_json.get("expires_in", self._default_expiration)
        self.expires_in = int(expiration) if expiration else None
        if self.expires_in is None:
            self.logger.debug(
                "No expires_in received in OAuth response and no "
                "default_expiration set. Token will be treated as if it never "
                "expires.",
            )

{
  "message": "Date range more than 14 days, metric name = NewUpdatedCandidateRecords",
  "data": {}
}

response.text

response.json()

streams\rest.py", line 187, in validate_response
    raise FatalAPIError(msg)
singer_sdk.exceptions.FatalAPIError: 404 Client Error:  for path: /apiv2/<https://api.jobdiva.com/apiv2/bi/NewUpdatedCandidateRecordsStream>

class NewUpdatedCandidateRecordsStream(JobDivaStream):
    """Define custom stream."""

    name = "NewUpdatedCandidateRecordsStream"
    path = "bi/NewUpdatedCandidateRecords"

class JobDivaStream(RESTStream):
    """JobDiva stream class."""

    @property
    def url_base(self) -> str:
        """Return the API URL root, configurable via tap settings."""
        # TODO: hardcode a value here, or retrieve it from self.config
        return "<https://api.jobdiva.com/apiv2/>"

self.access_token

update_access_token

token_json = token_response.text
self.access_token = token_json
expiration = 86400

/apiv2/bi/NewUpdatedCandidateRecords

GET

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

        Returns:
            An authenticator instance.
        """
        return JobDivaAuthenticator.create_for_stream(self)

curl

limit

offset