feat: New Reporting streams for Klaviyo

[mpabraham]

No description provided.

[edgarrmondragon]

This isn't a complete review, but hopefully it leads to a refactor that results in significantly less code to review.

[edgarrmondragon]

Suggested change

	def _get_report_config_value(config: Mapping[str, Any], key: str) -> dict[str, Any]:
	value = config.get(key, {})
	if value is None:
	return {}
	if isinstance(value, dict):
	return value
	if isinstance(value, str):
	parsed_value = json.loads(value)
	if isinstance(parsed_value, dict):
	return parsed_value
	msg = f"Expected '{key}' JSON to decode to an object."
	raise TypeError(msg)
	msg = f"Expected '{key}' to be an object or JSON string."
	raise TypeError(msg)

this doesn't seem to be used anywhere

[edgarrmondragon]

Suggested change

	def __init__(
	self,
	tap: Tap,
	*,
	report_config: dict[str, Any] | None = None,
	report_name: str | None = None,
	) -> None:
	"""Initialize the stream with an optional report config and name."""
	self._report_config = report_config
	super().__init__(tap=tap, name=report_name or self.name)
	def __init__(
	self,
	tap: Tap,
	*,
	report_config: dict[str, Any],
	report_name: str,
	) -> None:
	"""Initialize the stream with an optional report config and name."""
	self._report_config = report_config
	super().__init__(tap=tap, name=report_name)

This applies to all the new streams that follow this style.

[edgarrmondragon]

see above

Suggested change

	config = self._report_config or {}
	config = self._report_config

[edgarrmondragon]

I'm curious what the benefits of supporting a JSON string form of the array are. Both Meltano and the SDK will parse the JSON string if it's coming from the environment.

With that said, could we make this a simpler type? That way we can get rid of a lot of the validation code needed for the case when a string is passed. For example:

Suggested change

	def _named_interval_report_config_type() -> th.AnyOf:
	report_object = th.ObjectType(
	th.Property("name", th.StringType),
	th.Property("statistics", th.ArrayType(th.StringType)),
	th.Property("interval", th.StringType),
	th.Property(
	"timeframe",
	th.ObjectType(
	th.Property("key", th.StringType),
	),
	),
	)
	return th.AnyOf(th.ArrayType(report_object), th.StringType, th.NullType)
	def _named_interval_report_config_type() -> th.ArrayType:
	report_object = th.ObjectType(
	th.Property("name", th.StringType, required=True),
	th.Property("statistics", th.ArrayType(th.StringType)),
	th.Property("interval", th.StringType),
	th.Property(
	"timeframe",
	th.ObjectType(
	th.Property("key", th.StringType),
	),
	),
	)
	return th.ArrayType(report_object, nullable=True)

and ditto for the other config schemas.

[edgarrmondragon]

As long as the stream name matches the file name without the extension, you don't need this. See the older streams.

Suggested change

_schema_path = resources.files(schemas).joinpath("segment_series_report.json")

[edgarrmondragon]

See above

Suggested change

	@property
	def schema(self) -> dict[str, Any]: # type: ignore[override]
	"""Return the shared schema for all named segment series streams."""
	return cast("dict[str, Any]", json.loads(self._schema_path.read_text(encoding="utf-8")))