<s> """<STR_LIT>""" <EOL> import base64 <EOL> import copy <EOL> import tempfile <EOL> import os <EOL> import yaml <EOL> from pykube import exceptions <EOL> class KubeConfig ( object ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def from_service_account ( cls , path = "<STR_LIT>" ) : <EOL> with open ( os . path . join ( path , "<STR_LIT>" ) ) as fp : <EOL> token = fp . read ( ) <EOL> host = os . environ . get ( "<STR_LIT>" ) <EOL> if host is None : <EOL> host = os . environ [ "<STR_LIT>" ] <EOL> port = os . environ . get ( "<STR_LIT>" ) <EOL> if port is None : <EOL> port = os . environ [ "<STR_LIT>" ] <EOL> doc = { <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" . format ( host , port ) , <EOL> "<STR_LIT>" : os . path . join ( path , "<STR_LIT>" ) , <EOL> } , <EOL> } , <EOL> ] , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:user>" : { <EOL> "<STR_LIT>" : token , <EOL> } , <EOL> } , <EOL> ] , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:user>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> self = cls ( doc ) <EOL> return self <EOL> @ classmethod <EOL> def from_file ( cls , filename ) : <EOL> """<STR_LIT>""" <EOL> filename = os . path . expanduser ( filename ) <EOL> if not os . path . isfile ( filename ) : <EOL> raise exceptions . PyKubeError ( "<STR_LIT>" . format ( filename ) ) <EOL> with open ( filename ) as f : <EOL> doc = yaml . safe_load ( f . read ( ) ) <EOL> self = cls ( doc ) <EOL> self . filename = filename <EOL> return self <EOL> def __init__ ( self , doc ) : <EOL> """<STR_LIT>""" <EOL> self . doc = doc <EOL> self . current_context = None <EOL> if "<STR_LIT>" in doc and doc [ "<STR_LIT>" ] : <EOL> self . set_current_context ( doc [ "<STR_LIT>" ] ) <EOL> def set_current_context ( self , value ) : <EOL> """<STR_LIT>""" <EOL> self . current_context = value <EOL> @ property <EOL> def clusters ( self ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> cs = { } <EOL> for cr in self . doc [ "<STR_LIT>" ] : <EOL> cs [ cr [ "<STR_LIT:name>" ] ] = c = copy . deepcopy ( cr [ "<STR_LIT>" ] ) <EOL> if "<STR_LIT>" not in c : <EOL> c [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> BytesOrFile . maybe_set ( c , "<STR_LIT>" ) <EOL> self . _clusters = cs <EOL> return self . _clusters <EOL> @ property <EOL> def users ( self ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> us = { } <EOL> for ur in self . doc [ "<STR_LIT>" ] : <EOL> us [ ur [ "<STR_LIT:name>" ] ] = u = copy . deepcopy ( ur [ "<STR_LIT:user>" ] ) <EOL> BytesOrFile . maybe_set ( u , "<STR_LIT>" ) <EOL> BytesOrFile . maybe_set ( u , "<STR_LIT>" ) <EOL> self . _users = us <EOL> return self . _users <EOL> @ property <EOL> def contexts ( self ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( self , "<STR_LIT>" ) : <EOL> cs = { } <EOL> for cr in self . doc [ "<STR_LIT>" ] : <EOL> cs [ cr [ "<STR_LIT:name>" ] ] = copy . deepcopy ( cr [ "<STR_LIT>" ] ) <EOL> self . _contexts = cs <EOL> return self . _contexts <EOL> @ property <EOL> def cluster ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . current_context is None : <EOL> raise exceptions . PyKubeError ( "<STR_LIT>" ) <EOL> return self . clusters [ self . contexts [ self . current_context ] [ "<STR_LIT>" ] ] <EOL> @ property <EOL> def user ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . current_context is None : <EOL> raise exceptions . PyKubeError ( "<STR_LIT>" ) <EOL> return self . users [ self . contexts [ self . current_context ] [ "<STR_LIT:user>" ] ] <EOL> class BytesOrFile ( object ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def maybe_set ( cls , d , key ) : <EOL> file_key = key <EOL> data_key = "<STR_LIT>" . format ( key ) <EOL> if data_key in d : <EOL> d [ file_key ] = cls ( d [ data_key ] ) <EOL> del d [ data_key ] <EOL> elif file_key in d : <EOL> d [ file_key ] = cls ( d [ file_key ] ) <EOL> def __init__ ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . _filename = None <EOL> self . _bytes = None <EOL> if data . startswith ( "<STR_LIT:/>" ) : <EOL> self . _filename = data <EOL> else : <EOL> self . _bytes = base64 . b64decode ( data ) <EOL> def bytes ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _filename : <EOL> with open ( self . _filename , "<STR_LIT:rb>" ) as f : <EOL> return f . read ( ) <EOL> else : <EOL> return self . _bytes <EOL> def filename ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _filename : <EOL> return self . _filename <EOL> else : <EOL> with tempfile . NamedTemporaryFile ( delete = False ) as f : <EOL> f . write ( self . _bytes ) <EOL> return f . name </s>
