-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path__init__.py
More file actions
50 lines (35 loc) · 1.76 KB
/
Copy path__init__.py
File metadata and controls
50 lines (35 loc) · 1.76 KB
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