<s> import functools <EOL> import logging <EOL> import six <EOL> from oslo_utils . timeutils import parse_isotime <EOL> from barbicanclient import acls as acl_manager <EOL> from barbicanclient import base <EOL> from barbicanclient import formatter <EOL> from barbicanclient import secrets as secret_manager <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def _immutable_after_save ( func ) : <EOL> @ functools . wraps ( func ) <EOL> def wrapper ( self , * args ) : <EOL> if hasattr ( self , '<STR_LIT>' ) and self . _container_ref : <EOL> raise base . ImmutableException ( ) <EOL> return func ( self , * args ) <EOL> return wrapper <EOL> class ContainerFormatter ( formatter . EntityFormatter ) : <EOL> columns = ( "<STR_LIT>" , <EOL> "<STR_LIT:Name>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ) <EOL> def _get_formatted_data ( self ) : <EOL> formatted_secrets = None <EOL> formatted_consumers = None <EOL> if self . secrets : <EOL> formatted_secrets = '<STR_LIT:\n>' . join ( ( <EOL> '<STR_LIT:=>' . join ( ( name , secret_ref ) ) <EOL> for name , secret_ref in six . iteritems ( self . secret_refs ) <EOL> ) ) <EOL> if self . consumers : <EOL> formatted_consumers = '<STR_LIT:\n>' . join ( ( str ( c ) for c in self . consumers ) ) <EOL> data = ( self . container_ref , <EOL> self . name , <EOL> self . created , <EOL> self . status , <EOL> self . _type , <EOL> formatted_secrets , <EOL> formatted_consumers , <EOL> ) <EOL> return data <EOL> class Container ( ContainerFormatter ) : <EOL> """<STR_LIT>""" <EOL> _entity = '<STR_LIT>' <EOL> _type = '<STR_LIT>' <EOL> def __init__ ( self , api , name = None , secrets = None , consumers = None , <EOL> container_ref = None , created = None , updated = None , status = None , <EOL> secret_refs = None ) : <EOL> self . _api = api <EOL> self . _secret_manager = secret_manager . SecretManager ( api ) <EOL> self . _name = name <EOL> self . _container_ref = container_ref <EOL> self . _secret_refs = secret_refs <EOL> self . _cached_secrets = dict ( ) <EOL> self . _initialize_secrets ( secrets ) <EOL> if container_ref : <EOL> self . _consumers = consumers if consumers else list ( ) <EOL> self . _created = parse_isotime ( created ) if created else None <EOL> self . _updated = parse_isotime ( updated ) if updated else None <EOL> self . _status = status <EOL> else : <EOL> self . _consumers = list ( ) <EOL> self . _created = None <EOL> self . _updated = None <EOL> self . _status = None <EOL> self . _acl_manager = acl_manager . ACLManager ( api ) <EOL> self . _acls = None <EOL> def _initialize_secrets ( self , secrets ) : <EOL> try : <EOL> self . _fill_secrets_from_secret_refs ( ) <EOL> except Exception : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if secrets : <EOL> try : <EOL> for name , secret in six . iteritems ( secrets ) : <EOL> self . add ( name , secret ) <EOL> except Exception : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def _fill_secrets_from_secret_refs ( self ) : <EOL> if self . _secret_refs : <EOL> self . _cached_secrets = dict ( <EOL> ( name . lower ( ) , self . _secret_manager . get ( secret_ref = secret_ref ) ) <EOL> for name , secret_ref in six . iteritems ( self . _secret_refs ) <EOL> ) <EOL> @ property <EOL> def container_ref ( self ) : <EOL> return self . _container_ref <EOL> @ property <EOL> def name ( self ) : <EOL> if self . _container_ref and not self . _name : <EOL> self . _reload ( ) <EOL> return self . _name <EOL> @ property <EOL> def created ( self ) : <EOL> return self . _created <EOL> @ property <EOL> def updated ( self ) : <EOL> return self . _updated <EOL> @ property <EOL> def status ( self ) : <EOL> if self . _container_ref and not self . _status : <EOL> self . _reload ( ) <EOL> return self . _status <EOL> @ property <EOL> def acls ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _container_ref and not self . _acls : <EOL> self . _acls = self . _acl_manager . get ( self . container_ref ) <EOL> return self . _acls <EOL> @ property <EOL> def secret_refs ( self ) : <EOL> if self . _cached_secrets : <EOL> self . _secret_refs = dict ( <EOL> ( name , secret . secret_ref ) <EOL> for name , secret in six . iteritems ( self . _cached_secrets ) <EOL> ) <EOL> return self . _secret_refs <EOL> @ property <EOL> def secrets ( self , cache = True ) : <EOL> """<STR_LIT>""" <EOL> if not self . _cached_secrets or not cache : <EOL> self . _fill_secrets_from_secret_refs ( ) <EOL> return self . _cached_secrets <EOL> @ property <EOL> def consumers ( self ) : <EOL> return self . _consumers <EOL> @ name . setter <EOL> @ _immutable_after_save <EOL> def name ( self , value ) : <EOL> self . _name = value <EOL> @ _immutable_after_save <EOL> def add ( self , name , secret ) : <EOL> if not isinstance ( secret , secret_manager . Secret ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if name . lower ( ) in self . secrets : <EOL> raise KeyError ( "<STR_LIT>" ) <EOL> self . _cached_secrets [ name . lower ( ) ] = secret <EOL> @ _immutable_after_save <EOL> def remove ( self , name ) : <EOL> self . _cached_secrets . pop ( name . lower ( ) , None ) <EOL> if self . _secret_refs : <EOL> self . _secret_refs . pop ( name . lower ( ) , None ) <EOL> @ _immutable_after_save <EOL> def store ( self ) : <EOL> """<STR_LIT>""" <EOL> secret_refs = self . _get_secrets_and_store_them_if_necessary ( ) <EOL> container_dict = base . filter_null_keys ( { <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT:type>' : self . _type , <EOL> '<STR_LIT>' : secret_refs <EOL> } ) <EOL> LOG . debug ( "<STR_LIT>" . format ( container_dict ) ) <EOL> response = self . _api . post ( self . _entity , json = container_dict ) <EOL> if response : <EOL> self . _container_ref = response [ '<STR_LIT>' ] <EOL> return self . container_ref <EOL> def delete ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _container_ref : <EOL> self . _api . delete ( self . _container_ref ) <EOL> self . _container_ref = None <EOL> self . _status = None <EOL> self . _created = None <EOL> self . _updated = None <EOL> else : <EOL> raise LookupError ( "<STR_LIT>" ) <EOL> def _get_secrets_and_store_them_if_necessary ( self ) : <EOL> LOG . debug ( "<STR_LIT>" . format ( self . secrets ) ) <EOL> secret_refs = [ ] <EOL> for name , secret in six . iteritems ( self . secrets ) : <EOL> if secret and not secret . secret_ref : <EOL> secret . store ( ) <EOL> secret_refs . append ( { '<STR_LIT:name>' : name , '<STR_LIT>' : secret . secret_ref } ) <EOL> return secret_refs <EOL> def _reload ( self ) : <EOL> if not self . _container_ref : <EOL> raise AttributeError ( "<STR_LIT>" ) <EOL> LOG . debug ( '<STR_LIT>' <EOL> . format ( self . _container_ref ) ) <EOL> base . validate_ref ( self . _container_ref , '<STR_LIT>' ) <EOL> try : <EOL> response = self . _api . get ( self . _container_ref ) <EOL> except AttributeError : <EOL> raise LookupError ( '<STR_LIT>' <EOL> . format ( self . _container_ref ) ) <EOL> self . _name = response . get ( '<STR_LIT:name>' ) <EOL> self . _consumers = response . get ( '<STR_LIT>' , [ ] ) <EOL> created = response . get ( '<STR_LIT>' ) <EOL> updated = response . get ( '<STR_LIT>' ) <EOL> self . _created = parse_isotime ( created ) if created else None <EOL> self . _updated = parse_isotime ( updated ) if updated else None <EOL> self . _status = response . get ( '<STR_LIT:status>' ) <EOL> def _get_named_secret ( self , name ) : <EOL> return self . secrets . get ( name ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . name ) <EOL> class RSAContainerFormatter ( formatter . EntityFormatter ) : <EOL> _get_generic_data = ContainerFormatter . _get_formatted_data <EOL> def _get_generic_columns ( self ) : <EOL> return ContainerFormatter . columns <EOL> columns = ( "<STR_LIT>" , <EOL> "<STR_LIT:Name>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ) <EOL> def _get_formatted_data ( self ) : <EOL> formatted_public_key = None <EOL> formatted_private_key = None <EOL> formatted_pkp = None <EOL> formatted_consumers = None <EOL> if self . public_key : <EOL> formatted_public_key = self . public_key . secret_ref <EOL> if self . private_key : <EOL> formatted_private_key = self . private_key . secret_ref <EOL> if self . private_key_passphrase : <EOL> formatted_pkp = self . private_key_passphrase . secret_ref <EOL> if self . consumers : <EOL> formatted_consumers = '<STR_LIT:\n>' . join ( ( str ( c ) for c in self . consumers ) ) <EOL> data = ( self . container_ref , <EOL> self . name , <EOL> self . created , <EOL> self . status , <EOL> self . _type , <EOL> formatted_public_key , <EOL> formatted_private_key , <EOL> formatted_pkp , <EOL> formatted_consumers , <EOL> ) <EOL> return data <EOL> class RSAContainer ( RSAContainerFormatter , Container ) : <EOL> _required_secrets = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> _optional_secrets = [ "<STR_LIT>" ] <EOL> _type = '<STR_LIT>' <EOL> def __init__ ( self , api , name = None , public_key = None , private_key = None , <EOL> private_key_passphrase = None , consumers = [ ] , container_ref = None , <EOL> created = None , updated = None , status = None , public_key_ref = None , <EOL> private_key_ref = None , private_key_passphrase_ref = None ) : <EOL> secret_refs = { } <EOL> if public_key_ref : <EOL> secret_refs [ '<STR_LIT>' ] = public_key_ref <EOL> if private_key_ref : <EOL> secret_refs [ '<STR_LIT>' ] = private_key_ref <EOL> if private_key_passphrase_ref : <EOL> secret_refs [ '<STR_LIT>' ] = private_key_passphrase_ref <EOL> super ( RSAContainer , self ) . __init__ ( <EOL> api = api , <EOL> name = name , <EOL> consumers = consumers , <EOL> container_ref = container_ref , <EOL> created = created , <EOL> updated = updated , <EOL> status = status , <EOL> secret_refs = secret_refs <EOL> ) <EOL> if public_key : <EOL> self . public_key = public_key <EOL> if private_key : <EOL> self . private_key = private_key <EOL> if private_key_passphrase : <EOL> self . private_key_passphrase = private_key_passphrase <EOL> @ property <EOL> def public_key ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_named_secret ( "<STR_LIT>" ) <EOL> @ property <EOL> def private_key ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_named_secret ( "<STR_LIT>" ) <EOL> @ property <EOL> def private_key_passphrase ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_named_secret ( "<STR_LIT>" ) <EOL> @ public_key . setter <EOL> @ _immutable_after_save <EOL> def public_key ( self , value ) : <EOL> super ( RSAContainer , self ) . remove ( "<STR_LIT>" ) <EOL> super ( RSAContainer , self ) . add ( "<STR_LIT>" , value ) <EOL> @ private_key . setter <EOL> @ _immutable_after_save <EOL> def private_key ( self , value ) : <EOL> super ( RSAContainer , self ) . remove ( "<STR_LIT>" ) <EOL> super ( RSAContainer , self ) . add ( "<STR_LIT>" , value ) <EOL> @ private_key_passphrase . setter <EOL> @ _immutable_after_save <EOL> def private_key_passphrase ( self , value ) : <EOL> super ( RSAContainer , self ) . remove ( "<STR_LIT>" ) <EOL> super ( RSAContainer , self ) . add ( "<STR_LIT>" , value ) <EOL> def add ( self , name , sec ) : <EOL> raise NotImplementedError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . name ) <EOL> class CertificateContainerFormatter ( formatter . EntityFormatter ) : <EOL> _get_generic_data = ContainerFormatter . _get_formatted_data <EOL> def _get_generic_columns ( self ) : <EOL> return ContainerFormatter . columns <EOL> columns = ( "<STR_LIT>" , <EOL> "<STR_LIT:Name>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ) <EOL> def _get_formatted_data ( self ) : <EOL> formatted_certificate = None <EOL> formatted_private_key = None <EOL> formatted_pkp = None <EOL> formatted_intermediates = None <EOL> formatted_consumers = None <EOL> if self . certificate : <EOL> formatted_certificate = self . certificate . secret_ref <EOL> if self . intermediates : <EOL> formatted_intermediates = self . intermediates . secret_ref <EOL> if self . private_key : <EOL> formatted_private_key = self . private_key . secret_ref <EOL> if self . private_key_passphrase : <EOL> formatted_pkp = self . private_key_passphrase . secret_ref <EOL> if self . consumers : <EOL> formatted_consumers = '<STR_LIT:\n>' . join ( ( str ( c ) for c in self . consumers ) ) <EOL> data = ( self . container_ref , <EOL> self . name , <EOL> self . created , <EOL> self . status , <EOL> self . _type , <EOL> formatted_certificate , <EOL> formatted_intermediates , <EOL> formatted_private_key , <EOL> formatted_pkp , <EOL> formatted_consumers , <EOL> ) <EOL> return data <EOL> class CertificateContainer ( CertificateContainerFormatter , Container ) : <EOL> _required_secrets = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> _optional_secrets = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> _type = '<STR_LIT>' <EOL> def __init__ ( self , api , name = None , certificate = None , intermediates = None , <EOL> private_key = None , private_key_passphrase = None , consumers = [ ] , <EOL> container_ref = None , created = None , updated = None , status = None , <EOL> certificate_ref = None , intermediates_ref = None , <EOL> private_key_ref = None , private_key_passphrase_ref = None ) : <EOL> secret_refs = { } <EOL> if certificate_ref : <EOL> secret_refs [ '<STR_LIT>' ] = certificate_ref <EOL> if intermediates_ref : <EOL> secret_refs [ '<STR_LIT>' ] = intermediates_ref <EOL> if private_key_ref : <EOL> secret_refs [ '<STR_LIT>' ] = private_key_ref <EOL> if private_key_passphrase_ref : <EOL> secret_refs [ '<STR_LIT>' ] = private_key_passphrase_ref <EOL> super ( CertificateContainer , self ) . __init__ ( <EOL> api = api , <EOL> name = name , <EOL> consumers = consumers , <EOL> container_ref = container_ref , <EOL> created = created , <EOL> updated = updated , <EOL> status = status , <EOL> secret_refs = secret_refs <EOL> ) <EOL> if certificate : <EOL> self . certificate = certificate <EOL> if intermediates : <EOL> self . intermediates = intermediates <EOL> if private_key : <EOL> self . private_key = private_key <EOL> if private_key_passphrase : <EOL> self . private_key_passphrase = private_key_passphrase <EOL> @ property <EOL> def certificate ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_named_secret ( "<STR_LIT>" ) <EOL> @ property <EOL> def private_key ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_named_secret ( "<STR_LIT>" ) <EOL> @ property <EOL> def private_key_passphrase ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_named_secret ( "<STR_LIT>" ) <EOL> @ property <EOL> def intermediates ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_named_secret ( "<STR_LIT>" ) <EOL> @ certificate . setter <EOL> @ _immutable_after_save <EOL> def certificate ( self , value ) : <EOL> super ( CertificateContainer , self ) . remove ( "<STR_LIT>" ) <EOL> super ( CertificateContainer , self ) . add ( "<STR_LIT>" , value ) <EOL> @ private_key . setter <EOL> @ _immutable_after_save <EOL> def private_key ( self , value ) : <EOL> super ( CertificateContainer , self ) . remove ( "<STR_LIT>" ) <EOL> super ( CertificateContainer , self ) . add ( "<STR_LIT>" , value ) <EOL> @ private_key_passphrase . setter <EOL> @ _immutable_after_save <EOL> def private_key_passphrase ( self , value ) : <EOL> super ( CertificateContainer , self ) . remove ( "<STR_LIT>" ) <EOL> super ( CertificateContainer , self ) . add ( "<STR_LIT>" , value ) <EOL> @ intermediates . setter <EOL> @ _immutable_after_save <EOL> def intermediates ( self , value ) : <EOL> super ( CertificateContainer , self ) . remove ( "<STR_LIT>" ) <EOL> super ( CertificateContainer , self ) . add ( "<STR_LIT>" , value ) <EOL> def add ( self , name , sec ) : <EOL> raise NotImplementedError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( self . name ) <EOL> class ContainerManager ( base . BaseEntityManager ) : <EOL> """<STR_LIT>""" <EOL> _container_map = { <EOL> '<STR_LIT>' : Container , <EOL> '<STR_LIT>' : RSAContainer , <EOL> '<STR_LIT>' : CertificateContainer <EOL> } <EOL> def __init__ ( self , api ) : <EOL> super ( ContainerManager , self ) . __init__ ( api , '<STR_LIT>' ) <EOL> def get ( self , container_ref ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( '<STR_LIT>' <EOL> . format ( container_ref ) ) <EOL> base . validate_ref ( container_ref , '<STR_LIT>' ) <EOL> try : <EOL> response = self . _api . get ( container_ref ) <EOL> except AttributeError : <EOL> raise LookupError ( '<STR_LIT>' <EOL> . format ( container_ref ) ) <EOL> return self . _generate_typed_container ( response ) <EOL> def _generate_typed_container ( self , response ) : <EOL> resp_type = response . get ( '<STR_LIT:type>' , '<STR_LIT>' ) . lower ( ) <EOL> container_type = self . _container_map . get ( resp_type ) <EOL> if not container_type : <EOL> raise TypeError ( '<STR_LIT>' <EOL> . format ( resp_type ) ) <EOL> name = response . get ( '<STR_LIT:name>' ) <EOL> consumers = response . get ( '<STR_LIT>' , [ ] ) <EOL> container_ref = response . get ( '<STR_LIT>' ) <EOL> created = response . get ( '<STR_LIT>' ) <EOL> updated = response . get ( '<STR_LIT>' ) <EOL> status = response . get ( '<STR_LIT:status>' ) <EOL> secret_refs = self . _translate_secret_refs_from_json ( <EOL> response . get ( '<STR_LIT>' ) <EOL> ) <EOL> if container_type is RSAContainer : <EOL> public_key_ref = secret_refs . get ( '<STR_LIT>' ) <EOL> private_key_ref = secret_refs . get ( '<STR_LIT>' ) <EOL> private_key_pass_ref = secret_refs . get ( '<STR_LIT>' ) <EOL> return RSAContainer ( <EOL> api = self . _api , <EOL> name = name , <EOL> consumers = consumers , <EOL> container_ref = container_ref , <EOL> created = created , <EOL> updated = updated , <EOL> status = status , <EOL> public_key_ref = public_key_ref , <EOL> private_key_ref = private_key_ref , <EOL> private_key_passphrase_ref = private_key_pass_ref , <EOL> ) <EOL> elif container_type is CertificateContainer : <EOL> certificate_ref = secret_refs . get ( '<STR_LIT>' ) <EOL> intermediates_ref = secret_refs . get ( '<STR_LIT>' ) <EOL> private_key_ref = secret_refs . get ( '<STR_LIT>' ) <EOL> private_key_pass_ref = secret_refs . get ( '<STR_LIT>' ) <EOL> return CertificateContainer ( <EOL> api = self . _api , <EOL> name = name , <EOL> consumers = consumers , <EOL> container_ref = container_ref , <EOL> created = created , <EOL> updated = updated , <EOL> status = status , <EOL> certificate_ref = certificate_ref , <EOL> intermediates_ref = intermediates_ref , <EOL> private_key_ref = private_key_ref , <EOL> private_key_passphrase_ref = private_key_pass_ref , <EOL> ) <EOL> return container_type ( <EOL> api = self . _api , <EOL> name = name , <EOL> secret_refs = secret_refs , <EOL> consumers = consumers , <EOL> container_ref = container_ref , <EOL> created = created , <EOL> updated = updated , <EOL> status = status <EOL> ) <EOL> @ staticmethod <EOL> def _translate_secret_refs_from_json ( json_refs ) : <EOL> return dict ( <EOL> ( ref_pack . get ( '<STR_LIT:name>' ) , ref_pack . get ( '<STR_LIT>' ) ) <EOL> for ref_pack in json_refs <EOL> ) <EOL> def create ( self , name = None , secrets = None ) : <EOL> """<STR_LIT>""" <EOL> return Container ( <EOL> api = self . _api , <EOL> name = name , <EOL> secrets = secrets <EOL> ) <EOL> def create_rsa ( self , name = None , public_key = None , private_key = None , <EOL> private_key_passphrase = None ) : <EOL> """<STR_LIT>""" <EOL> return RSAContainer ( <EOL> api = self . _api , <EOL> name = name , <EOL> public_key = public_key , <EOL> private_key = private_key , <EOL> private_key_passphrase = private_key_passphrase <EOL> ) <EOL> def create_certificate ( self , name = None , certificate = None , <EOL> intermediates = None , private_key = None , <EOL> private_key_passphrase = None ) : <EOL> """<STR_LIT>""" <EOL> return CertificateContainer ( <EOL> api = self . _api , <EOL> name = name , <EOL> certificate = certificate , <EOL> intermediates = intermediates , <EOL> private_key = private_key , <EOL> private_key_passphrase = private_key_passphrase <EOL> ) <EOL> def delete ( self , container_ref ) : <EOL> """<STR_LIT>""" <EOL> if not container_ref : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . _api . delete ( container_ref ) <EOL> def list ( self , limit = <NUM_LIT:10> , offset = <NUM_LIT:0> , name = None , type = None ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( '<STR_LIT>' <EOL> . format ( offset , limit , name , type ) ) <EOL> params = { '<STR_LIT>' : limit , '<STR_LIT>' : offset } <EOL> if name : <EOL> params [ '<STR_LIT:name>' ] = name <EOL> if type : <EOL> params [ '<STR_LIT:type>' ] = type <EOL> response = self . _api . get ( self . _entity , params = params ) <EOL> return [ self . _generate_typed_container ( container ) <EOL> for container in response . get ( '<STR_LIT>' , [ ] ) ] <EOL> def register_consumer ( self , container_ref , name , url ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( container_ref , name , url ) ) <EOL> href = '<STR_LIT>' . format ( self . _entity , <EOL> container_ref . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:1> ] ) <EOL> consumer_dict = dict ( ) <EOL> consumer_dict [ '<STR_LIT:name>' ] = name <EOL> consumer_dict [ '<STR_LIT>' ] = url <EOL> response = self . _api . post ( href , json = consumer_dict ) <EOL> return self . _generate_typed_container ( response ) <EOL> def remove_consumer ( self , container_ref , name , url ) : <EOL> """<STR_LIT>""" <EOL> LOG . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( container_ref , name , url ) ) <EOL> href = '<STR_LIT>' . format ( self . _entity , <EOL> container_ref . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:1> ] ) <EOL> consumer_dict = { <EOL> '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : url <EOL> } <EOL> self . _api . delete ( href , json = consumer_dict ) </s>
