<s> """<STR_LIT>""" <EOL> import logging <EOL> from keystoneclient import exceptions as ks_exceptions <EOL> from oslo_utils import encodeutils <EOL> from glance_store import exceptions <EOL> from glance_store . i18n import _ <EOL> from glance_store . i18n import _LI <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class SwiftConnectionManager ( object ) : <EOL> """<STR_LIT>""" <EOL> AUTH_HEADER_NAME = '<STR_LIT>' <EOL> def __init__ ( self , store , store_location , context = None , <EOL> allow_reauth = False ) : <EOL> """<STR_LIT>""" <EOL> self . _client = None <EOL> self . store = store <EOL> self . location = store_location <EOL> self . context = context <EOL> self . allow_reauth = allow_reauth <EOL> self . storage_url = self . _get_storage_url ( ) <EOL> self . connection = self . _init_connection ( ) <EOL> def get_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . allow_reauth : <EOL> auth_ref = self . client . session . auth . get_auth_ref ( <EOL> self . client . session ) <EOL> if auth_ref . will_expire_soon ( <EOL> self . store . conf . glance_store . swift_store_expire_soon_interval <EOL> ) : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) ) <EOL> auth_token = self . client . session . get_auth_headers ( ) . get ( <EOL> self . AUTH_HEADER_NAME ) <EOL> LOG . info ( _LI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> self . connection = self . store . get_store_connection ( <EOL> auth_token , self . storage_url ) <EOL> return self . connection <EOL> @ property <EOL> def client ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _client is None : <EOL> self . _client = self . _init_client ( ) <EOL> return self . _client <EOL> def _init_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> auth_token = self . client . session . get_auth_headers ( ) . get ( <EOL> self . AUTH_HEADER_NAME ) <EOL> return self . store . get_store_connection ( <EOL> auth_token , self . storage_url ) <EOL> def _init_client ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . store . init_client ( location = self . location , <EOL> context = self . context ) <EOL> def _get_storage_url ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> pass <EOL> class SingleTenantConnectionManager ( SwiftConnectionManager ) : <EOL> def _get_storage_url ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . store . auth_version == '<STR_LIT:3>' : <EOL> try : <EOL> return self . client . session . get_endpoint ( <EOL> service_type = self . store . service_type , <EOL> interface = self . store . endpoint_type , <EOL> region_name = self . store . region <EOL> ) <EOL> except Exception as e : <EOL> msg = _ ( "<STR_LIT>" <EOL> "<STR_LIT:%s>" ) % encodeutils . exception_to_unicode ( e ) <EOL> raise exceptions . BackendException ( msg ) <EOL> def _init_connection ( self ) : <EOL> if self . store . auth_version == '<STR_LIT:3>' : <EOL> return super ( SingleTenantConnectionManager , <EOL> self ) . _init_connection ( ) <EOL> else : <EOL> self . allow_reauth = False <EOL> return self . store . get_connection ( self . location , self . context ) <EOL> class MultiTenantConnectionManager ( SwiftConnectionManager ) : <EOL> def __init__ ( self , store , store_location , context = None , <EOL> allow_reauth = False ) : <EOL> if context is None : <EOL> reason = _ ( "<STR_LIT>" ) <EOL> raise exceptions . BadStoreConfiguration ( store_name = "<STR_LIT>" , <EOL> reason = reason ) <EOL> super ( MultiTenantConnectionManager , self ) . __init__ ( <EOL> store , store_location , context , allow_reauth ) <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> if self . _client and self . client . trust_id : <EOL> LOG . info ( _LI ( "<STR_LIT>" ) , self . client . trust_id ) <EOL> self . client . trusts . delete ( self . client . trust_id ) <EOL> def _get_storage_url ( self ) : <EOL> try : <EOL> return self . store . _get_endpoint ( self . context ) <EOL> except ( exceptions . BadStoreConfiguration , <EOL> ks_exceptions . EndpointNotFound ) as e : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , e ) <EOL> return self . location . swift_url <EOL> def _init_connection ( self ) : <EOL> if self . allow_reauth : <EOL> try : <EOL> return super ( MultiTenantConnectionManager , <EOL> self ) . _init_connection ( ) <EOL> except Exception as e : <EOL> LOG . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , e ) <EOL> self . allow_reauth = False <EOL> return self . store . get_store_connection ( <EOL> self . context . auth_token , self . storage_url ) </s>
