cengal.parallel_execution.coroutines.coro_tools.low_latency.json.versions.v_0.ajson

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

View Source

 1#!/usr/bin/env python
 2# coding=utf-8
 3
 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
 5# 
 6# Licensed under the Apache License, Version 2.0 (the "License");
 7# you may not use this file except in compliance with the License.
 8# You may obtain a copy of the License at
 9# 
10#     http://www.apache.org/licenses/LICENSE-2.0
11# 
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17
18
19"""
20Module Docstring
21Docstrings: http://www.python.org/dev/peps/pep-0257/
22"""
23
24
25__author__ = "ButenkoMS <gtalk@butenkoms.space>"
26__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
27__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
28__license__ = "Apache License, Version 2.0"
29__version__ = "4.4.1"
30__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
31__email__ = "gtalk@butenkoms.space"
32# __status__ = "Prototype"
33__status__ = "Development"
34# __status__ = "Production"
35
36
37__all__ = ['adump', 'adumps', 'aload', 'aloads']
38
39from cengal.parallel_execution.coroutines.coro_scheduler import CoroScheduler, CoroSchedulerType, Interface
40from .json import *
41from cengal.parallel_execution.coroutines.coro_tools.await_coro import await_coro
42
43
44async def adump(cs: CoroSchedulerType, *args, **kwargs):
45    # You may use
46    # ly = None, priority: CoroPriority = CoroPriority.normal
47    # as an additional parameters in order to configure LowLatencyJSONEncoder
48    
49    def coro(i: Interface):
50        kwargs['cls'] = LowLatencyJSONEncoder
51        return original_json_dump(*args, **kwargs)
52    
53    return await await_coro(cs, coro)
54
55
56async def adumps(cs: CoroSchedulerType, *args, **kwargs):
57    # You may use
58    # ly = None, priority: CoroPriority = CoroPriority.normal
59    # as an additional parameters in order to configure LowLatencyJSONEncoder
60    
61    def coro(i: Interface):
62        kwargs['cls'] = LowLatencyJSONEncoder
63        return original_json_dumps(*args, **kwargs)
64    
65    return await await_coro(cs, coro)
66
67
68async def aload(cs: CoroSchedulerType, *args, **kwargs):
69    # You may use
70    # ly = None, priority: CoroPriority = CoroPriority.normal
71    # as an additional parameters in order to configure LowLatencyJSONEncoder
72    
73    def coro(i: Interface):
74        kwargs['cls'] = LowLatencyJSONDecoder
75        kwargs['object_pairs_hook'] = LowLatencyObjectPairsHook(**kwargs)
76        return original_json_load(*args, **kwargs)
77    
78    return await await_coro(cs, coro)
79
80
81async def aloads(cs: CoroSchedulerType, *args, **kwargs):
82    # You may use
83    # ly = None, priority: CoroPriority = CoroPriority.normal
84    # as an additional parameters in order to configure LowLatencyJSONEncoder
85    
86    def coro(i: Interface):
87        kwargs['cls'] = LowLatencyJSONDecoder
88        kwargs['object_pairs_hook'] = LowLatencyObjectPairsHook(**kwargs)
89        return original_json_loads(*args, **kwargs)
90    
91    return await await_coro(cs, coro)

async def adump( cs: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable], *args, **kwargs): View Source

45async def adump(cs: CoroSchedulerType, *args, **kwargs):
46    # You may use
47    # ly = None, priority: CoroPriority = CoroPriority.normal
48    # as an additional parameters in order to configure LowLatencyJSONEncoder
49    
50    def coro(i: Interface):
51        kwargs['cls'] = LowLatencyJSONEncoder
52        return original_json_dump(*args, **kwargs)
53    
54    return await await_coro(cs, coro)

async def adumps( cs: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable], *args, **kwargs): View Source

57async def adumps(cs: CoroSchedulerType, *args, **kwargs):
58    # You may use
59    # ly = None, priority: CoroPriority = CoroPriority.normal
60    # as an additional parameters in order to configure LowLatencyJSONEncoder
61    
62    def coro(i: Interface):
63        kwargs['cls'] = LowLatencyJSONEncoder
64        return original_json_dumps(*args, **kwargs)
65    
66    return await await_coro(cs, coro)

async def aload( cs: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable], *args, **kwargs): View Source

69async def aload(cs: CoroSchedulerType, *args, **kwargs):
70    # You may use
71    # ly = None, priority: CoroPriority = CoroPriority.normal
72    # as an additional parameters in order to configure LowLatencyJSONEncoder
73    
74    def coro(i: Interface):
75        kwargs['cls'] = LowLatencyJSONDecoder
76        kwargs['object_pairs_hook'] = LowLatencyObjectPairsHook(**kwargs)
77        return original_json_load(*args, **kwargs)
78    
79    return await await_coro(cs, coro)

async def aloads( cs: typing.Union[cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerGreenlet, cengal.parallel_execution.coroutines.coro_scheduler.versions.v_0.coro_scheduler.CoroSchedulerAwaitable], *args, **kwargs): View Source

82async def aloads(cs: CoroSchedulerType, *args, **kwargs):
83    # You may use
84    # ly = None, priority: CoroPriority = CoroPriority.normal
85    # as an additional parameters in order to configure LowLatencyJSONEncoder
86    
87    def coro(i: Interface):
88        kwargs['cls'] = LowLatencyJSONDecoder
89        kwargs['object_pairs_hook'] = LowLatencyObjectPairsHook(**kwargs)
90        return original_json_loads(*args, **kwargs)
91    
92    return await await_coro(cs, coro)