refactor: Implement msgspec encoding

[edgarrmondragon]

• Docs
• UX. Relying on mro magic is not great, e.g. why SQLiteTap(MsgSpecWriter, SQLTap) and not SQLiteTap(SQLTap, MsgSpecWriter). A better approach might be to make the IO implementation an attribute of the Singer class.
• Fix tests

---

📚 Documentation preview 📚: https://meltano-sdk--2541.org.readthedocs.build/en/2541/

[codspeed-hq]

CodSpeed Performance Report

Merging #2541 will improve performances by ×12

_{Comparing edgarrmondragon/refactor/msgspec-impl-naive (9e46c44) with main (4ce312f)}

Summary

⚡ 2 improvements
✅ 4 untouched benchmarks

Benchmarks breakdown

	Benchmark	main	edgarrmondragon/refactor/msgspec-impl-naive	Change
⚡	test_bench_deserialize_json	24.1 ms	5.5 ms	×4.4
⚡	test_bench_format_message	53.3 ms	4.4 ms	×12

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 87.32%. Comparing base (4ce312f) to head (9e46c44).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2541      +/-   ##
==========================================
- Coverage   89.82%   87.32%   -2.51%     
==========================================
  Files          58       59       +1     
  Lines        4886     4923      +37     
  Branches      960      960              
==========================================
- Hits         4389     4299      -90     
- Misses        346      462     +116     
- Partials      151      162      +11     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[BuzzCutNorman]

I found that it helped to add a defualt_output to the for the SingerWriter to use. This allows you to make the write message a little generic.

default_output = sys.stdout.buffer

def write_message(self, message: Message) -> None:
	"""Write a message to stdout.

	Args:
		message: The message to write.
	"""
	self.default_output.write(self.format_message(message))
	self.default_output.flush()

[BuzzCutNorman]

In the json.py file I found to match the msgspec performance suggestions and fit into the framework you put in place I created a function for generating jsonl so I could keep the functionality of seralize_json to return strings. This way the serialize_json can be used in the connector engine creation and also in process_batch_files.

https://jcristharif.com/msgspec/perf-tips.html#line-delimited-json

json.py:

def serialize_json(obj: object, **kwargs: t.Any) -> str:
    """Serialize a dictionary into a line of json.

    Args:
        obj: A Python object usually a dict.
        **kwargs: Optional key word arguments.

    Returns:
        A string of serialized json.
    """
    return encoder.encode(obj).decode()

msg_buffer = bytearray(64)

def serialize_jsonl(obj: object, **kwargs: t.Any) -> bytes:
        """Serialize a dictionary into a line of jsonl.

        Args:
            obj: A Python object usually a dict.
            **kwargs: Optional key word arguments.

        Returns:
            A bytes of serialized json.
        """
        encoder.encode_into(obj, msg_buffer)
        msg_buffer.extend(b"\n")
        return msg_buffer

SingerWriter:

    def serialize_message(self, message: Message) -> str | bytes:
        """Serialize a dictionary into a line of json.

        Args:
            message: A Singer message object.

        Returns:
            A string of serialized json.
        """
        return serialize_jsonl(message.to_dict())

[edgarrmondragon]

I found that it helped to add a defualt_output to the for the SingerWriter to use. This allows you to make the write message a little generic.

default_output = sys.stdout.buffer

def write_message(self, message: Message) -> None:
	"""Write a message to stdout.

	Args:
		message: The message to write.
	"""
	self.default_output.write(self.format_message(message))
	self.default_output.flush()

Do you mean in

sdk/singer_sdk/_singerlib/encoding/_msgspec.py

Lines 73 to 94 in 08b58bf

	class MsgSpecWriter(GenericSingerWriter[bytes, Message]):
	"""Interface for all plugins writing Singer messages to stdout."""
	def serialize_message(self, message: Message) -> bytes: # noqa: PLR6301
	"""Serialize a dictionary into a line of json.
	Args:
	message: A Singer message object.
	Returns:
	A string of serialized json.
	"""
	return encoder.encode(message.to_dict())
	def write_message(self, message: Message) -> None:
	"""Write a message to stdout.
	Args:
	message: The message to write.
	"""
	sys.stdout.buffer.write(self.format_message(message) + b"\n")
	sys.stdout.flush()

?

[BuzzCutNorman]

Yes, that is exactly what I meant. Could have definitely been stated clearer on my part😅.

 class MsgSpecWriter(GenericSingerWriter[bytes, Message]): 
     """Interface for all plugins writing Singer messages to stdout.""" 
     
     default_output = sys.stdout.buffer
     
     def serialize_message(self, message: Message) -> bytes:  # noqa: PLR6301 
         """Serialize a dictionary into a line of json. 
  
         Args: 
             message: A Singer message object. 
  
         Returns: 
             A string of serialized json. 
         """ 
         return serialize_jsonl(message.to_dict()) 
  
     def write_message(self, message: Message) -> None: 
         """Write a message to stdout. 
  
         Args: 
             message: The message to write. 
         """ 
 	self.default_output.write(self.format_message(message))
	self.default_output.flush()

[edgarrmondragon]

Naive of me to think I could get this across in 1/2 a day of work 😅. I'll come back to this later, there's plenty of time until the planned release date.

[BuzzCutNorman]

Like the pun 😊. Great dad joke material. Kind an inside joke now since you dropped (naive) from the title of the PR.