bug: support lag for non-UTC datetime cursor fields

[hairrrrr]

This PR:

• Add ensure_pendulum_datetime_non_utc to parse datetime strings into non-UTC datetime objects.
• Add _datetime_obj_to_str to preserve the colon in the timezone when converting datetime objects back to strings.
• Skip writing back state if no valid rows are found for last_value in the transformer, which may otherwise cause incorrect behavior.

Fix issure: #2169

Please let me know if there are any problems with my code. Thank you!

[netlify]

✅ Deploy Preview for dlt-hub-docs canceled.

Name	Link
🔨 Latest commit	e18fb40
🔍 Latest deploy log	https://app.netlify.com/sites/dlt-hub-docs/deploys/67691a4bbec71a00083b99fe

[rudolfix]

thanks for catching edge cases for the lag! the problem you've found is deeper than just empty results. please see my comments.

[rudolfix]

this is really a good catch but the problem is deeper than this. we cannot let the state go back and indeed when there's a lag this can unfortunately happen (like in your test, where the "last value" record is missing in the second read).

so IMO this code will fix it

        if self.lag and (cached_last_value := self._cached_state.get("last_value")):
            transformer.last_value = self.last_value_func(transformer.last_value, cached_last_value)
        self._cached_state["last_value"] = transformer.last_value

it uses last_value_func to make sure that "highest" last_value is preserved ("highest" per order established by the last_value_func)

rows is of TDataItems type: so it may be a single item or a list of items. items may be dicts, panda frames, arrow tables or batches.

so for example rows may be a pandas frame or a list of pandas frames. that's way the checks below are not sufficient. but as I said the problem is deeper (but solution is at the end simpler)

[hairrrrr]

Thanks, @rudolfix! I think this modification aligns well with the dlt code style. As you mentioned, there might be deeper issues that I’m not aware of at the moment.

so for example rows may be a pandas frame or a list of pandas frames. that's way the checks below are not sufficient. but as I said the problem is deeper (but solution is at the end simpler)

I have a question. If rows is an empty list of pandas DataFrames, I think the code rows != [] will handle this condition. This is because of the following code:

        if isinstance(rows, list):
            rows = [
                item
                for item in (self._transform_item(transformer, row) for row in rows)
                if item is not None
            ]
        else:
            rows = self._transform_item(transformer, rows)

[hairrrrr]

Hi, @rudolfix. I have resovled all the changes you ask! Please review again. Thank you!