<s> import logging <EOL> import six <EOL> import warnings <EOL> from . . auth import auth <EOL> from . . constants import INSECURE_REGISTRY_DEPRECATION_WARNING <EOL> from . . import utils <EOL> from . . import errors <EOL> log = logging . getLogger ( __name__ ) <EOL> class ImageApiMixin ( object ) : <EOL> @ utils . check_resource <EOL> def get_image ( self , image ) : <EOL> res = self . _get ( self . _url ( "<STR_LIT>" , image ) , stream = True ) <EOL> self . _raise_for_status ( res ) <EOL> return res . raw <EOL> @ utils . check_resource <EOL> def history ( self , image ) : <EOL> res = self . _get ( self . _url ( "<STR_LIT>" , image ) ) <EOL> return self . _result ( res , True ) <EOL> def images ( self , name = None , quiet = False , all = False , viz = False , <EOL> filters = None ) : <EOL> if viz : <EOL> if utils . compare_version ( '<STR_LIT>' , self . _version ) >= <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> return self . _result ( self . _get ( self . _url ( "<STR_LIT>" ) ) ) <EOL> params = { <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT>' : <NUM_LIT:1> if quiet else <NUM_LIT:0> , <EOL> '<STR_LIT:all>' : <NUM_LIT:1> if all else <NUM_LIT:0> , <EOL> } <EOL> if filters : <EOL> params [ '<STR_LIT>' ] = utils . convert_filters ( filters ) <EOL> res = self . _result ( self . _get ( self . _url ( "<STR_LIT>" ) , params = params ) , <EOL> True ) <EOL> if quiet : <EOL> return [ x [ '<STR_LIT>' ] for x in res ] <EOL> return res <EOL> def import_image ( self , src = None , repository = None , tag = None , image = None ) : <EOL> if src : <EOL> if isinstance ( src , six . string_types ) : <EOL> try : <EOL> result = self . import_image_from_file ( <EOL> src , repository = repository , tag = tag ) <EOL> except IOError : <EOL> result = self . import_image_from_url ( <EOL> src , repository = repository , tag = tag ) <EOL> else : <EOL> result = self . import_image_from_data ( <EOL> src , repository = repository , tag = tag ) <EOL> elif image : <EOL> result = self . import_image_from_image ( <EOL> image , repository = repository , tag = tag ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> return result <EOL> def import_image_from_data ( self , data , repository = None , tag = None ) : <EOL> u = self . _url ( "<STR_LIT>" ) <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT:->' , <EOL> '<STR_LIT>' : repository , <EOL> '<STR_LIT>' : tag <EOL> } <EOL> headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> } <EOL> return self . _result ( <EOL> self . _post ( u , data = data , params = params , headers = headers ) ) <EOL> def import_image_from_file ( self , filename , repository = None , tag = None ) : <EOL> u = self . _url ( "<STR_LIT>" ) <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT:->' , <EOL> '<STR_LIT>' : repository , <EOL> '<STR_LIT>' : tag <EOL> } <EOL> headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> } <EOL> with open ( filename , '<STR_LIT:rb>' ) as f : <EOL> return self . _result ( <EOL> self . _post ( u , data = f , params = params , headers = headers , <EOL> timeout = None ) ) <EOL> def import_image_from_stream ( self , stream , repository = None , tag = None ) : <EOL> u = self . _url ( "<STR_LIT>" ) <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT:->' , <EOL> '<STR_LIT>' : repository , <EOL> '<STR_LIT>' : tag <EOL> } <EOL> headers = { <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> return self . _result ( <EOL> self . _post ( u , data = stream , params = params , headers = headers ) ) <EOL> def import_image_from_url ( self , url , repository = None , tag = None ) : <EOL> u = self . _url ( "<STR_LIT>" ) <EOL> params = { <EOL> '<STR_LIT>' : url , <EOL> '<STR_LIT>' : repository , <EOL> '<STR_LIT>' : tag <EOL> } <EOL> return self . _result ( <EOL> self . _post ( u , data = None , params = params ) ) <EOL> def import_image_from_image ( self , image , repository = None , tag = None ) : <EOL> u = self . _url ( "<STR_LIT>" ) <EOL> params = { <EOL> '<STR_LIT>' : image , <EOL> '<STR_LIT>' : repository , <EOL> '<STR_LIT>' : tag <EOL> } <EOL> return self . _result ( <EOL> self . _post ( u , data = None , params = params ) ) <EOL> @ utils . check_resource <EOL> def insert ( self , image , url , path ) : <EOL> if utils . compare_version ( '<STR_LIT>' , self . _version ) >= <NUM_LIT:0> : <EOL> raise errors . DeprecatedMethod ( <EOL> '<STR_LIT>' <EOL> ) <EOL> api_url = self . _url ( "<STR_LIT>" , image ) <EOL> params = { <EOL> '<STR_LIT:url>' : url , <EOL> '<STR_LIT:path>' : path <EOL> } <EOL> return self . _result ( self . _post ( api_url , params = params ) ) <EOL> @ utils . check_resource <EOL> def inspect_image ( self , image ) : <EOL> return self . _result ( <EOL> self . _get ( self . _url ( "<STR_LIT>" , image ) ) , True <EOL> ) <EOL> def load_image ( self , data ) : <EOL> res = self . _post ( self . _url ( "<STR_LIT>" ) , data = data ) <EOL> self . _raise_for_status ( res ) <EOL> def pull ( self , repository , tag = None , stream = False , <EOL> insecure_registry = False , auth_config = None , decode = False ) : <EOL> if insecure_registry : <EOL> warnings . warn ( <EOL> INSECURE_REGISTRY_DEPRECATION_WARNING . format ( '<STR_LIT>' ) , <EOL> DeprecationWarning <EOL> ) <EOL> if not tag : <EOL> repository , tag = utils . parse_repository_tag ( repository ) <EOL> registry , repo_name = auth . resolve_repository_name ( repository ) <EOL> params = { <EOL> '<STR_LIT>' : tag , <EOL> '<STR_LIT>' : repository <EOL> } <EOL> headers = { } <EOL> if utils . compare_version ( '<STR_LIT>' , self . _version ) >= <NUM_LIT:0> : <EOL> if auth_config is None : <EOL> log . debug ( '<STR_LIT>' ) <EOL> if not self . _auth_configs : <EOL> log . debug ( <EOL> "<STR_LIT>" <EOL> ) <EOL> self . _auth_configs = auth . load_config ( ) <EOL> authcfg = auth . resolve_authconfig ( self . _auth_configs , registry ) <EOL> if authcfg : <EOL> log . debug ( '<STR_LIT>' ) <EOL> headers [ '<STR_LIT>' ] = auth . encode_header ( <EOL> authcfg <EOL> ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' ) <EOL> headers [ '<STR_LIT>' ] = auth . encode_header ( auth_config ) <EOL> response = self . _post ( <EOL> self . _url ( '<STR_LIT>' ) , params = params , headers = headers , <EOL> stream = stream , timeout = None <EOL> ) <EOL> self . _raise_for_status ( response ) <EOL> if stream : <EOL> return self . _stream_helper ( response , decode = decode ) <EOL> return self . _result ( response ) <EOL> def push ( self , repository , tag = None , stream = False , <EOL> insecure_registry = False , decode = False ) : <EOL> if insecure_registry : <EOL> warnings . warn ( <EOL> INSECURE_REGISTRY_DEPRECATION_WARNING . format ( '<STR_LIT>' ) , <EOL> DeprecationWarning <EOL> ) <EOL> if not tag : <EOL> repository , tag = utils . parse_repository_tag ( repository ) <EOL> registry , repo_name = auth . resolve_repository_name ( repository ) <EOL> u = self . _url ( "<STR_LIT>" , repository ) <EOL> params = { <EOL> '<STR_LIT>' : tag <EOL> } <EOL> headers = { } <EOL> if utils . compare_version ( '<STR_LIT>' , self . _version ) >= <NUM_LIT:0> : <EOL> if not self . _auth_configs : <EOL> self . _auth_configs = auth . load_config ( ) <EOL> authcfg = auth . resolve_authconfig ( self . _auth_configs , registry ) <EOL> if authcfg : <EOL> headers [ '<STR_LIT>' ] = auth . encode_header ( authcfg ) <EOL> response = self . _post_json ( <EOL> u , None , headers = headers , stream = stream , params = params <EOL> ) <EOL> self . _raise_for_status ( response ) <EOL> if stream : <EOL> return self . _stream_helper ( response , decode = decode ) <EOL> return self . _result ( response ) <EOL> @ utils . check_resource <EOL> def remove_image ( self , image , force = False , noprune = False ) : <EOL> params = { '<STR_LIT>' : force , '<STR_LIT>' : noprune } <EOL> res = self . _delete ( self . _url ( "<STR_LIT>" , image ) , params = params ) <EOL> self . _raise_for_status ( res ) <EOL> def search ( self , term ) : <EOL> return self . _result ( <EOL> self . _get ( self . _url ( "<STR_LIT>" ) , params = { '<STR_LIT>' : term } ) , <EOL> True <EOL> ) <EOL> @ utils . check_resource <EOL> def tag ( self , image , repository , tag = None , force = False ) : <EOL> params = { <EOL> '<STR_LIT>' : tag , <EOL> '<STR_LIT>' : repository , <EOL> '<STR_LIT>' : <NUM_LIT:1> if force else <NUM_LIT:0> <EOL> } <EOL> url = self . _url ( "<STR_LIT>" , image ) <EOL> res = self . _post ( url , params = params ) <EOL> self . _raise_for_status ( res ) <EOL> return res . status_code == <NUM_LIT> </s>
