build classes using compososition
Different Domo Auth classes will have a variety of required vs optional parameters. To avoid multiple initialization and post_intialization statements, we mix multiple classes together such that classes with optional parameters are mixed in before classes with required parameters.
define the DomoAuth base class
source
DomoAuth
DomoAuth (domo_instance:str, url_manual_login:str, auth_header:dict,
token_name:str, token:str, user_id:str, is_valid_token:bool)
abstract DomoAuth class
Exported source
@dataclass
class DomoAuth:
"""abstract DomoAuth class"""
domo_instance: str
url_manual_login: str
auth_header: dict = field(repr = False )
token_name: str
token: str = field(repr = False )
user_id: str
is_valid_token: bool
def __post_init__(self ):
self .url_manual_login = self ._generate_manual_login(
domo_instance= self .domo_instance
)
@staticmethod
def _generate_manual_login(domo_instance):
return f"https:// { domo_instance} .domo.com/auth/index?domoManualLogin=true"
@classmethod
def from_domo_instance(cls, domo_instance, token_name= None ):
return cls(
domo_instance= domo_instance,
url_manual_login= cls._generate_manual_login(domo_instance),
auth_header= {},
token_name= token_name,
token= None ,
user_id= None ,
is_valid_token= False ,
)
async def who_am_i(
self ,
session: httpx.AsyncClient = None ,
debug_api: bool = False ,
debug_num_stacks_to_drop= 2 ,
):
if not self .auth_header:
self .generate_auth_header()
res = await auth_routes.who_am_i(
auth= None ,
auth_header= self .auth_header,
domo_instance= self .domo_instance,
parent_class= self .__class__.__name__ ,
session= session,
debug_api= debug_api,
debug_num_stacks_to_drop= debug_num_stacks_to_drop,
)
self .user_id = res.response["id" ]
return res
@abstractmethod
async def get_auth_token(self ) -> Union[str , None ]:
"""placeholder method"""
pass
@abstractmethod
def _generate_auth_header(self ) -> Union[dict , None ]:
"""returns auth header appropriate for this authentication method"""
pass
def generate_auth_header(self ) -> Union[dict , None ]:
return self ._generate_auth_header()
async def print_is_token(
self ,
debug_api: bool = False ,
token_name= None ,
session: httpx.AsyncClient = None ,
) -> bool :
self .token_name = token_name or self .token_name
if not self .token:
await self .get_auth_token(debug_api= debug_api, session = session)
token_str = f" { self . token_name} " or ""
if not self .token:
print (f"🚧 failed to retrieve { token_str} token from { self . domo_instance} " )
return False
print (f"🎉 { token_str } token retrieved from { self . domo_instance} ⚙️" )
return True
sample_implementation of set_manual_auth
# validate can print manual login link
domo_instance = "test"
DomoAuth.from_domo_instance(domo_instance)
DomoAuth(domo_instance='test', url_manual_login='https://test.domo.com/auth/index?domoManualLogin=true', token_name=None, user_id=None, is_valid_token=False)
DomoFullAuth
sample implementations of DomoFullAuth
try :
full_auth = DomoFullAuth(
domo_instance= "test" , domo_username= "test12@domo.com" , domo_password= "test1234"
)
await full_auth.get_auth_token()
except InvalidInstanceError as e:
print (e)
🛑 InvalidInstanceError 🛑 - function: DomoFullAuth.get_full_auth || status 403 || Forbidden at test
try :
full_auth = DomoFullAuth(
domo_instance= os.environ["DOMO_INSTANCE" ],
domo_username= os.environ["DOMO_USERNAME" ],
domo_password= os.environ["DOMO_PASSWORD" ],
)
await full_auth.get_auth_token()
except InvalidInstanceError as e:
print (e)
assert full_auth.token
await full_auth.who_am_i()
ResponseGetData(status=200, response={'id': 1893952720, 'invitorUserId': 587894148, 'displayName': 'Jae Wilson1', 'department': 'Business Improvement', 'userName': 'jae@onyxreporting.com', 'emailAddress': 'jae@datacrew.space', 'avatarKey': 'c605f478-0cd2-4451-9fd4-d82090b71e66', 'accepted': True, 'userType': 'USER', 'modified': 1729114351093, 'created': 1588960518, 'anonymous': False, 'systemUser': False, 'pending': False, 'active': True}, is_success=True, parent_class=None)
source
test_is_full_auth
test_is_full_auth (auth, function_name=None, num_stacks_to_drop=1)
auth
function_name
NoneType
None
num_stacks_to_drop
int
1
pass q for route pass 2 for class
DomoTokenAuth
source
DomoTokenAuth
DomoTokenAuth (domo_access_token:str, domo_instance:str,
token_name:str=None)
Sample implementation of DomoTokenAuth
try :
domo_auth = DomoTokenAuth(
domo_instance= os.environ["DOMO_INSTANCE" ], domo_access_token= "fake password"
)
# domo_auth.generate_auth_header()
await domo_auth.who_am_i()
except InvalidCredentialsError as e:
print (e)
🛑 InvalidCredentialsError 🛑 - function: DomoTokenAuth.who_am_i || status 401 || Unauthorized at domo-community
try :
domo_auth = DomoTokenAuth(
domo_instance= os.environ["DOMO_INSTANCE" ], domo_access_token= "fake password"
)
# await domo_auth.who_am_i()
await domo_auth.get_auth_token(debug_api= False )
except InvalidCredentialsError as e:
print (e)
🛑 InvalidCredentialsError 🛑 - function: DomoTokenAuth.who_am_i || status 401 || Unauthorized at domo-community
# # | eval : false
domo_auth = DomoTokenAuth(
domo_instance= os.environ["DOMO_INSTANCE" ],
domo_access_token= os.environ["DOMO_ACCESS_TOKEN" ],
)
await domo_auth.print_is_token()
domo_auth
🎉 token_auth token retrieved from domo-community ⚙️
DomoTokenAuth(domo_instance='domo-community', url_manual_login='https://domo-community.domo.com/auth/index?domoManualLogin=true', token_name='token_auth', user_id=1893952720, is_valid_token=True)
DomoDeveloperAuth
source
DomoDeveloperAuth
DomoDeveloperAuth (domo_client_id, domo_client_secret,
token_name:str=None, domo_instance:str=None)
Sample implementations of DomoDeveloperAuth
domo_client_id = "test_client"
domo_client_secret = "test_secret"
try :
domo_auth = DomoDeveloperAuth(
domo_client_id= domo_client_id, domo_client_secret= domo_client_secret
)
await domo_auth.get_auth_token()
except InvalidCredentialsError as e:
print (e)
🛑 InvalidCredentialsError 🛑 - function: get_auth_token || status 401 || Unauthorized
DomoJupyterAuth
source
DomoJupyterAuth
DomoJupyterAuth (jupyter_token:str, service_location:str,
service_prefix:str)
base class
define DomoJupyterFullAuth
source
DomoJupyterFullAuth
DomoJupyterFullAuth (jupyter_token:str, service_location:str,
service_prefix:str, domo_instance:str,
url_manual_login:str, auth_header:dict,
token_name:str, token:str, user_id:str,
is_valid_token:bool, domo_username:str,
domo_password:str)
sample implementation of convert_auth
domo_instance = os.environ["DOMO_INSTANCE" ]
full_auth = DomoFullAuth(
domo_instance= domo_instance,
domo_password= os.environ["DOMO_PASSWORD" ],
domo_username= os.environ["DOMO_USERNAME" ],
)
dj_auth = DomoJupyterFullAuth.convert_auth(
auth= full_auth,
jupyter_token= "abc" ,
service_location= "service_location_123" ,
service_prefix= "service_prefix_32a" ,
)
# await dj_auth.get_auth_token()
dj_auth.generate_auth_header()
{'x-domo-authentication': None, 'authorization': 'Token abc'}
define DomoJupyterTokenAuth
source
DomoJupyterTokenAuth
DomoJupyterTokenAuth (jupyter_token:str, service_location:str,
service_prefix:str, domo_instance:str,
url_manual_login:str, auth_header:dict,
token_name:str, token:str, user_id:str,
is_valid_token:bool, domo_access_token:str)
sample implementation of convert
token_auth = DomoTokenAuth(
domo_instance= os.environ["DOMO_INSTANCE" ],
domo_access_token= os.environ["DOMO_ACCESS_TOKEN" ],
)
dj_auth = DomoJupyterTokenAuth.convert_auth(
auth= token_auth,
jupyter_token= "abc" ,
service_location= "service_location_123" ,
service_prefix= "service_prefix_32a" ,
)
dj_auth.generate_auth_header()
{'x-domo-developer-token': '83ece44f1451d4b581e1191f98cd411164f0b5b6ad2755b3',
'authorization': 'Token abc'}
test is jupyter auth
source
test_is_jupyter_auth
test_is_jupyter_auth (auth:__main__.DomoJupyterAuth, function_name=None,
required_auth_type_ls=[<class
'__main__.DomoJupyterFullAuth'>, <class
'__main__.DomoJupyterTokenAuth'>])