<s> """<STR_LIT>""" <EOL> import requests <EOL> import yaml <EOL> import posixpath <EOL> class K8sConfig ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> self . filename = filename <EOL> self . doc = None <EOL> def parse ( self ) : <EOL> """<STR_LIT>""" <EOL> with open ( self . filename ) as f : <EOL> self . doc = yaml . safe_load ( f . read ( ) ) <EOL> if "<STR_LIT>" in self . doc and self . doc [ "<STR_LIT>" ] : <EOL> current_context = filter ( lambda x : <EOL> x [ '<STR_LIT:name>' ] == self . doc [ '<STR_LIT>' ] , <EOL> self . doc [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> username = current_context [ '<STR_LIT:user>' ] <EOL> clustername = current_context [ '<STR_LIT>' ] <EOL> self . user = filter ( lambda x : x [ '<STR_LIT:name>' ] == username , <EOL> self . doc [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] [ '<STR_LIT:user>' ] <EOL> self . cluster = filter ( lambda x : x [ '<STR_LIT:name>' ] == clustername , <EOL> self . doc [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> class K8sClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cfg_file , url = None , token = None , ca = None , <EOL> insecure_skip_tls_verify = False , version = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> self . url = url <EOL> self . token = token <EOL> self . insecure_skip_tls_verify = insecure_skip_tls_verify <EOL> self . ca = ca <EOL> if cfg_file : <EOL> self . config = K8sConfig ( cfg_file ) <EOL> self . config . parse ( ) <EOL> self . url = self . config . cluster [ "<STR_LIT>" ] <EOL> self . token = self . config . user [ "<STR_LIT>" ] <EOL> if "<STR_LIT>" in self . config . cluster : <EOL> self . ca = self . config . cluster [ "<STR_LIT>" ] <EOL> elif self . config . cluster . get ( '<STR_LIT>' ) : <EOL> self . insecure_skip_tls_verify = True <EOL> self . version = version <EOL> self . session = self . build_session ( ) <EOL> def build_session ( self ) : <EOL> """<STR_LIT>""" <EOL> s = requests . Session ( ) <EOL> if self . ca : <EOL> s . verify = self . ca <EOL> elif self . insecure_skip_tls_verify : <EOL> s . verify = False <EOL> s . headers [ "<STR_LIT>" ] = "<STR_LIT>" . format ( self . token ) <EOL> return s <EOL> def get_kwargs ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> bits = [ <EOL> self . version , <EOL> ] <EOL> if "<STR_LIT>" in kwargs : <EOL> bits . extend ( [ <EOL> "<STR_LIT>" , <EOL> kwargs . pop ( "<STR_LIT>" ) , <EOL> ] ) <EOL> url = kwargs . get ( "<STR_LIT:url>" , "<STR_LIT>" ) <EOL> if url . startswith ( "<STR_LIT:/>" ) : <EOL> url = url [ <NUM_LIT:1> : ] <EOL> bits . append ( url ) <EOL> kwargs [ "<STR_LIT:url>" ] = self . url + posixpath . join ( * bits ) <EOL> return kwargs <EOL> def request ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . session . request ( * args , ** self . get_kwargs ( ** kwargs ) ) <EOL> def get ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . session . get ( * args , ** self . get_kwargs ( ** kwargs ) ) </s>
