-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy path__init__.py
50 lines (35 loc) · 1.76 KB
/
__init__.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
import json
from typing import TypeVar, Callable, Optional, List, Any, Dict
from ..file import Logger as TextLoger
from .message import Message
from .format import Format
from .writer import Writer
from demure_logger import MessageDataType
from demure_logger.log import Level
MessageType = TypeVar( 'MessageType', bound=Message )
FormatType = TypeVar( 'FormatType' , bound=Format )
WriterType = TypeVar( 'WriterType' , bound=Writer )
class Logger( TextLoger ):
def __init__( self,
writer : Optional[ Callable[ ..., WriterType ] | WriterType ] = None,
format : Callable[ ..., FormatType ] | FormatType = Format( ),
message_class: Callable[ ..., MessageType ] | MessageType = Message,
*args,
**kwargs
):
super( ).__init__( writer, format, message_class, *args, **kwargs )
def _write( self, event: Level, *messages: List[MessageDataType] ) -> List[Message]:
result = [ ]
for message in messages:
message = self.message_class( event=event, **message ) if isinstance( message, dict ) else self.message_class( event=event, message=message )
if event <= self.level:
self.event_hanler( event, message )
def read_and_write( writer ):
state = writer.read( True )
state = { } if state is None or len( state ) == 0 else json.loads( state )
formated = self.format.prepare( message, state )
writer.truncate( True )
writer.write( formated, True )
self.writer.safe( read_and_write )
result.append( message )
return result