hii All can any one help me i am getting an error while defining incremental key if i defing in schema DateaTimetype so getting error File "/home/azure/testing/tap-azure_logs/.meltano/extractors/tap-azure_logs/venv/lib/python3.10/site-packages/singer_sdk/helpers/_state.py", line 238, in _greater_than_signpost new_value > signpost # type: ignore TypeError: '>' not supported between instances of 'int' and 'str' when i will define string or integer type it will not give me error but when use DateTimetype i am getting error

So what's the right underlying data type for your replication key?

import datetime import time import json import pendulum from pathlib import Path from azure.identity import ClientSecretCredential from azure.monitor.query import LogsQueryClient, LogsQueryStatus from datetime import datetime, timezone import pandas as pd from typing import Any, Dict, Optional, Union, List, Iterable, cast import singer from singer_sdk import typing as th # JSON Schema typing helpers import pytz from .utils import conversion from tap_azure_logs.client import azure_logsStream SCHEMAS_DIR = Path(file).parent / Path("./schemas") LOGGER = singer.get_logger() class LogStream(azure_logsStream): """Define custom stream.""" name = "LogStream" #primary_keys = ["id"] replication_method = "INCREMENTAL" replication_key = "state_endTime" # Optionally, you may also use

schema_filepath

in place of `schema`: # schema_filepath = SCHEMAS_DIR / "users.json" schema = th.PropertiesList( # th.Property("response", th.ObjectType()), th.Property("startTime", th.DateTimeType), th.Property("state_endTime", th.DateTimeType), th.Property("state_starttime", th.StringType), th.Property("state_currenttime", th.DateTimeType), th.Property("TimeGenerated", th.DateTimeType), # Adjust the data type as needed th.Property("Success", th.BooleanType), th.Property("Url", th.StringType), th.Property("Name", th.StringType), th.Property("ResultCode", th.IntegerType), th.Property("Id", th.StringType), th.Property("endTime_query",th.DateTimeType) ).to_dict() def get_starting_timestamp( self, context: Optional[dict] ) -> Optional[datetime]: value = self.get_starting_replication_key_value(context) if value is None: return None try: value = int(value) # dt = parser.parse(value) except ValueError as verr: value = pendulum.from_format(value, 'YYYY-MM-DD HHmmss.SSS') value = int(value.timestamp()*1000) pass except ValueError as verr: value = pendulum.from_format(value, 'YYYY-MM-DD HHmmss') value = int(value.timestamp()*1000) pass except Exception as ex: LOGGER.info("==========Exception occurred while converting to int==========") pass # if not self.is_timestamp_replication_key: # raise ValueError( # f"The replication key {self.replication_key} is not of timestamp type" # ) LOGGER.info("==========return value %s==========",value) return value def get_records(self, context: Optional[dict]) -> Iterable[dict]: """Return a generator of row-type dictionary objects.""" next_page = True start_date = self.get_starting_timestamp(context) LOGGER.info("initial_start_time %s",self.config.get("intial_start_time")) if start_date is None: if(self.config.get("intial_start_time")): startTime = self.config.get("intial_start_time") else: startTime = pendulum.now('UTC').subtract(hours=24) startTime = (startTime.int_timestamp*1000) else: startTime=start_date #startTime = pendulum.now('UTC') #end_date = self._get_end_date() #startTime=int((datetime.datetime.today() - datetime.timedelta(hours=56)).timestamp()), endTime = pendulum.now('UTC') endTime = (endTime.int_timestamp*1000) #endTime=int(datetime.datetime.now().timestamp()*1000) #changed by avinash on 24-Feb-2023, to remove repeat records endTime = int(endTime) - 300000 #changed by avinash on 22-Feb-2023, to remove repeat records startTime = int(startTime)+1000 self.state_starttime=startTime …

Well you shouldn't normally need to override

get_starting_timestamp

. So, what are the values of

new_value

and

signpost

when the comparison fails?