<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , unicode_literals , print_function <EOL> import re <EOL> import datetime <EOL> from contextlib import closing <EOL> from jmx4py . util import json , auxiliary <EOL> from jmx4py . jolokia . errors import JmxResponseError <EOL> from jmx4py . jolokia . connection import JmxConnection <EOL> _QUOTE_RE = re . compile ( r"<STR_LIT>" ) <EOL> _UNQUOTE_RE = re . compile ( r"<STR_LIT>" ) <EOL> def quote ( path ) : <EOL> """<STR_LIT>""" <EOL> return _QUOTE_RE . sub ( r"<STR_LIT>" , path ) if path else path <EOL> def unquote ( path ) : <EOL> """<STR_LIT>""" <EOL> if not path : <EOL> return path <EOL> if path . endswith ( '<STR_LIT:!>' ) and ( len ( path ) - len ( path . rstrip ( '<STR_LIT:!>' ) ) ) % <NUM_LIT:2> : <EOL> raise ValueError ( "<STR_LIT>" % path ) <EOL> return _UNQUOTE_RE . sub ( r"<STR_LIT>" , path ) <EOL> class JmxResponse ( dict ) : <EOL> """<STR_LIT>""" <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return json . dumps ( self , indent = <NUM_LIT:2> ) <EOL> def __repr__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> self . __class__ . __name__ , self [ "<STR_LIT>" ] [ "<STR_LIT:type>" ] , self [ "<STR_LIT:status>" ] , <EOL> datetime . datetime . fromtimestamp ( self [ "<STR_LIT>" ] ) . isoformat ( ) , <EOL> auxiliary . digest_repr ( self [ "<STR_LIT:value>" ] ) , <EOL> ) <EOL> def __getitem__ ( self , key ) : <EOL> try : <EOL> return super ( JmxResponse , self ) . __getitem__ ( key ) <EOL> except KeyError : <EOL> if '<STR_LIT:.>' in key : <EOL> namespace = self [ "<STR_LIT:value>" ] <EOL> for part in key . split ( '<STR_LIT:.>' ) : <EOL> namespace = namespace . get ( part , None ) <EOL> if not namespace : <EOL> break <EOL> return namespace <EOL> elif key in self [ "<STR_LIT:value>" ] : <EOL> return self [ "<STR_LIT:value>" ] [ key ] <EOL> raise <EOL> def __getattr__ ( self , name ) : <EOL> try : <EOL> return self [ name ] <EOL> except KeyError : <EOL> return getattr ( super ( JmxResponse , self ) , name ) <EOL> class JmxClientConfig ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . url = url <EOL> self . user = kw . get ( "<STR_LIT:user>" , None ) <EOL> self . password = kw . get ( "<STR_LIT:password>" , None ) <EOL> self . method = kw . get ( "<STR_LIT>" , "<STR_LIT:POST>" ) <EOL> class JmxClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , ** kw ) : <EOL> """<STR_LIT>""" <EOL> self . cfg = JmxClientConfig ( url , ** kw ) <EOL> self . connection = JmxConnection . from_url ( self . cfg . url ) <EOL> def __repr__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( self . __class__ . __name__ , self . connection . url ) <EOL> def _execute ( self , ** kw ) : <EOL> """<STR_LIT>""" <EOL> req = kw <EOL> with closing ( self . connection . open ( ) ) as handle : <EOL> resp = JmxResponse ( handle . send ( req ) ) <EOL> if resp . status != <NUM_LIT:200> : <EOL> raise JmxResponseError ( resp , req ) <EOL> return resp <EOL> def read ( self , mbean , attribute = None , path = None , ** kw ) : <EOL> """<STR_LIT>""" <EOL> req = dict ( type = "<STR_LIT>" , mbean = mbean ) <EOL> if attribute : <EOL> req [ "<STR_LIT>" ] = attribute <EOL> if path : <EOL> req [ "<STR_LIT:path>" ] = quote ( path ) <EOL> if kw : <EOL> req . update ( kw ) <EOL> resp = self . _execute ( ** req ) <EOL> return resp <EOL> def write ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = self . _execute ( type = "<STR_LIT>" ) <EOL> return resp <EOL> def invoke ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = self . _execute ( type = "<STR_LIT>" ) <EOL> return resp <EOL> def search ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = self . _execute ( type = "<STR_LIT>" ) <EOL> return resp <EOL> def version ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = self . _execute ( type = "<STR_LIT:version>" ) <EOL> resp [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] = tuple ( int ( i ) for i in resp . protocol . split ( '<STR_LIT:.>' ) ) <EOL> resp [ "<STR_LIT:value>" ] [ "<STR_LIT>" ] = tuple ( int ( i ) for i in resp . agent . split ( '<STR_LIT:.>' ) ) <EOL> return resp </s>
