65 lines
1.7 KiB
Python
65 lines
1.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
# bthlabs-jsonrpc-core | (c) 2022-present Tomek Wójcik | MIT License
|
|
from __future__ import annotations
|
|
|
|
import abc
|
|
import json
|
|
import typing
|
|
|
|
|
|
class Codec(abc.ABC):
|
|
"""Base class for codecs."""
|
|
|
|
# pragma mark - Abstract public interface
|
|
|
|
@abc.abstractmethod
|
|
def decode(self, payload: str | bytes, **decoder_kwargs) -> typing.Any:
|
|
"""
|
|
Decode the *payload*. *decoder_kwargs* are implementation specific.
|
|
|
|
Subclasses must implement this method.
|
|
"""
|
|
...
|
|
|
|
@abc.abstractmethod
|
|
def encode(self, payload: typing.Any, **encoder_kwargs) -> str:
|
|
"""
|
|
Encode the *payload*. *encoder_kwargs* are implementation specific.
|
|
|
|
Subclasses must implement this method.
|
|
"""
|
|
...
|
|
|
|
@abc.abstractmethod
|
|
def get_content_type(self) -> str:
|
|
"""
|
|
Return the MIME type for the encoded content.
|
|
|
|
Subclasses must implement this method.
|
|
"""
|
|
...
|
|
|
|
|
|
class JSONCodec(Codec):
|
|
"""JSON codec"""
|
|
|
|
# pragma mark - Public interface
|
|
|
|
def decode(self, payload: str | bytes, **decoder_kwargs) -> typing.Any:
|
|
"""
|
|
Decode *payload* using :py:func:`json.loads`. *decoder_kwargs* will
|
|
be passed verbatim to the decode function.
|
|
"""
|
|
return json.loads(payload, **decoder_kwargs)
|
|
|
|
def encode(self, payload: typing.Any, **encoder_kwargs) -> str:
|
|
"""
|
|
Encode *payload* using :py:func:`json.dumps`. *encoder_kwargs* will
|
|
be passed verbatim to the encode function.
|
|
"""
|
|
return json.dumps(payload, **encoder_kwargs)
|
|
|
|
def get_content_type(self):
|
|
"""Returns ``application/json``."""
|
|
return 'application/json'
|